DeepSeek V4 API Python: 流式响应的最少代码示例

I’ll translate this article to Simplified Chinese. Let me create the translated version:

你好各位！我是Dora。一切始于一个小烦恼：我不断地在项目之间复制相同的聊天完成样板代码，像标签一样交换基础URL和模型名称。这不是难的工作，只是那种会给你的一天增添摩擦的工作。我已经看到DeepSeek出现的次数足够多，引起了我的好奇心，所以我在2026年1月下旬花了几个早晨的时间，将他们的”V4”API集成到我的Python堆栈中，看看它在实际使用中的感受如何。

我没有追逐基准。我想知道：客户端是否不会妨碍我，我能否可靠地流传输，错误是否以易于推理的方式失败？这是我尝试的、绊住我的地方，以及无声工作的地方。我们开始吧！

环境设置

依赖项

我在macOS上使用Python 3.11保持设置简单。你可以用标准库做到这一点，但三个小包使生活更轻松：

requests（直接的HTTP：足够用于大多数情况）
httpx（异步和行为良好的超时）
python-dotenv（所以我不会到处粘贴密钥）如果你计划使用服务器发送事件进行流传输，你可以使用requests并自己解析行（我所做的），或者引入一个像sseclient-py这样的帮助程序。我坚持使用requests，更少的活动部件。

安装

pip install requests httpx python-dotenv

我还为每个项目创建了一个最小的虚拟环境。这是乏味的建议，但当你三个月后重新访问时，它可以让你免受依赖漂移的困扰。

API密钥配置

我将密钥存储在环境变量中。没什么花哨的：

# .env
DEEPSEEK_API_KEY=your_key_here

然后在Python中：

from dotenv import load_dotenv
import os

load_dotenv()

API_KEY = os.getenv("DEEPSEEK_API_KEY")

if not API_KEY:
    raise RuntimeError("Missing DEEPSEEK_API_KEY")

设置中的两个小注意事项：

基础URL和模型名称比你想象的更频繁地变化。我在每次运行前检查了官方DeepSeek API文档以确认路径和可用模型。
我保持超时明确。这是一个习惯，一旦你触及速率限制或网络噪音就会得到回报。

基本聊天请求

如果你在其他地方使用过聊天完成，心智模型会很熟悉。DeepSeek公开了一个带有messages=[{"role": "...", "content": "..."}]的聊天端点。这很有帮助，因为我不必重新框架化我的提示。这是我使用requests的最小请求。模型名称因账户和地区而异，在我的测试中我看到了像deepseek-chat和deepseek-reasoner这样的引用。如果你的文档提到”V4”模型字符串，请使用该字符串。否则，选择你的控制台中列出的最接近的通用聊天模型。

import os
import requests

API_KEY = os.environ["DEEPSEEK_API_KEY"]
BASE_URL = "https://api.deepseek.com/v1/chat/completions"

payload = {
    "model": "deepseek-chat", # check docs/console for the exact model
    "messages": [
        {"role": "system", "content": "You are a concise assistant."},
        {"role": "user", "content": "Give me two bullet points on the value of clear commit messages."}
    ],
    "temperature": 0.3,
    "max_tokens": 200
}

resp = requests.post(
    BASE_URL,
    headers={
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    },
    json=payload,
    timeout=30
)

resp.raise_for_status()

data = resp.json()
content = data["choices"][0]["message"]["content"]
print(content)

现场笔记

第一次运行无事件发生（令人欣慰）。结构与我的预期相符，这使得迁移一个小的提示库很快。
我保持温度低以获得可重复的答案。这听起来很明显，但我在故障排除时仍然会忘记。
如果你需要确定性的运行，如果API支持，也固定top_p和seed。当文档沉默时，我假设非确定性。

如果你在比较供应商，这里的优势是低摩擦。缺点是差异隐藏在边缘、错误有效负载、令牌计数和流传输形状中。这些边缘是你的集成感觉稳健还是烦人的地方。

代码生成示例

我不要求模型编写完整模块。这变成了一个清理工作。但对于小助手，比如”解析这个时间戳格式”或”使用占位符起草SQL”，这很方便。我使用了一个狭隘的提示、一个清晰的合同和小的输出限制。这使模型不会游荡，并使差异易于审查。

import requests, os

API_KEY = os.environ["DEEPSEEK_API_KEY"]
BASE_URL = "https://api.deepseek.com/v1/chat/completions"

messages = [
    {"role": "system", "content": (
        "You generate small, safe Python helpers. "
        "Return only code inside one block."
    )},
    {"role": "user", "content": (
        "Write a Python function `parse_yyyymmdd` that takes a string like '2026-01-31' "
        "and returns a datetime.date. If invalid, return None. No external deps."
    )}
]

resp = requests.post(
    BASE_URL,
    headers={
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    },
    json={
        "model": "deepseek-chat", # or your V4-capable model
        "messages": messages,
        "temperature": 0,
        "max_tokens": 250
    },
    timeout=30
)

resp.raise_for_status()
code = resp.json()["choices"][0]["message"]["content"]
print(code)

实践中有帮助的是

我总是告诉它只返回代码。如果我跳过这个，我会得到我不需要的总结句子。
温度0减少了繁琐的编辑。
我无论如何都要读一遍逻辑。在我的运行中，它处理了ValueError，但我仍然为空格添加了额外的测试。现在额外的两分钟可以节省数小时的意外。

这在第一次射击中没有节省时间。经过三四个小助手后，我注意到它减少了精神努力：更少的标签切换，更少的”确切的strptime代码是什么？“时刻。这对我来说足够了。

流传输响应

我喜欢对任何可能增长的提示进行流传输。它让我可以在答案漂移时提前保释，并使长响应感觉不那么沉重。在我的测试中，DeepSeek的流传输使用了通常的模式：设置stream=true并读取数据行直到[DONE]。我不需要一个特殊的客户端，带有iter_lines的requests就可以了。

import os, json, requests

API_KEY = os.environ["DEEPSEEK_API_KEY"]
BASE_URL = "https://api.deepseek.com/v1/chat/completions"

payload = {
    "model": "deepseek-chat",
    "messages": [
        {"role": "system", "content": "Be brief."},
        {"role": "user", "content": "Summarize this: Streaming keeps the UI responsive and lets me stop early."}
    ],
    "stream": True,
    "temperature": 0.2,
}

with requests.post(
    BASE_URL,
    headers={
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    },
    json=payload,
    stream=True,
    timeout=60
) as r:
    r.raise_for_status()
    for line in r.iter_lines(decode_unicode=True):
        if not line:
            continue
        if line.startswith("data: "):
            chunk = line[len("data: "):]
            if chunk == "[DONE]":
                break
            try:
                obj = json.loads(chunk)
                delta = obj["choices"][0]["delta"].get("content", "")
                if delta:
                    print(delta, end="", flush=True)
            except json.JSONDecodeError:
                # I keep a small log when this happens: usually network blips
                pass

print()

我喜欢的两个小行为：

早期令牌到达很快（在干净的连接上一两秒钟）。不科学，只是足以在我将其连接到CLI工具时感到敏捷。
[DONE]标记可靠地显示。这听起来微不足道，直到它不显示为止，缺少的终止符使UI挂起。

如果你需要流传输到Web应用程序，我会在其间放置一个薄的服务器层来规范化事件。这是一个额外的步骤，但它使你的前端保持简单。

服务器发送事件

在幕后，你实际上是在读取服务器发送事件。如果你更喜欢一个帮助程序，sseclient-py有效，但只要你防范部分行和超时，这里自己动手就可以了。DeepSeek API文档上的流传输页面足以在没有意外的情况下运行。

错误处理

我的大多数错误都是可预测的：缺少密钥、错误的模型名称，或当我限制网络以模拟旅行Wi-Fi时超时。一个我重复使用的小模式：

import httpx, time, os

API_KEY = os.environ["DEEPSEEK_API_KEY"]
BASE_URL = "https://api.deepseek.com/v1/chat/completions"

RETRIABLE = {408, 409, 425, 429, 500, 502, 503, 504}

async def chat_once(client, messages):
    resp = await client.post(
        BASE_URL,
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "model": "deepseek-chat",
            "messages": messages,
            "temperature": 0.2,
            "max_tokens": 300,
        },
        timeout=30,
    )
    if resp.status_code == 401:
        raise RuntimeError("Unauthorized. Check DEEPSEEK_API_KEY and account access.")
    if resp.status_code == 404:
        raise RuntimeError("Endpoint or model not found. Confirm model name in console/docs.")
    if resp.status_code in RETRIABLE:
        raise RuntimeError(f"Retryable status: {resp.status_code}")
    resp.raise_for_status()
    return resp.json()

async def chat_with_retries(messages, attempts=4):
    backoff = 0.5
    async with httpx.AsyncClient() as client:
        for i in range(attempts):
            try:
                return await chat_once(client, messages)
            except RuntimeError as e:
                msg = str(e)
                if "Retryable status" in msg and i < attempts - 1:
                    time.sleep(backoff)
                    backoff *= 2
                    continue
                raise

一些实际的注意事项：

速率限制：当我进行并行测试时，我看到了429。指数退避有帮助，但我也添加了小的抖动（随机50–150ms）以避免雷鸣牲畜。
超时卫生：我为快速检查设置了更短的连接/读取超时（5–10s），对于大提示设置了更长的超时。超时不应该默认都是30s：它隐藏了问题。
错误有效负载：当事情失败时，JSON主体包含了一条我可以显示到日志的消息。我仍然将其包装在我自己的异常中，以便我控制到达UI的内容。

如果你的代码库已经说OpenAI风格的架构，这是可管理的：相同的消息形状，略有不同的边缘。主要的事情是对模型名称严格，并在非2xx时记录完整的响应体，以便你不要猜测。从文档的角度来看，我依靠官方DeepSeek API文档来获取参数名称和流传输形状。每当提供商使用熟悉的端点时，就很容易假设对等。我已经学会了首先检查文档，并在客户端之间复制的较少。

谁可能喜欢这个