DeepSeek API 生产级调用实战：流式传输、连接池与指数退避重试

1. 从一次线上事故说起：为什么裸调 DeepSeek API 迟早要出事

去年底我接手了一个内部知识问答工具，后端用 Python 调 DeepSeek API 做流式问答。上线第一周风平浪静，第二周开始陆续有同事反馈"回答卡住不动""偶尔报错要刷新重试"。我一开始以为是前端渲染问题，排查了半天才发现根子在后端：每次请求都新建一个requests.Session，没有连接复用；网络抖动时直接抛异常给前端，没有任何重试；流式响应读取时没做超时控制，遇到服务端慢响应就无限挂起。

这三个问题单独看都不致命，但叠在一起就是生产环境的定时炸弹。后来我把这套调用逻辑彻底重写了一遍，核心就三件事：流式传输保证首字延迟和用户体验，连接池复用降低握手开销和端口耗尽风险，指数退避重试兜住瞬时故障。改完之后，同样的并发量下平均响应时间降了约 40%，错误率从千分之几压到了几乎为零。

这篇就把这套生产级调用方案完整拆开讲。适合已经会写 Python、能跑通 DeepSeek API 基础调用，但准备把它放进真实业务里的朋友。如果你还停留在"能调通就行"的阶段，这篇里的很多坑你迟早会踩。下面所有代码都是我在实际项目里跑过的，参数和结构可以直接抄。

2. 流式传输：不只是"边收边显示"那么简单

2.1 流式到底解决了什么问题

很多人对流式的理解停留在"打字机效果好看"。这只是表象。流式真正的价值在于首字节时间（TTFT）。非流式调用下，用户要等模型把整段话生成完才能看到第一个字，一个 500 字的回答可能要等 8 到 15 秒，期间界面一片空白，用户会以为卡死了。流式调用下，通常 1 到 2 秒内就能吐出第一个 token，用户立刻知道"它在工作"。

从工程角度看，流式还带来一个隐性好处：内存占用可控。非流式要把整个响应体缓存在内存里再解析，长回答加上高并发，内存峰值很难看。流式是逐块处理，处理完就丢，内存曲线平稳得多。

DeepSeek API 的流式接口遵循 OpenAI 兼容格式，请求体里加"stream": true，响应就是 SSE（Server-Sent Events）格式的文本流，每行以data:开头，最后以data: [DONE]结束。理解这个格式是正确解析的前提。

2.2 用 requests 做流式读取的正确姿势

先看一段能直接用的代码：

import json import requests def stream_chat(session, url, headers, payload, timeout=(5, 60)): """ session: 复用的 requests.Session timeout: (连接超时, 读取超时) 元组 """ with session.post( url, headers=headers, json=payload, stream=True, # 关键：开启流式 timeout=timeout, ) as resp: resp.raise_for_status() for line in resp.iter_lines(decode_unicode=True): if not line: continue if line.startswith("data: "): data = line[6:] if data.strip() == "[DONE]": break try: chunk = json.loads(data) except json.JSONDecodeError: continue delta = chunk["choices"][0].get("delta", {}) content = delta.get("content") if content: yield content

这段代码有几个细节值得说。stream=True是必须的，否则iter_lines拿不到增量数据。timeout用元组形式，连接超时设短一点（5 秒），读取超时设长一点（60 秒），因为模型生成过程中两次数据块之间可能有间隔，读取超时太短会误杀正常请求。

iter_lines(decode_unicode=True)会自动处理编码，但要注意它按行切分，SSE 的每个data:行是独立的 JSON。空行要跳过，[DONE]要识别并终止循环。json.loads一定要包 try，因为网络传输中偶尔会出现半截数据，直接抛异常会中断整个流。

2.3 流式场景下最容易忽略的三个坑

第一个坑：超时设置不当导致长回答被截断。我见过有人把读取超时设成 10 秒，结果模型思考时间稍长就被掐断，用户看到回答说到一半没了。正确做法是读取超时给足，比如 60 到 120 秒，同时在前端做"心跳"提示，让用户知道还在生成。

第二个坑：没有处理delta里可能为空的情况。有些 chunk 的delta只有role字段没有content，或者content是空字符串。如果不做判断直接拼接，会拼进一堆空值。上面代码里if content:就是干这个的。

第三个坑：异常处理缺失导致连接泄漏。流式响应如果中途抛异常，with语句能保证连接被正确关闭。但如果你用的是手动resp = session.post(...)而不加with，异常时连接不会归还连接池，跑久了连接池就枯竭了。这一点和下一节的连接池直接相关。

提示：流式接口的raise_for_status()要在开始迭代之前调用。如果服务端返回 4xx/5xx，响应体可能不是 SSE 格式，提前检查能避免解析出一堆垃圾。

3. 连接池复用：省下的不只是那几十毫秒

3.1 每次新建连接到底浪费了什么

HTTP 请求建立连接要经历 TCP 三次握手，如果是 HTTPS 还要加上 TLS 握手，这一套下来在局域网内可能几十毫秒，跨公网可能上百毫秒甚至更多。如果你每次调 API 都新建一个连接，这部分开销就白白重复了。

更严重的是端口耗尽。每个 TCP 连接占用一个本地端口，操作系统默认的临时端口范围有限（通常几千个）。高并发下如果连接不复用又不及时关闭，会出现Cannot assign requested address错误。我有个朋友的服务在压测时突然大面积报错，查了半天就是这个原因。

requests.Session内部维护了一个连接池，同一个 Session 实例对同一个 host 的请求会复用底层 TCP 连接。所以核心原则很简单：全局复用一个 Session，不要每次请求都新建。

3.2 连接池参数怎么配才合理

requests底层用的是urllib3的连接池，可以通过HTTPAdapter精细控制：

import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def build_session(pool_size=20, pool_maxsize=50): session = requests.Session() adapter = HTTPAdapter( pool_connections=pool_size, # 连接池数量（不同 host 各一个池） pool_maxsize=pool_maxsize, # 每个池最大连接数 max_retries=Retry(total=0), # 重试我们自己控制，这里关掉 pool_block=False, # 池满时是否阻塞等待 ) session.mount("https://", adapter) session.mount("http://", adapter) return session

pool_connections是池的数量，一般等于你会访问的不同域名数。只调 DeepSeek 一个域名的话，设成 10 到 20 足够。pool_maxsize是每个池能保持的最大连接数，这个要结合你的并发量来定。假设你的服务峰值 QPS 是 30，平均每个请求耗时 2 秒，那并发连接数大约 60，pool_maxsize至少要给到 60 以上，否则超出的请求会排队。

pool_block=False意味着池满时新建连接而不是阻塞等待。这个设置要谨慎：设 False 在高并发下可能瞬间创建大量连接，设 True 则请求会排队。我的经验是配合合理的pool_maxsize，用 False 更稳，因为排队会累积延迟。

max_retries这里设成 0，是因为我们要自己实现指数退避重试，用 urllib3 自带的重试会和我们的逻辑打架，还不好控制退避策略。

3.3 连接池复用的实测收益与注意事项

我在一个日请求量约 50 万的场景下做过对比：不复用连接时，P99 延迟约 1.8 秒；改用全局 Session 加连接池后，P99 降到约 1.1 秒。这 700 毫秒里，握手开销占了相当一部分。

但连接池不是配了就万事大吉，有几个坑要注意。

坑一：Session 不是线程安全的。官方文档明确说Session对象在多线程下不保证安全。如果你用多线程并发调用，要么每个线程一个 Session，要么用线程局部存储。我一般用threading.local()给每个线程分配独立 Session，既复用了连接又避免了竞争。

坑二：连接会被服务端主动关闭。长连接闲置一段时间后，服务端或中间设备可能悄悄关掉它，而客户端不知道，下次复用时就报Connection aborted。解决办法是设置keep-alive相关的参数，或者干脆在重试逻辑里把这类异常也纳入重试范围。这也是为什么下一节的重试机制必不可少。

坑三：DNS 缓存问题。如果服务端 IP 变了，Session 可能还拿着旧 IP 的连接。生产环境建议给 Session 配置合理的生命周期，定期重建。不过对于 DeepSeek 这种稳定服务，这个问题不突出，了解即可。

4. 指数退避重试：把瞬时故障挡在用户视线之外

4.1 为什么是"指数退避"而不是"固定间隔"

网络请求失败的原因分两类：一类是永久性错误，比如参数错误、鉴权失败，重试多少次都没用；另一类是瞬时故障，比如网络抖动、服务端限流、临时过载，这类错误等一会儿再试往往就成功了。

对于瞬时故障，固定间隔重试有个致命问题：如果服务端正在过载，所有客户端都以相同频率猛敲，会形成"惊群效应"，把服务端压得更死。指数退避让每次重试的等待时间翻倍，给服务端喘息时间，同时加上随机抖动（jitter）避免多个客户端同步重试。

DeepSeek API 在限流时通常返回 429 状态码，服务端临时故障返回 5xx。这两类都值得重试。而 400、401、403 这类就别重试了，纯属浪费。

4.2 一个可直接用的退避重试实现

import time import random import requests RETRYABLE_STATUS = {429, 500, 502, 503, 504} RETRYABLE_EXC = ( requests.exceptions.ConnectionError, requests.exceptions.Timeout, requests.exceptions.ChunkedEncodingError, ) def call_with_backoff(func, max_retries=5, base_delay=1.0, max_delay=30.0): """ func: 无参可调用对象，内部执行一次请求 返回 func 的结果，全部失败则抛出最后一次异常 """ last_exc = None for attempt in range(max_retries + 1): try: return func() except requests.exceptions.HTTPError as e: status = e.response.status_code if e.response is not None else None if status not in RETRYABLE_STATUS: raise # 不可重试，直接抛 last_exc = e except RETRYABLE_EXC as e: last_exc = e except Exception: raise # 其他异常不重试 if attempt == max_retries: break # 指数退避 + 抖动 delay = min(base_delay * (2 ** attempt), max_delay) delay = delay * (0.5 + random.random()) # 抖动系数 0.5~1.5 time.sleep(delay) raise last_exc

这段逻辑的核心是区分可重试与不可重试。RETRYABLE_STATUS只包含限流和 5xx，RETRYABLE_EXC只包含网络类异常。其他异常直接抛出，避免无意义重试。

退避公式base_delay * (2 ** attempt)是标准指数退避：第 0 次等 1 秒，第 1 次等 2 秒，第 2 次等 4 秒，以此类推。max_delay封顶防止等待过久。抖动系数0.5 + random.random()让实际等待在计算值的 0.5 到 1.5 倍之间浮动，打散重试时间。

4.3 重试与流式的配合：一个容易翻车的组合

流式请求的重试比普通请求麻烦得多。普通请求失败了重发一次就行，流式请求如果已经吐了一部分内容再失败，重试会导致内容重复。我的处理原则是：只在"还没收到任何内容"时重试。

具体做法是在流式函数里记录是否已经 yield 过内容。如果第一个 chunk 都没收到就失败，可以安全重试；如果已经吐了内容，就把异常抛给上层，由业务决定是提示用户重试还是拼接后续内容。

def stream_with_retry(session, url, headers, payload, max_retries=3): for attempt in range(max_retries + 1): got_content = False try: for chunk in stream_chat(session, url, headers, payload): got_content = True yield chunk return except RETRYABLE_EXC as e: if got_content: raise # 已经吐过内容，不能重试 if attempt == max_retries: raise time.sleep(min(1.0 * (2 ** attempt), 10.0))

这个设计的关键判断就是got_content。一旦它为 True，说明用户已经看到部分回答了，重试会造成内容错乱，此时宁可报错让上层处理。

注意：流式重试的退避时间建议比普通请求短一些，因为用户正在等待，等太久体验很差。我一般把上限压到 10 秒以内。

5. 把三块拼起来：一个完整的生产级调用封装

5.1 整体结构设计

前面三节分别讲了流式、连接池、重试，现在把它们组装成一个可直接用的类。设计思路是：Session 全局唯一（或线程局部），连接池参数可配，重试逻辑内聚，流式和非流式共用同一套基础设施。

import threading import requests from requests.adapters import HTTPAdapter class DeepSeekClient: def __init__(self, api_key, base_url, pool_maxsize=50): self.api_key = api_key self.base_url = base_url.rstrip("/") self._local = threading.local() self._pool_maxsize = pool_maxsize @property def session(self): # 每个线程一个 Session，兼顾复用与线程安全 if not hasattr(self._local, "session"): s = requests.Session() adapter = HTTPAdapter( pool_connections=10, pool_maxsize=self._pool_maxsize, max_retries=0, ) s.mount("https://", adapter) s.mount("http://", adapter) self._local.session = s return self._local.session def _headers(self): return { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", "Accept": "text/event-stream", } def chat_stream(self, messages, model="deepseek-chat", **kwargs): payload = { "model": model, "messages": messages, "stream": True, **kwargs, } url = f"{self.base_url}/chat/completions" return stream_with_retry( self.session, url, self._headers(), payload )

用threading.local()存 Session 是个折中方案。它保证了线程安全，同时每个线程内部复用连接。如果你的服务是单线程异步（比如用 asyncio），那这套 requests 方案就不合适了，得换 httpx 的异步客户端，但连接池和重试的思路完全一致。

5.2 参数配置的取舍逻辑

几个关键参数我列个表，方便你按自己的场景调：

参数	建议值	调整依据
pool_connections	10	访问的域名数量，单域名 10 足够
pool_maxsize	50~100	峰值并发连接数，按 QPS × 平均耗时估算
连接超时	5 秒	握手很快，超时短一点能快速失败
读取超时	60~120 秒	模型生成慢，给足时间
max_retries	3~5	太多会累积延迟，太少兜不住故障
base_delay	1 秒	首次退避，太小没意义，太大体验差
max_delay	30 秒	退避上限，防止等待过久

pool_maxsize的估算方法再强调一遍：假设峰值 QPS 是 20，每个请求平均耗时 3 秒，那同时活跃的连接约 60 个，pool_maxsize至少给 60，留点余量给 80 到 100 比较稳。

5.3 上线前必须做的几项验证

代码写完不代表能上生产，我一般会做这几项验证。

第一，压测连接复用效果。用ab或wrk打一轮，观察netstat里的连接数是否稳定，而不是随请求数线性增长。如果连接数一直涨，说明复用没生效。

第二，模拟故障验证重试。把 base_url 指向一个会返回 503 的本地服务，看客户端是否正确退避重试，日志里能不能看到退避间隔。这一步能验证重试逻辑真的在跑，而不是被异常吞掉了。

第三，验证流式中断处理。在流式过程中手动断开网络，看客户端是否在"已收到内容"后正确抛出异常而不是无限重试。

第四，观察内存和文件描述符。长时间运行后检查进程的 fd 数量是否稳定。如果持续增长，多半是连接没正确关闭，回头检查with语句和异常分支。

6. 那些文档里不会写的实战经验

6.1 日志要打对地方

生产环境排查问题全靠日志，但日志打多了影响性能，打少了查不到问题。我的做法是：只在重试和异常时打详细日志，正常请求只记耗时和状态。

具体来说，每次重试记录 attempt 次数、退避时长、失败原因；流式请求记录首字节时间和总耗时；异常时把响应体前 500 字符打出来（注意脱敏）。正常成功的请求只记一行摘要。这样日志量可控，出问题时又能快速定位。

6.2 限流要主动配合，别硬刚

DeepSeek API 有速率限制，返回 429 时除了退避重试，更优雅的做法是主动限流。在客户端加一个令牌桶或信号量，把出站请求速率控制在配额以内，从源头减少 429。这比被动重试体验好得多，因为重试意味着用户要多等。

我一般用threading.Semaphore控制并发数，或者用简单的令牌桶控制 QPS。具体阈值参考你的账号配额，留 20% 余量。

6.3 别把重试当成万能药

重试能兜住瞬时故障，但兜不住系统性问题。如果某个接口持续 5xx，重试只会让请求堆积、延迟飙升。所以要在监控里区分"重试后成功"和"重试后仍失败"，后者说明有系统性问题，得从根上查，而不是加大重试次数。

我踩过的一个坑是：某次服务端区域性故障，我的客户端疯狂重试，结果把本地线程池占满，连健康检查都做不了。后来加了熔断逻辑，连续失败到阈值就快速失败一段时间，给系统恢复的机会。

6.4 关于超时的一个反直觉结论

很多人觉得超时设长一点更安全，其实不然。超时设太长，故障时请求会长时间挂起，占用连接和线程资源，反而拖垮整个服务。正确做法是超时设短，配合重试。比如读取超时 60 秒，超过就断开重试，比傻等 300 秒强得多。快速失败加快速重试，整体可用性更高。

这套方案我在两个项目里跑了半年多，日请求量从几万涨到几十万，没再出过因为调用层导致的线上事故。核心就是那三件事：流式保证体验，连接池保证效率，退避重试保证稳定。代码不复杂，难的是把每个参数的取舍想清楚，把每个异常分支处理到位。你要是正准备把 DeepSeek API 接入生产，照着这个结构搭一遍，能省下不少排查时间。