DeepSeek V3 的 API 性价比在 2026 年依然没有对手——同等能力价格只有 GPT-5.5 的 1/5。但翻了一圈中文技术社区，发现大多数「教程」只讲到第一段 chat.completions.create 就停了。

生产环境真正需要的东西——流式输出怎么接、Function Calling 踩了什么坑、高并发下怎么做重试、Token 成本怎么算——几乎没人讲。

这篇文章用一套可运行的 Python 代码把上面这些全串起来。

---

1. 环境准备

# 创建虚拟环境
python -m venv deepseek-env
source deepseek-env/bin/activate  # Windows: deepseek-env\Scripts\activate

# 安装依赖
pip install openai python-dotenv httpx

# config.py — 配置管理
import os
from dotenv import load_dotenv

load_dotenv()

DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_BASE_URL = "https://api.deepseek.com/v1"
DEFAULT_MODEL = "deepseek-chat"

if not DEEPSEEK_API_KEY:
    raise RuntimeError("请在 .env 中设置 DEEPSEEK_API_KEY")

.env 文件内容：

DEEPSEEK_API_KEY=sk-your-key-here

---

2. 基础调用 — 不只是 hello world

# basic_chat.py
from openai import OpenAI
from config import DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL, DEFAULT_MODEL

client = OpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL,
)

response = client.chat.completions.create(
    model=DEFAULT_MODEL,
    messages=[
        {"role": "system", "content": "你是一个精通 Python 的后端工程师。"},
        {"role": "user", "content": "用 asyncio 写一个并发抓取 10 个 URL 的爬虫框架。"},
    ],
    temperature=0.7,
    max_tokens=4096,
)

print(response.choices[0].message.content)
print(f"\nToken 消耗: prompt={response.usage.prompt_tokens}, "
      f"completion={response.usage.completion_tokens}, "
      f"total={response.usage.total_tokens}")

跑完这一段你会发现：DeepSeek 生成代码的质量在 Python 和 Shell 场景下跟 GPT-5.5 几乎没差别，但 Token 成本只有后者的 1/5。

---

3. 流式输出 — 对接前端 SSE

生产环境不可能等 30 秒让用户看白屏。流式输出是刚需。

# streaming.py
from openai import OpenAI
from config import DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL, DEFAULT_MODEL

client = OpenAI(api_key=DEEPSEEK_API_KEY, base_url=DEEPSEEK_BASE_URL)

def stream_chat(prompt: str):
    """流式输出，逐 token 返回，适配 SSE 推送。"""
    stream = client.chat.completions.create(
        model=DEFAULT_MODEL,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7,
        max_tokens=4096,
        stream=True,
    )

    full_response = ""
    for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            full_response += delta.content
            print(delta.content, end="", flush=True)
        # 获取 usage（流式最后一个 chunk 包含 token 统计）
        if hasattr(chunk, "usage") and chunk.usage:
            print(f"\n\n--- Token: {chunk.usage.total_tokens} ---")

    return full_response

if __name__ == "__main__":
    stream_chat("解释一下 Transformer 的 Multi-Head Attention 机制")

踩坑提醒：流式模式下 usage 只在最后一个 chunk 返回。如果你在前端展示「已消耗 Token」，需要等到流结束再更新。

---

4. Function Calling — 让模型调用你的工具

这是 Agent 开发中最关键的能力。DeepSeek 兼容 OpenAI 的 tool calling 协议。

# function_calling.py
import json
from openai import OpenAI
from config import DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL, DEFAULT_MODEL

client = OpenAI(api_key=DEEPSEEK_API_KEY, base_url=DEEPSEEK_BASE_URL)

# 定义工具 — 跟 OpenAI function calling 格式完全兼容
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_knowledge_base",
            "description": "搜索内部知识库，返回匹配的文档片段",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词",
                    },
                    "top_k": {
                        "type": "integer",
                        "description": "返回结果数量，默认 5",
                        "default": 5,
                    },
                },
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "send_email",
            "description": "发送邮件",
            "parameters": {
                "type": "object",
                "properties": {
                    "to": {"type": "string", "description": "收件人邮箱"},
                    "subject": {"type": "string", "description": "邮件主题"},
                    "body": {"type": "string", "description": "邮件正文"},
                },
                "required": ["to", "subject", "body"],
            },
        },
    },
]


def execute_tool(name: str, args: dict) -> str:
    """模拟工具执行。生产环境替换为真实实现。"""
    if name == "search_knowledge_base":
        return json.dumps({
            "results": [
                {"score": 0.92, "content": f"关于 '{args['query']}' 的文档片段..."},
                {"score": 0.87, "content": f"'{args['query']}' 相关的 API 文档..."},
            ]
        }, ensure_ascii=False)
    elif name == "send_email":
        return json.dumps({"status": "sent", "to": args["to"]})
    return json.dumps({"error": "unknown tool"})


def agent_loop(user_input: str):
    """简化的 Agent 循环：模型决定调用工具 → 执行 → 模型用结果回复。"""
    messages = [{"role": "user", "content": user_input}]

    # 第一轮：模型决定是否调用工具
    response = client.chat.completions.create(
        model=DEFAULT_MODEL,
        messages=messages,
        tools=tools,
        tool_choice="auto",
    )

    msg = response.choices[0].message

    # 如果模型要求调用工具
    if msg.tool_calls:
        for tool_call in msg.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)

            print(f"[Agent] 调用工具: {func_name}({func_args})")

            result = execute_tool(func_name, func_args)

            # 把工具调用和结果追加到对话历史
            messages.append({
                "role": "assistant",
                "content": None,
                "tool_calls": [
                    {
                        "id": tool_call.id,
                        "type": "function",
                        "function": {
                            "name": func_name,
                            "arguments": tool_call.function.arguments,
                        },
                    }
                ],
            })
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            })

        # 第二轮：模型基于工具结果生成最终回复
        final_response = client.chat.completions.create(
            model=DEFAULT_MODEL,
            messages=messages,
        )
        print(f"\n[Agent] 最终回复:\n{final_response.choices[0].message.content}")
    else:
        print(f"[Agent] 直接回复:\n{msg.content}")


if __name__ == "__main__":
    agent_loop("帮我查一下知识库里关于 API 认证的文档，然后发一份摘要到 admin@company.com")

---

5. 多轮对话 + 上下文管理

# conversation.py
from openai import OpenAI
from config import DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL, DEFAULT_MODEL

client = OpenAI(api_key=DEEPSEEK_API_KEY, base_url=DEEPSEEK_BASE_URL)


class Conversation:
    """支持多轮对话，自动管理上下文窗口，超限时自动截断。"""

    def __init__(self, system_prompt: str = "", max_history: int = 20):
        self.max_history = max_history
        self.messages = []
        if system_prompt:
            self.messages.append({"role": "system", "content": system_prompt})

    def ask(self, user_input: str) -> str:
        self.messages.append({"role": "user", "content": user_input})

        response = client.chat.completions.create(
            model=DEFAULT_MODEL,
            messages=self.messages,
            temperature=0.7,
            max_tokens=2048,
        )

        reply = response.choices[0].message.content
        self.messages.append({"role": "assistant", "content": reply})

        # 保持消息数在限制内（保留 system prompt + 最近 N 轮）
        if len(self.messages) > self.max_history + 1:
            system_msgs = [m for m in self.messages if m["role"] == "system"]
            other_msgs = [m for m in self.messages if m["role"] != "system"]
            self.messages = system_msgs + other_msgs[-(self.max_history):]

        return reply


if __name__ == "__main__":
    conv = Conversation(system_prompt="你是资深 Python 技术顾问。")

    print("Q: 装饰器在 Python 里有什么高级用法？")
    print(f"A: {conv.ask('装饰器在 Python 里有什么高级用法？')}\n")

    print("Q: 能给我一个带参数的装饰器示例吗？")
    print(f"A: {conv.ask('能给我一个带参数的装饰器示例吗？')}\n")

    print("Q: 这个装饰器怎么处理异步函数？")
    print(f"A: {conv.ask('这个装饰器怎么处理异步函数？')}")

小技巧：DeepSeek 的上下文窗口是 128K，但如果每次请求都带满 128K 的上下文，成本会非常高。Conversation 类的 max_history 限制就是为了控制这个——只保留最近 N 轮，超出的自动截掉。

---

6. 异常处理 + 指数退避重试

生产环境没有完美网络。API 会超时、会限流、会 500。不加重试的调用在生产环境撑不过一天。

# retry_handler.py
import time
import random
from openai import OpenAI, APIError, APITimeoutError, RateLimitError
from config import DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL, DEFAULT_MODEL

client = OpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL,
    timeout=60.0,
    max_retries=0,  # 禁用 SDK 内置重试，用我们自己的策略
)


def chat_with_retry(messages: list, max_retries: int = 3, base_delay: float = 1.0):
    """指数退避 + 抖动重试。"""
    last_error = None

    for attempt in range(max_retries + 1):
        try:
            return client.chat.completions.create(
                model=DEFAULT_MODEL,
                messages=messages,
                temperature=0.7,
                max_tokens=4096,
            )
        except RateLimitError as e:
            last_error = e
            if attempt < max_retries:
                delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                print(f"[RateLimit] 重试 {attempt+1}/{max_retries}, 等待 {delay:.1f}s")
                time.sleep(delay)
        except (APITimeoutError, APIError) as e:
            last_error = e
            if attempt < max_retries:
                delay = base_delay * (2 ** attempt)
                print(f"[APIError] 重试 {attempt+1}/{max_retries}, 等待 {delay:.1f}s")
                time.sleep(delay)

    raise last_error


if __name__ == "__main__":
    try:
        response = chat_with_retry([
            {"role": "user", "content": "写一个 Python 单例模式的实现"},
        ])
        print(response.choices[0].message.content)
    except Exception as e:
        print(f"所有重试均失败: {e}")

几个关键参数： - base_delay=1.0：首次重试等 1 秒，之后 2 秒、4 秒 - random.uniform(0, 1)：加抖动，避免 thundering herd - max_retries=3：总共 4 次尝试（含首次），根据业务 SLA 调整

---

7. Token 成本计算

# cost_calculator.py
from decimal import Decimal, ROUND_HALF_UP

# DeepSeek 官方定价（2026年5月）
PRICING = {
    "deepseek-chat": {
        "prompt": Decimal("0.27"),       # ¥/百万 token
        "completion": Decimal("1.10"),   # ¥/百万 token
    },
    "deepseek-reasoner": {
        "prompt": Decimal("0.55"),
        "completion": Decimal("2.19"),
    },
}


def calculate_cost(model: str, prompt_tokens: int, completion_tokens: int) -> dict:
    """计算单次调用的成本（人民币 元）。"""
    price = PRICING.get(model)
    if not price:
        raise ValueError(f"未知模型: {model}")

    prompt_cost = (Decimal(prompt_tokens) / 1_000_000 * price["prompt"])
    completion_cost = (Decimal(completion_tokens) / 1_000_000 * price["completion"])
    total = prompt_cost + completion_cost

    return {
        "model": model,
        "prompt_tokens": prompt_tokens,
        "completion_tokens": completion_tokens,
        "prompt_cost_yuan": float(prompt_cost.quantize(Decimal("0.0001"), ROUND_HALF_UP)),
        "completion_cost_yuan": float(completion_cost.quantize(Decimal("0.0001"), ROUND_HALF_UP)),
        "total_cost_yuan": float(total.quantize(Decimal("0.0001"), ROUND_HALF_UP)),
    }


def estimate_monthly_cost(
    model: str,
    requests_per_day: int,
    avg_prompt_tokens: int,
    avg_completion_tokens: int,
) -> dict:
    """估算月度成本。"""
    daily = calculate_cost(
        model,
        requests_per_day * avg_prompt_tokens,
        requests_per_day * avg_completion_tokens,
    )
    return {
        **daily,
        "requests_per_day": requests_per_day,
        "monthly_cost_yuan": round(daily["total_cost_yuan"] * 30, 2),
    }


if __name__ == "__main__":
    # 示例：每天 1000 次请求，每次 prompt 500 token，completion 1000 token
    result = estimate_monthly_cost(
        model="deepseek-chat",
        requests_per_day=1000,
        avg_prompt_tokens=500,
        avg_completion_tokens=1000,
    )
    print(f"模型: {result['model']}")
    print(f"每日请求: {result['requests_per_day']}")
    print(f"单次成本: ¥{result['total_cost_yuan']}")
    print(f"月度预估: ¥{result['monthly_cost_yuan']}")

    # 对比：同等用量的 GPT-5.5 成本约 ¥35/月，DeepSeek 约 ¥7/月
    print("\n对比 GPT-5.5（同等用量约 5 倍价格）:")
    print(f"  DeepSeek: ¥{result['monthly_cost_yuan']}/月")
    print(f"  GPT-5.5:  ¥{result['monthly_cost_yuan'] * 5}/月（估算）")

---

8. 生产部署：FastAPI 封装

把上面的代码组装成一个可部署的 API 服务：

# api_server.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from openai import OpenAI, APIError, RateLimitError
from config import DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL, DEFAULT_MODEL

import time, random

app = FastAPI(title="DeepSeek Proxy API")
client = OpenAI(api_key=DEEPSEEK_API_KEY, base_url=DEEPSEEK_BASE_URL, max_retries=0)


class ChatRequest(BaseModel):
    prompt: str = Field(..., min_length=1, max_length=50000)
    temperature: float = Field(default=0.7, ge=0, le=2.0)
    max_tokens: int = Field(default=4096, ge=1, le=8192)
    stream: bool = Field(default=False)


@app.post("/v1/chat")
async def chat(req: ChatRequest):
    last_error = None
    for attempt in range(3):
        try:
            response = client.chat.completions.create(
                model=DEFAULT_MODEL,
                messages=[{"role": "user", "content": req.prompt}],
                temperature=req.temperature,
                max_tokens=req.max_tokens,
                stream=req.stream,
            )
            if req.stream:
                # 生产环境用 StreamingResponse 逐个 yield
                return {"content": "streaming — use SSE endpoint"}
            return {
                "content": response.choices[0].message.content,
                "usage": {
                    "prompt_tokens": response.usage.prompt_tokens,
                    "completion_tokens": response.usage.completion_tokens,
                    "total_tokens": response.usage.total_tokens,
                },
            }
        except RateLimitError:
            time.sleep(2 ** attempt + random.uniform(0, 1))
        except APIError as e:
            last_error = e
            time.sleep(2 ** attempt)
    raise HTTPException(status_code=502, detail=str(last_error))


@app.get("/health")
async def health():
    return {"status": "ok"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

启动：

pip install fastapi uvicorn
python api_server.py
# → http://localhost:8000/docs 查看 Swagger 文档

---

结尾

DeepSeek API 在生产上的体验是：能力够用，价格离谱低，协议跟 OpenAI 完全兼容。

这意味着你不需要改一行代码就能从 GPT 切到 DeepSeek——换了 base_url 就行。省下来的 80% Token 成本，要么降低产品定价，要么提高利润率。

上面这套代码覆盖了从开发到部署的全链路：基础调用 → 流式 → Function Calling → 多轮对话 → 重试 → 成本计算 → FastAPI 封装。直接复制粘贴就能跑。