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响应就是 SSEServer-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, headersheaders, jsonpayload, streamTrue, # 关键开启流式 timeouttimeout, ) as resp: resp.raise_for_status() for line in resp.iter_lines(decode_unicodeTrue): 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这段代码有几个细节值得说。streamTrue是必须的否则iter_lines拿不到增量数据。timeout用元组形式连接超时设短一点5 秒读取超时设长一点60 秒因为模型生成过程中两次数据块之间可能有间隔读取超时太短会误杀正常请求。iter_lines(decode_unicodeTrue)会自动处理编码但要注意它按行切分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_size20, pool_maxsize50): session requests.Session() adapter HTTPAdapter( pool_connectionspool_size, # 连接池数量不同 host 各一个池 pool_maxsizepool_maxsize, # 每个池最大连接数 max_retriesRetry(total0), # 重试我们自己控制这里关掉 pool_blockFalse, # 池满时是否阻塞等待 ) session.mount(https://, adapter) session.mount(http://, adapter) return sessionpool_connections是池的数量一般等于你会访问的不同域名数。只调 DeepSeek 一个域名的话设成 10 到 20 足够。pool_maxsize是每个池能保持的最大连接数这个要结合你的并发量来定。假设你的服务峰值 QPS 是 30平均每个请求耗时 2 秒那并发连接数大约 60pool_maxsize至少要给到 60 以上否则超出的请求会排队。pool_blockFalse意味着池满时新建连接而不是阻塞等待。这个设置要谨慎设 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_retries5, base_delay1.0, max_delay30.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只包含限流和 5xxRETRYABLE_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_retries3): 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_maxsize50): 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_connections10, pool_maxsizeself._pool_maxsize, max_retries0, ) s.mount(https://, adapter) s.mount(http://, adapter) self._local.session s return self._local.session def _headers(self): return { Authorization: fBearer {self.api_key}, Content-Type: application/json, Accept: text/event-stream, } def chat_stream(self, messages, modeldeepseek-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_connections10访问的域名数量单域名 10 足够pool_maxsize50~100峰值并发连接数按 QPS × 平均耗时估算连接超时5 秒握手很快超时短一点能快速失败读取超时60~120 秒模型生成慢给足时间max_retries3~5太多会累积延迟太少兜不住故障base_delay1 秒首次退避太小没意义太大体验差max_delay30 秒退避上限防止等待过久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 接入生产照着这个结构搭一遍能省下不少排查时间。