DeepSeek 应用开发实战：从 API 调用到智能问答系统构建

pip install openai requests langchain chromadb
export DEEPSEEK_API_KEY="sk-xxx"

from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1"  # DeepSeek 官方端点
)

response = client.chat.completions.create(
    model="deepseek-chat",  # 或 deepseek-reasoner（推理模型）
    messages=[{"role": "user", "content": "请用 Python 实现快速排序"}],
    temperature=0.3,
    max_tokens=2048
)
print(response.choices[0].message.content)

class DeepSeekSession:
    def __init__(self, system_prompt=None):
        self.messages = []
        if system_prompt:
            self.messages.append({"role": "system", "content": system_prompt})
    
    def add_user(self, content):
        self.messages.append({"role": "user", "content": content})
    
    def add_assistant(self, content):
        self.messages.append({"role": "assistant", "content": content})
    
    def stream_chat(self, **kwargs):
        """流式生成，返回生成器"""
        stream = client.chat.completions.create(
            model="deepseek-chat",
            messages=self.messages,
            stream=True,
            **kwargs
        )
        for chunk in stream:
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content

# 使用示例
session = DeepSeekSession("你是一名 Python 助手，回答需附带代码示例。")
session.add_user("如何读取 CSV 文件并统计缺失值？")
for token in session.stream_chat():
    print(token, end="", flush=True)

tools = [{
    "type": "function",
    "function": {
        "name": "get_current_weather",
        "description": "获取指定城市的当前天气",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名称，如 Beijing"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
            },
            "required": ["city"]
        }
    }
}]

def get_current_weather(city, unit="celsius"):
    # 模拟真实 API 调用，实际可接入和风天气等
    return f"{city} 当前气温 22°C，晴间多云"

def chat_with_tools(user_query):
    messages = [{"role": "user", "content": user_query}]
    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=messages,
        tools=tools,
        tool_choice="auto"  # 让模型自主决定
    )
    msg = response.choices[0].message
    if msg.tool_calls:
        # 执行工具调用
        for tc in msg.tool_calls:
            if tc.function.name == "get_current_weather":
                args = json.loads(tc.function.arguments)
                result = get_current_weather(args["city"], args.get("unit", "celsius"))
                # 将工具结果返回给模型
                messages.append(msg)  # 助手消息（含工具调用）
                messages.append({
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": result
                })
        # 获取最终回答
        final = client.chat.completions.create(
            model="deepseek-chat",
            messages=messages
        )
        return final.choices[0].message.content
    else:
        return msg.content

print(chat_with_tools("北京今天天气怎么样？"))
# 输出：北京当前气温 22°C，晴间多云

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma

# 1. 加载并切分文档
loader = TextLoader("product_manual.txt")
docs = loader.load()
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = splitter.split_documents(docs)

# 2. 生成向量库（使用开源嵌入模型）
embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-small-zh")
vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory="./chroma_db")

# 3. 检索 + 生成
def rag_query(query):
    retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
    docs = retriever.invoke(query)
    context = "\n\n".join([doc.page_content for doc in docs])
    prompt = f"""基于以下资料回答用户问题，若无法从资料中找到答案则说明。
    资料：{context}
    问题：{query}
    回答："""
    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1
    )
    return response.choices[0].message.content

print(rag_query("产品的保修政策是什么？"))