DeepSeek API Python 示例：聊天、流式输出、工具具

安装并发送首次请求

该服务并非官方 DeepSeek API。它是一个托管的 OpenAI 兼容接口，运行单一开放权重模型，ID 为 uncensored，由独立提供商运营。由于遵循 OpenAI 格式，标准 Python SDK 无需修改即可使用，只需更改 URL 和密钥。

Copypip install openai

从 API 密钥页面 获取密钥。注册后，你只需点击一次即可领取 $0.50试用额度，无需信用卡，有效期 7 天。将密钥存储在环境变量中，而不是源代码文件中：

Copyexport API_KEY="YOUR_KEY"

Copyimport os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.deepseekapi.top/v1",
    api_key=os.environ["API_KEY"],
)

resp = client.chat.completions.create(
    model="uncensored",
    messages=[
        {"role": "system", "content": "You are a concise assistant."},
        {"role": "user", "content": "Explain list comprehensions in two sentences."},
    ],
    temperature=0.7,
)

print(resp.choices[0].message.content)
print(resp.usage.prompt_tokens, resp.usage.completion_tokens)

这两个使用量数字是计费依据。将它们乘以价格即可查看任何调用的成本：输入 token 数乘以 $0.15，输出 token 数乘以 $0.50，每个结果除以 1,000,000。

使用异步客户端

对于 Web 服务器和处理大量用户的场景，请使用异步客户端。它与同步客户端具有相同的参数，在大多数代码中是即插即用的更改。

Copyimport asyncio
import os
from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="https://api.deepseekapi.top/v1",
    api_key=os.environ["API_KEY"],
)

sem = asyncio.Semaphore(8)

async def ask(prompt):
    async with sem:
        r = await client.chat.completions.create(
            model="uncensored",
            messages=[{"role": "user", "content": prompt}],
        )
        return r.choices[0].message.content

async def main():
    prompts = ["Define idempotent.", "Define monotonic.", "Define pure function."]
    answers = await asyncio.gather(*(ask(p) for p in prompts))
    for a in answers:
        print(a)

asyncio.run(main())

信号量设置为 8，因为这是完整密钥的并发上限。试用密钥请使用 2。如果没有上限，任务突发将产生 concurrency 错误，即使每个请求都是有效的。

流式响应

设置 stream=True，接口将发送服务器发送事件（SSE）。每个数据块在 delta.content 中包含一小段文本。请求使用量意味着最后一个数据块将报告 token 计数。

Copy# client = OpenAI(base_url="https://api.deepseekapi.top/v1", ...) as above
stream = client.chat.completions.create(
    model="uncensored",
    messages=[{"role": "user", "content": "Write a haiku about debugging."}],
    stream=True,
    stream_options={"include_usage": True},
)

for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)
    if chunk.usage:
        print("\n", chunk.usage.prompt_tokens, chunk.usage.completion_tokens)

两个细节容易让人出错。最终的 usage chunk 可能包含空的 choices 列表，这就是循环首先检查 chunk.choices 的原因。流式请求会占用并发槽位直到流结束，因此请干净地跳出循环并关闭你放弃的流。完整密钥允许 8 个并发请求；试用密钥允许 2 个。

JSON 模式

对于程序需要解析的输出，请使用 response_format 配合 json_object。同时在提示词中提及 JSON 并描述你需要的结构。模型需要该指令；仅靠标志位无法提供结构定义。

Copy# client = OpenAI(base_url="https://api.deepseekapi.top/v1", ...) as above
import json

resp = client.chat.completions.create(
    model="uncensored",
    messages=[
        {"role": "system", "content": "Reply with JSON only: {\"title\": str, \"tags\": [str]}"},
        {"role": "user", "content": "Summarize: a guide to caching HTTP responses."},
    ],
    response_format={"type": "json_object"},
    temperature=0,
)

data = json.loads(resp.choices[0].message.content)
print(data["title"], data["tags"])

在 try/except 中包裹 json.loads。如果回复因 token 限制被截断，将无法解析，因此避免在 JSON 中使用过小的 max_tokens。将 temperature 设为 0 并设置 seed 可使重复运行更具可重复性，因为两者都会被原样透传。

函数调用

API 接受 tools 和 tool_choice，并在助手消息中返回 tool_calls。你自行运行函数，并将结果作为角色为 role: "tool" 的消息发送回去。

Copy# client = OpenAI(base_url="https://api.deepseekapi.top/v1", ...) as above
import json

tools = [{
    "type": "function",
    "function": {
        "name": "add",
        "description": "Add two numbers.",
        "parameters": {
            "type": "object",
            "properties": {"a": {"type": "number"}, "b": {"type": "number"}},
            "required": ["a", "b"],
        },
    },
}]

messages = [{"role": "user", "content": "What is 17.5 plus 24.25?"}]
first = client.chat.completions.create(
    model="uncensored", messages=messages, tools=tools, tool_choice="auto"
)

msg = first.choices[0].message
if msg.tool_calls:
    messages.append(msg)
    for call in msg.tool_calls:
        args = json.loads(call.function.arguments)
        result = args["a"] + args["b"]
        messages.append({
            "role": "tool",
            "tool_call_id": call.id,
            "content": str(result),
        })
    final = client.chat.completions.create(
        model="uncensored", messages=messages, tools=tools
    )
    print(final.choices[0].message.content)
else:
    print(msg.content)

始终防御性地解析 arguments。它是模型生成的 JSON 字符串，可能存在格式错误。在执行操作前验证它，特别是当函数涉及文件或数据库时。

错误、限制和重试

超出限制的请求返回 HTTP 429。错误代码会告知你哪种限制：rate_limited 表示每分钟请求数（完整密钥为 300，试用密钥为 60），concurrency 表示同时请求过多。在临时过载时，你也会收到带有 Retry-After: 5 的 429 错误。SDK 会抛出 openai.RateLimitError。

Copy# client = OpenAI(base_url="https://api.deepseekapi.top/v1", ...) as above
import time
import openai

def ask(prompt, tries=4):
    for attempt in range(tries):
        try:
            r = client.chat.completions.create(
                model="uncensored",
                messages=[{"role": "user", "content": prompt}],
            )
            return r.choices[0].message.content
        except openai.RateLimitError as e:
            wait = int(e.response.headers.get("Retry-After", 5))
            time.sleep(wait * (attempt + 1))
    raise RuntimeError("still rate limited")

print(ask("Name three Python web frameworks."))

如果你发送大量请求，请使用信号量或拥有 8 个工作线程的线程池限制你自己的并发数，或者在试用密钥上使用 2。这样可以避免 concurrency 错误，而不是通过重试来克服它们。

读取标头

每个响应都包含 X-Request-Id、X-Balance-USD、X-RateLimit-Limit-Requests 和 X-RateLimit-Limit-Concurrency。SDK 通过原始响应暴露这些标头：

Copy# client = OpenAI(base_url="https://api.deepseekapi.top/v1", ...) as above
raw = client.chat.completions.with_raw_response.create(
    model="uncensored",
    messages=[{"role": "user", "content": "ping"}],
)
print(raw.headers.get("X-Balance-USD"), raw.headers.get("X-Request-Id"))
completion = raw.parse()

在每次失败时记录请求 ID 会使支持问题更容易回答。失败或被拒绝的请求不会收费。

保持对话继续

API 是无状态的。它不记住之前的请求，因此多轮聊天意味着每次都要发送完整的历史记录。保持一个列表并将其追加：

Copy# client = OpenAI(base_url="https://api.deepseekapi.top/v1", ...) as above
history = [{"role": "system", "content": "You are a helpful tutor."}]

def chat(user_text):
    history.append({"role": "user", "content": user_text})
    r = client.chat.completions.create(model="uncensored", messages=history)
    reply = r.choices[0].message.content
    history.append({"role": "assistant", "content": reply})
    return reply

print(chat("What is a closure?"))
print(chat("Show me one in Python."))

每一轮对话都会重新发送所有内容，因此输入 token 会逐轮增长。一旦消息列表变长，请删除旧消息，或要求模型总结早期内容并替换它。拥有 100,000 token 的上下文窗口时，你有充足的空间，但账单会随着历史记录增长，因此精简列表更便宜。

选择采样值

对于代码和提取，使用 temperature 0 到 0.3；对于一般聊天，使用约 0.7；对于小说，使用更高的值。top_p、stop、seed 以及频率和存在惩罚都会传递，因此它们的行为与 OpenAI 文档描述的一致。一次更改一个并进行比较。

规划成本和上下文

上下文窗口为 100,000 token，包括提示词和补全，请求体限制为 8 MB。补全可以使用窗口中剩余的任何部分，max_tokens 是可选的。对于批处理任务，设置它仍然是良好的实践。

一个工作示例：1,000 个摘要的批处理，每个摘要包含 1,500 个输入 token 和 200 个输出 token。

• 输入：1,000 × 1,500 = 1,500,000 token，即 1.5 × $0.15 = $0.225
• 输出：1,000 × 200 = 200,000 token，即 0.2 × $0.50 = $0.10
• 总计：$0.325

定价为按量付费，从预付额度扣除，无订阅，且额度永不失效。充值金额为 $20 至 $500 的整数，$50 以上加 5%，$100 以上加 10%，$200 以上加 15%，支持信用卡或加密货币支付。详见 定价页面，以及 快速入门文档 中的 curl 和 Node 示例。如果你只需要查找密钥位置，API 密钥概览 对此进行了说明。

提示词不用于训练。该服务仅支持文本：无图像、音频、嵌入或微调，因此不要围绕这些接口进行规划。

更多指南

• 在 Cursor 中使用 DeepSeek API：自定义 Base URL 设置
• 如何在 SillyTavern 中使用 DeepSeek API：自定义接口
• DeepSeek API 密钥：无审查 LLM API 访问