Python 调用 DeepSeek API 流式输出超时怎么调

原创 1小时前 2

#DeepSeek API Python 流式输出 #DeepSeek API streaming timeout #Python requests 流式请求超时设置 #DeepSeek API 调用示例 Python #OpenAI SDK stream 超时重试

在 Python 中调用 DeepSeek API 做流式输出时，很多开发者会直接照搬非流式示例，结果遇到请求中途挂起、read timeout 或数据丢失。流式场景的超时逻辑与非流式完全不同，需要单独设计。

为什么非流式超时配置在流式下会失效

非流式请求中，timeout 通常指整个请求从发出到收到完整响应的总时长。服务端在较短时间内返回完整 JSON，超时设置相对简单。

流式请求则不同：连接建立后，服务端会持续推送 SSE 分块，响应体不会立即结束。如果沿用非流式的单一 timeout，可能出现两种问题：一是把总超时设得太短，正常的长回答被提前掐断；二是只设了连接超时，读取阶段无限等待，连接假死时无法感知。

因此流式调用必须把超时拆成连接超时和读取超时两个维度。

用 requests 分别设置连接超时与读取超时

requests 的 timeout 参数支持元组形式 (connect_timeout, read_timeout)。连接超时控制 TCP 建连阶段，读取超时控制两次数据到达之间的最大间隔。

import requests

url = "https://api.deepseek.com/chat/completions"
headers = {
    "Authorization": "Bearer ",
    "Content-Type": "application/json",
}
payload = {
    "model": "deepseek-chat",
    "messages": [{"role": "user", "content": "写一段说明"}],
    "stream": True,
    "max_tokens": 2048,
}

with requests.post(
    url,
    headers=headers,
    json=payload,
    stream=True,
    timeout=(5, 30),
) as resp:
    resp.raise_for_status()
    for line in resp.iter_lines(decode_unicode=True):
        if not line:
            continue
        if line.startswith("data: "):
            chunk = line[len("data: "):]
            if chunk == "[DONE]":
                break
            # 解析 chunk 并处理增量内容

这里 (5, 30) 表示建连最多等 5 秒，两次分块之间最多等 30 秒。读取超时不是整个流的总时长，而是相邻数据块的空闲间隔，这一点非常关键。

用 OpenAI 兼容 SDK 控制流式超时

如果使用 OpenAI 兼容 SDK，可以在 client 级别设置 timeout，也可以在单次请求中覆盖。SDK 内部同样区分连接与读取阶段，但具体参数名以所用版本为准。

from openai import OpenAI

client = OpenAI(
    api_key="",
    base_url="https://api.deepseek.com",
    timeout=30.0,
)

stream = client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": "写一段说明"}],
    stream=True,
    max_tokens=2048,
)

for event in stream:
    delta = event.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)

SDK 的 timeout 通常作用于底层 HTTP 客户端。若需要更细粒度控制，可以传入 httpx 的 Timeout 对象，分别指定 connect、read、write、pool 四个阶段。

迭代读取 SSE 分块，避免长时间阻塞

流式响应的本质是逐行读取 SSE。不要一次性调用 resp.text 或 resp.json()，那会等待整个响应结束，失去流式意义，也更容易触发读取超时。

推荐用 iter_lines 或 iter_content 逐块消费。每收到一个有效分块就立即处理并输出，保持连接活跃。如果业务侧有 UI 刷新或落库需求，可以按固定条数或固定时间窗口批量处理，但不要积压到流结束才统一处理。

需要特别注意空行和 data: [DONE] 结束标记。空行是 SSE 的分隔符，直接跳过；[DONE] 表示服务端正常结束，收到后应主动退出循环，避免继续等待。

对中断的流做指数退避重试

流式请求可能因为网络抖动、网关空闲回收或服务端限流而中断。重试时不能简单从头再来，否则会重复输出已生成内容。

一种可控策略是：记录已成功接收的内容长度或最后一条消息 ID，重试时把已生成内容作为上下文的一部分，或者只对尚未完成的部分重新请求。同时配合指数退避，避免短时间内反复冲击服务端。

import time

max_retries = 3
for attempt in range(max_retries):
    try:
        # 发起流式请求并消费
        break
    except (requests.exceptions.ReadTimeout,
            requests.exceptions.ConnectionError):
        if attempt == max_retries - 1:
            raise
        time.sleep(2 ** attempt)

退避基数建议从 1 到 2 秒起步，上限控制在 30 秒以内。如果连续多次读取超时，应检查 max_tokens 是否过大、网络出口是否稳定，而不是无限重试。

max_tokens 与 timeout 的组合边界

max_tokens 决定服务端最多生成多少 token，直接影响流的总时长。timeout 中的读取超时决定相邻分块之间能容忍多久的空闲。两者需要配合：

max_tokens 越大，总生成时间越长，但读取超时不应随之线性放大，因为读取超时管的是分块间隔，不是总时长。
如果读取超时设得过小，比如 5 秒，而模型在复杂推理时出现较长思考间隔，就可能被误判为超时。
如果读取超时设得过大，比如 300 秒，连接假死时无法及时释放资源。

工程上可以先用 30 到 60 秒的读取超时做基线，再根据实际分块间隔分布调整。同时给整个流设置一个业务侧的总时长上限，超过后主动关闭连接，避免单个请求占用过久。

常见失败模式与排查顺序

遇到流式超时，可以按以下顺序排查：

确认是否使用了 stream=True，以及是否逐行迭代而非一次性读取。
检查 timeout 是否为元组形式，读取超时是否被误设为总超时。
观察超时发生在建连阶段还是读取阶段，前者查网络与 DNS，后者查分块间隔。
检查 max_tokens 是否过大导致总时长超出网关限制。
确认重试逻辑是否会导致重复内容或无限循环。

流式输出的超时配置没有万能值，核心是把连接、读取、总时长三个维度分开管理，再结合指数退避和幂等处理，才能在长回答场景下保持稳定。

暂无评论，来说两句吧