如何快速實(shí)現(xiàn)REST API集成以優(yōu)化業(yè)務(wù)流程
從成本考量，DeepSeek 幾乎是最佳方案，因 DeepSeek API 調(diào)用價(jià)格之便宜曾被戲稱為 “AI 界的拼多多”。在 DeepSeek 價(jià)格公開后不久，多家模型廠商卷入價(jià)格戰(zhàn)，現(xiàn)在的模型調(diào)用價(jià)格是真真正正的被 “打下來”了。多家公司頻繁更新自家模型價(jià)格，截止目前，可以說 “沒有最低，只有更低”。
從效果考量 ，因之前使用過 deepseek-coder、和 deepseek-chat 兩個(gè)模型，效果上可以說是在中文模型領(lǐng)域的第一梯隊(duì)。當(dāng)然這只是我個(gè)人的使用體驗(yàn)。
從權(quán)威的角度，通過?LMSYS Chatbot Arena Leaderboard（LMSYS Chatbot Arena Leaderboard 是一個(gè)大型語言模型的評(píng)測(cè)排行榜，提供了一個(gè)匿名競(jìng)技場(chǎng)，用于評(píng)估和比較不同模型的性能。） 這個(gè)大型語言模型的評(píng)測(cè)排行榜可以了解 DeepSeek 的能力如何
最近的幾個(gè)月里，國(guó)產(chǎn)模型中與 DeepSeek 排名競(jìng)爭(zhēng)最激烈的是阿里的?Qwen2.5
前文中我們提到 DeepSeek 的 API 是需要付費(fèi)調(diào)用的，所以到底收多少錢是一個(gè)關(guān)鍵的問題。
首先，如果你是一個(gè)新用戶，那么 DeeepSeek 會(huì)送你 500w 個(gè) tokens （在自然語言處理中，Token 是指將文本分割成的最小單位。這些單位可以是單詞、子詞、字符等，具體取決于所使用的分詞策略）。簡(jiǎn)單理解就是 500w 個(gè)字。需要注意的是，送的 tokens 有有效期，一個(gè)月后就過期了。
其次，如果送的 tokens 用完了，就需要花真金白銀去充值了。
簡(jiǎn)單說， 10 元 500w tokens，如果你是個(gè)人使用，一個(gè)人放開了用，一個(gè)月足夠了。
無論是通過贈(zèng)送還是付費(fèi)，當(dāng)你擁有了 tokens，你就可以根據(jù)文檔創(chuàng)建自己的 API key 并進(jìn)行 API 調(diào)用了。
由于是走網(wǎng)絡(luò) API 的這種方式，在編程語言上就沒有限制了，你可以選用你覺得合適的語言。DeepSeek 官方也比較貼心的給出了各種語言調(diào)用的示例：
這里我用?Python?寫了一個(gè)簡(jiǎn)單的調(diào)用 Demo， 以下是具體代碼：
from openai import OpenAI
class DeepSeekChat:
def __init__(self, api_key, base_url="https://api.deepseek.com"):
self.client = OpenAI(api_key=api_key, base_url=base_url)
def chat(
self,
system_message,
user_message,
model="deepseek-chat",
max_tokens=1024,
temperature=0.7,
stream=True,
):
response = self.client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": system_message},
{"role": "user", "content": user_message},
],
max_tokens=max_tokens,
temperature=temperature,
stream=stream,
)
if stream:
return self._stream_response(response)
else:
return response.choices[0].message.content
def _stream_response(self, response):
full_response = ""
for chunk in response:
if chunk.choices[0].delta.content is not None:
content = chunk.choices[0].delta.content
print(content, end="", flush=True)
full_response += content
print("\r\n===============我是分隔線===============")
return full_response
# 使用示例
if __name__ == "__main__":
deepseek_chat = DeepSeekChat(api_key="[你的 API Key]")
response = deepseek_chat.chat(
system_message="你是一個(gè)聰明的 AI 助手",
user_message="三國(guó)演義中戰(zhàn)斗力排名前 10 的武將有誰？",
stream=True,
)
print("完整回答：", response)可以看到我們只引入了 openai 這一個(gè)庫，原因是 DeepSeek 的 API 和 OpenAI 的 API 是兼容的。
“
DeepSeek API 使用與 OpenAI 兼容的 API 格式，通過修改配置，您可以使用 OpenAI SDK 來訪問 DeepSeek API，或使用與 OpenAI API 兼容的軟件。
--源自 DeepSeek 文檔
引入 openai 這個(gè)庫以后我們不需要再引入其他多余的庫就可以進(jìn)行 API 請(qǐng)求了。
這段代碼比較簡(jiǎn)單，我的問題是 ：“三國(guó)演義中戰(zhàn)斗力排名前 10 的武將有誰？” 我們來看一下大模型給我的回答：
在上一篇文章中我們能夠方便地調(diào)用 Ollama 進(jìn)而調(diào)用本地下載好的模型，是因?yàn)?LlamaIndex 的庫封裝好了：
# 設(shè)置語言模型，使用 Ollama 提供的 Qwen2 7B 模型，并設(shè)置請(qǐng)求超時(shí)時(shí)間
Settings.llm = Ollama(model="qwen2:7b", request_timeout=360.0)現(xiàn)在，我們想用在線的模型 DeepSeek，讓 LlamaIndex 去調(diào)用 DeepSeek API 就不能用之前的方式了。
通過查看 LlamaIndex 的文檔，總結(jié)來說，它支持的 LLM 集成方式有三種：
我們需要解釋一下：
Replicate
如何通過 Custom LLM 的方式將 DeepSeek 與 LlamaIndex 進(jìn)行集成呢？
其實(shí)很容易，我們只需要?jiǎng)?chuàng)建一個(gè)類并實(shí)現(xiàn)三個(gè)方法即可（用 python 代碼實(shí)現(xiàn)）。
文檔中給出的代碼是這樣的：
from typing import Optional, List, Mapping, Any
from llama_index.core import SimpleDirectoryReader, SummaryIndex
from llama_index.core.callbacks import CallbackManager
from llama_index.core.llms import (
CustomLLM,
CompletionResponse,
CompletionResponseGen,
LLMMetadata,
)
from llama_index.core.llms.callbacks import llm_completion_callback
from llama_index.core import Settings
class OurLLM(CustomLLM):
context_window: int = 3900
num_output: int = 256
model_name: str = "custom"
dummy_response: str = "My response"
@property
def metadata(self) -> LLMMetadata:
"""Get LLM metadata."""
return LLMMetadata(
context_window=self.context_window,
num_output=self.num_output,
model_name=self.model_name,
)
@llm_completion_callback()
def complete(self, prompt: str, **kwargs: Any) -> CompletionResponse:
return CompletionResponse(text=self.dummy_response)
@llm_completion_callback()
def stream_complete(
self, prompt: str, **kwargs: Any
) -> CompletionResponseGen:
response = ""
for token in self.dummy_response:
response += token
yield CompletionResponse(text=response, delta=token)
# define our LLM
Settings.llm = OurLLM()
# define embed model
Settings.embed_model = "local:BAAI/bge-base-en-v1.5"
# Load the your data
documents = SimpleDirectoryReader("./data").load_data()
index = SummaryIndex.from_documents(documents)
# Query and print response
query_engine = index.as_query_engine()
response = query_engine.query("<query_text>")
print(response)OurLLM 就是要?jiǎng)?chuàng)建的類，要實(shí)現(xiàn)的三個(gè)方法是：
實(shí)際上一般 metadata 方法可以直接返回 LLMMetadata() ，最主要的就是實(shí)現(xiàn)后面兩個(gè)方法。
根據(jù)上一節(jié) Custom LLM 所述，我將上一篇文章中的 Ollama 模型調(diào)用換成自定義的 DeepSeek，以下是主要代碼：
import os
import sys
import logging
from openai import OpenAI
from typing import Any, Generator
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.core.llms import (
CustomLLM,
CompletionResponse,
CompletionResponseGen,
LLMMetadata,
)
from llama_index.core.llms.callbacks import llm_completion_callback
from pydantic import BaseModel, Field
from dotenv import load_dotenv
from functools import cached_property
# 配置日志 創(chuàng)建一個(gè)與當(dāng)前模塊同名的 logger
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 從環(huán)境變量獲取 API 密鑰
load_dotenv()
API_KEY = os.getenv("DEEPSEEK_API_KEY")
if not API_KEY:
raise ValueError("DEEPSEEK_API_KEY environment variable is not set")
class DeepSeekChat(BaseModel):
"""DeepSeek 聊天模型的封裝類。"""
api_key: str = Field(default=API_KEY)
base_url: str = Field(default="https://api.deepseek.com")
class Config:
"""Pydantic 配置類。"""
arbitrary_types_allowed = True # 允許模型接受任意類型的字段
# 這增加了靈活性，但可能降低類型安全性
# 在本類中，這可能用于允許使用 OpenAI 客戶端等復(fù)雜類型
@cached_property
def client(self) -> OpenAI:
"""創(chuàng)建并緩存 OpenAI 客戶端實(shí)例。"""
return OpenAI(api_key=self.api_key, base_url=self.base_url)
def chat(
self,
system_message: str,
user_message: str,
model: str = "deepseek-chat",
max_tokens: int = 1024,
temperature: float = 0.7,
stream: bool = False,
) -> Any:
"""
使用 DeepSeek API 發(fā)送聊天請(qǐng)求。
返回流式響應(yīng)或完整響應(yīng)內(nèi)容。
"""
try:
response = self.client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": system_message},
{"role": "user", "content": user_message},
],
max_tokens=max_tokens,
temperature=temperature,
stream=stream,
)
return response if stream else response.choices[0].message.content
except Exception as e:
logger.error(f"Error in DeepSeek API call: {e}")
raise
def _stream_response(self, response) -> Generator[str, None, None]:
"""處理流式響應(yīng)，逐塊生成內(nèi)容。"""
for chunk in response:
if chunk.choices[0].delta.content is not None:
yield chunk.choices[0].delta.content
class DeepSeekLLM(CustomLLM):
"""DeepSeek 語言模型的自定義實(shí)現(xiàn)。"""
deep_seek_chat: DeepSeekChat = Field(default_factory=DeepSeekChat)
@property
def metadata(self) -> LLMMetadata:
"""返回 LLM 元數(shù)據(jù)。"""
return LLMMetadata()
@llm_completion_callback()
def complete(self, prompt: str, **kwargs: Any) -> CompletionResponse:
"""執(zhí)行非流式完成請(qǐng)求。"""
response = self.deep_seek_chat.chat(
system_message="你是一個(gè)聰明的 AI 助手", user_message=prompt, stream=False
)
return CompletionResponse(text=response)
@llm_completion_callback()
def stream_complete(self, prompt: str, **kwargs: Any) -> CompletionResponseGen:
"""執(zhí)行流式完成請(qǐng)求。"""
response = self.deep_seek_chat.chat(
system_message="你是一個(gè)聰明的 AI 助手", user_message=prompt, stream=True
)
def response_generator():
"""生成器函數(shù)，用于逐步生成響應(yīng)內(nèi)容。"""
response_content = ""
for chunk in self.deep_seek_chat._stream_response(response):
if chunk:
response_content += chunk
yield CompletionResponse(text=response_content, delta=chunk)
return response_generator()
# 設(shè)置環(huán)境變量，禁用 tokenizers 的并行處理，以避免潛在的死鎖問題
os.environ["TOKENIZERS_PARALLELISM"] = "false"
def main():
"""主程序函數(shù)，演示如何使用 DeepSeekLLM 進(jìn)行文檔查詢。"""
# 從指定目錄加載文檔數(shù)據(jù)
documents = SimpleDirectoryReader("data").load_data()
# 設(shè)置 LLM 和嵌入模型
Settings.llm = DeepSeekLLM()
Settings.embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-base-zh-v1.5")
# 創(chuàng)建索引和查詢引擎
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine(streaming=True)
# 執(zhí)行查詢
print("查詢結(jié)果：")
response = query_engine.query("作者學(xué)習(xí)過的編程語言有哪些？")
# 處理并輸出響應(yīng)
if hasattr(response, "response_gen"):
# 流式輸出
for text in response.response_gen:
print(text, end="", flush=True)
sys.stdout.flush() # 確保立即輸出
else:
# 非流式輸出
print(response.response, end="", flush=True)
print("\n 查詢完成")
if __name__ == "__main__":
main()你別看代碼寫的長(zhǎng)，那是因?yàn)槲易鲞^重構(gòu)，其實(shí)可以實(shí)現(xiàn)的更短。不要被篇幅嚇到，其實(shí)主要執(zhí)行邏輯與上一篇文章中寫的沒什么區(qū)別，只在自定義 DeepSeekLLM 這里有所不同，如果你把本文從頭看到尾，其實(shí)其中的第一步分解拆開都有解釋過，也比較簡(jiǎn)單。
我們來看一下效果，測(cè)試數(shù)據(jù)仍然是上一篇文章中的文本內(nèi)容，問題仍然是 ：“作者學(xué)習(xí)過的編程語言有哪些？”
本文我們介紹了如何通過調(diào)用國(guó)產(chǎn)大模型?DeepSeek?的 API 來提升?RAG（檢索增強(qiáng)生成）應(yīng)用的執(zhí)行效率。相比使用本地 Ollama 模型，DeepSeek 的 API 不僅解決了本地計(jì)算資源不足導(dǎo)致的運(yùn)行速度慢的問題，還保持了高質(zhì)量的生成結(jié)果。DeepSeek 在成本和效果上表現(xiàn)出色，特別適合中文模型的應(yīng)用。通過自定義 LLM 的方式，我們成功將 DeepSeek 與?LlamaIndex?集成，展示了如何實(shí)現(xiàn)高效的數(shù)據(jù)處理和生成。本文提供的方法和示例代碼為構(gòu)建高性能 RAG 應(yīng)用提供了一種實(shí)用的解決方案。
文章轉(zhuǎn)自微信公眾號(hào)@小盒子的技術(shù)分享
對(duì)比大模型API的內(nèi)容創(chuàng)意新穎性、情感共鳴力、商業(yè)轉(zhuǎn)化潛力
一鍵對(duì)比試用API 限時(shí)免費(fèi)
鍵.png)
