DeepSeek Agent Harness 2026终极指南 - 第14章第63节 性能优化：连接池、预热与并发调用
第13章 MCP 收官了，工具生态接入完毕。但同样的功能，高手写出来更快更省——性能优化是 Agent 从"能用"到"好用"的一道坎。这节进入第14章：高级实战，先做性能优化：HTTP 连接池复用、API 预热、并发调用与限流、首 token 延迟优化，附优化前后的实测对比数据。
本文导航
- 性能的三个瓶颈
- HTTP 连接池复用
- API 预热：连接握手提前做
- 并发调用与限流
- 首 token 延迟优化
- 完整实现：performance.py
- 实测：优化前后对比
- 小结
性能的三个瓶颈
Agent 的性能损失主要来自三个地方：
| 瓶颈 | 现象 | 原因 |
|---|---|---|
| 连接开销 | 每次调用都慢 | 每次请求重新握手（TCP + TLS） |
| 串行等待 | 多个任务一个等一个 | 没并发，IO 时间被浪费 |
| 首 token 延迟 | 用户要等很久 | 请求到第一个字的时间长 |
这三个瓶颈各有解法：连接池、预热、并发。逐个击破。
HTTP 连接池复用
问题：每次调client.chat()，openai 库底层如果每次新建连接，会重复 TCP 三次握手 + TLS 握手，耗时几百毫秒到几秒。
解法：连接池复用——保持连接不关闭，请求之间复用同一个 HTTP 连接。
openai 库默认就用了httpx的连接池，但我们要确保：
- Client 实例是单例（不要每次调用都 new 一个）
- 连接池参数设置合理
# deep_pilot/performance.py —— 性能优化 v0.8from__future__importannotationsimporthttpxfromopenaiimportOpenAIfromdeep_pilot.configimportsettingsdefcreate_optimized_client()->OpenAI:"""创建带连接池优化的客户端"""# 自定义 httpx 客户端，配置连接池http_client=httpx.Client(timeout=httpx.Timeout(connect=5.0,# 连接超时 5 秒read=120.0,# 读取超时 120 秒（模型生成可能很久）write=5.0,# 写入超时 5 秒pool=10.0,# 连接池获取超时 10 秒),limits=httpx.Limits(max_keepalive_connections=20,# 保持 20 个空闲连接max_connections=100,# 最大 100 个连接keepalive_expiry=30.0,# 空闲连接 30 秒过期),transport=httpx.HTTPTransport(retries=0),# 重试交给上层)returnOpenAI(api_key=settings.deepseek_api_key.get_secret_value(),base_url=settings.deepseek_base_url,http_client=http_client,)关键参数：
max_keepalive_connections=20：保持 20 个连接，复用不重建keepalive_expiry=30：连接空闲 30 秒才关闭max_connections=100：并发上限 100
API 预热：连接握手提前做
问题：第一个请求要经历完整的 TCP + TLS 握手，比后续请求慢。用户第一次调用时，感知的延迟最长。
解法：预热——应用启动时先发一个轻量请求，把连接建立好、缓存热起来。
defwarm_up(client:OpenAI)->None:""" 预热：启动时发一个最小请求，建立连接 + 热缓存。 """try:client.chat.completions.create(model=settings.deepseek_model,messages=[{"role":"user","content":"hi"}],max_tokens=1,# 最小输出，几乎零成本)print("[预热] 连接已建立，缓存已热")exceptExceptionase:print(f"[预热] 预热失败（可忽略）:{e}")预热的两大收益：
- TCP/TLS 握手提前完成：后续请求复用连接
- 服务端缓存预热：DeepSeek 的 Prefix Caching（第64节详讲）——预热请求让常用前缀进入缓存
并发调用与限流
问题：多个独立任务串行跑，IO 时间浪费。
解法：并发（第56节的 asyncio）+限流（防止打爆 API）。
DeepSeek V4.1-Flash 支持 2500 并发，但个人账号有额度限制，需要合理限流。
importasyncioimporttimeasyncdefbatch_call_with_limit(client:OpenAI,prompts:list[str],max_concurrent:int=10,)->list[str]:""" 并发调用，带限流。 prompts: 要处理的提示词列表 max_concurrent: 最大并发数 """semaphore=asyncio.Semaphore(max_concurrent)asyncdef_call_one(prompt:str)->str:asyncwithsemaphore:# 用 to_thread 避免阻塞事件循环resp=awaitasyncio.to_thread(client.chat.completions.create,model=settings.deepseek_model,messages=[{"role":"user","content":prompt}],)returnresp.choices[0].message.contentor""tasks=[_call_one(p)forpinprompts]returnawaitasyncio.gather(*tasks)defrun_batch(prompts:list[str],max_concurrent:int=10)->list[str]:"""同步入口"""returnasyncio.run(batch_call_with_limit(optimized_client,prompts,max_concurrent))首 token 延迟优化
首 token 延迟（Time To First Token, TTFT）是用户体验的关键——用户要等多久才能看到第一个字。
优化手段：
- 流式输出（第38节讲过）：不等完整结果，逐 token 返回
- 预热：连接已建立，省握手时间
- 减少系统提示词：系统提示词太长，模型要先读完才能开始生成
defmeasure_ttft(client:OpenAI,messages:list[dict],stream:bool=True)->float:"""测量首 token 延迟（秒）"""start=time.time()first_token_time=Noneifstream:stream_resp=client.chat.completions.create(model=settings.deepseek_model,messages=messages,stream=True,)forchunkinstream_resp:ifchunk.choices[0].delta.content:first_token_time=time.time()-startbreakelse:resp=client.chat.completions.create(model=settings.deepseek_model,messages=messages,)first_token_time=time.time()-start# 非流式，首 token = 完整结果时间returnfirst_token_timeor(time.time()-start)完整实现：performance.py
# deep_pilot/performance.py —— 完整版from__future__importannotationsimportasyncioimporttimeimporthttpxfromopenaiimportOpenAIfromdeep_pilot.configimportsettingsdefcreate_optimized_client()->OpenAI:http_client=httpx.Client(timeout=httpx.Timeout(connect=5.0,read=120.0,write=5.0,pool=10.0),limits=httpx.Limits(max_keepalive_connections=20,max_connections=100,keepalive_expiry=30.0,),)returnOpenAI(api_key=settings.deepseek_api_key.get_secret_value(),base_url=settings.deepseek_base_url,http_client=http_client,)optimized_client=create_optimized_client()defwarm_up(client:OpenAI=optimized_client)->None:try:client.chat.completions.create(model=settings.deepseek_model,messages=[{"role":"user","content":"hi"}],max_tokens=1,)print("[预热] 连接已建立，缓存已热")exceptExceptionase:print(f"[预热] 预热失败（可忽略）:{e}")defrun_batch(prompts:list[str],max_concurrent:int=10)->list[str]:returnasyncio.run(_batch(client=optimized_client,prompts=prompts,max_concurrent=max_concurrent))asyncdef_batch(client,prompts,max_concurrent):semaphore=asyncio.Semaphore(max_concurrent)asyncdef_one(p):asyncwithsemaphore:resp=awaitasyncio.to_thread(client.chat.completions.create,model=settings.deepseek_model,messages=[{"role":"user","content":p}],)returnresp.choices[0].message.contentor""returnawaitasyncio.gather(*[_one(p)forpinprompts])defmeasure_ttft(client:OpenAI,messages:list[dict])->float:start=time.time()stream_resp=client.chat.completions.create(model=settings.deepseek_model,messages=messages,stream=True,)forchunkinstream_resp:ifchunk.choices[0].delta.content:returntime.time()-startreturntime.time()-start实测：优化前后对比
uv run python-c" import time from openai import OpenAI from deep_pilot.config import settings from deep_pilot.performance import optimized_client, warm_up, run_batch # 未优化：每次新建 client（无连接池） def unoptimized_call(prompt): c = OpenAI(api_key=settings.deepseek_api_key.get_secret_value(), base_url=settings.deepseek_base_url) c.chat.completions.create(model=settings.deepseek_model, messages=[{'role':'user','content':prompt}]) # 1. 连接池对比：10 次调用 start = time.time() for i in range(10): unoptimized_call('hello') unopt_time = time.time() - start start = time.time() for i in range(10): optimized_client.chat.completions.create(model=settings.deepseek_model, messages=[{'role':'user','content':'hello'}]) opt_time = time.time() - start print(f'10次调用：未优化 {unopt_time:.2f}s vs 连接池 {opt_time:.2f}s') # 2. 预热对比 start = time.time() optimized_client.chat.completions.create(model=settings.deepseek_model, messages=[{'role':'user','content':'hi'}]) cold_time = time.time() - start print(f'冷启动首调用: {cold_time:.2f}s') warm_up() start = time.time() optimized_client.chat.completions.create(model=settings.deepseek_model, messages=[{'role':'user','content':'hi'}]) warm_time = time.time() - start print(f'预热后调用: {warm_time:.2f}s') "控制台输出（精简）：
10次调用：未优化 8.5s vs 连接池 5.2s （加速 1.6x） 冷启动首调用: 1.8s [预热] 连接已建立，缓存已热 预热后调用: 1.2s （首调用快 33%）优化效果：
- 连接池：10 次调用从 8.5s 降到 5.2s，加速 1.6 倍
- 预热：首调用从 1.8s 降到 1.2s，快 33%
小结
- 三个瓶颈：连接开销、串行等待、首 token 延迟，各有解法。
- 连接池复用：httpx 保持 20 个连接不重建，10 次调用加速 1.6x。
- API 预热：启动时发最小请求，握手提前 + 缓存热，首调用快 33%。
- 并发调用：asyncio + 信号量限流，IO 密集任务并行，不浪费等待时间。
- 首 token 延迟：流式输出 + 预热 + 精简提示词，用户更快看到第一个字。
- 单例 Client：不要每次 new 客户端，否则连接池形同虚设。
- DeepPilot v0.8 性能优化完成——从"能用"到"好用又快"。
下节预告
性能优化让 Agent 更快了，但成本还没动——同样的功能，有人花 10 元，有人只花 1 元。下一节讲前缀缓存的秘密：DeepSeek 缓存命中只要 0.02 元/百万 token，比未命中便宜 50 倍。系统提示词和工具定义的稳定前缀设计 + 命中率监控，把账单砍掉一个零。
如果觉得本文对你有帮助，欢迎点赞、收藏、关注三连！
本系列持续更新中，关注不迷路~
