性能优化加速推理和缓存和批量处理提升响应速度Agent跑起来了功能没问题就是慢。用户发一句话转圈等了8秒才出结果。8秒钟在聊天场景里已经算很慢了用户以为卡死了。今天这篇聊怎么把Agent的响应时间压下来。从找瓶颈开始到推理加速、缓存策略、并发优化最后给一个完整的优化方案。做完你会得到一套可运行的性能优化代码实测能把响应时间从8秒压到2秒以内。性能瓶颈在哪优化之前先搞清楚慢在哪。Agent的一次请求大致经过这几个阶段用户输入进来如果走RAG先检索知识库然后把prompt发给LLM推理LLM返回结果后再做后处理。每一步都可能成为瓶颈。LLM推理是最大的耗时来源。调一次GPT-4o的API等个3到5秒很正常。如果你的Agent要多轮调用LLM比如先规划再执行时间翻倍。网络IO也不小。检索向量数据库、调外部API、读缓存这些都是网络往返。单次几十毫秒不多但串起来累加就明显了。检索延迟看知识库大小。几万条数据的向量库检索一次几百毫秒。如果做了rerank再加几百毫秒。找到瓶颈以后就能对症下药。LLM慢就上流式输出和缓存检索慢就优化索引和批量查询网络IO慢就上并发。推理加速流式输出是性价比最高的优化。不等LLM生成完整个回复再返回而是生成一个字就推一个字给用户。总时间没变但用户感知上快了很多因为他立刻看到了响应开始。fromopenaiimportOpenAI clientOpenAI()defstream_chat(messages):流式输出逐字返回生成内容 参数: messages: OpenAI格式的消息列表 格式为[{role: user, content: ...}] 返回: 生成器每次yield一个文本片段 调用方可以用for循环逐个接收 # streamTrue开启流式输出# API会通过SSE持续推送生成的内容片段responseclient.chat.completions.create(modelgpt-4o-mini,messagesmessages,streamTrue,# 关键参数开启流式返回)# 遍历流式响应每次拿到一个chunk# chunk.choices[0].delta.content是本次增量内容# 可能是None需要判断forchunkinresponse:contentchunk.choices[0].delta.contentifcontent:yieldcontent模型量化是另一条路适用于本地部署模型的情况。把模型从16位精度压缩到8位甚至4位推理速度能快不少显存占用也降下来。代价是精度损失输出质量会略微下降。用bitsandbytes或AutoGPTQ这些库可以做量化。这个方案只对你自己部署的开源模型有效调商业API的话用不了。vLLM是当前最流行的推理加速框架。它用PagedAttention技术管理显存吞吐量比原生Transformers高好几倍。如果你要自己部署开源大模型vLLM几乎是目前最优解。部署方式也很简单一条命令拉起一个兼容OpenAI API格式的服务。python -m vllm.entrypoints.openai.api_server --model meta-llama/Llama-2-7b-chat-hf缓存策略缓存是减少重复计算的利器。用户问公司地址是什么第一次去查LLM第二次同样的问题直接从缓存取结果。Agent场景里重复问题其实不少尤其是客服类应用。精确缓存最简单用问题文本做keyMD5一下存到Redis里。缺点是稍微换个说法就命中不了公司地址和公司在哪里明明是同一个问题但key不一样。语义缓存更聪明。把问题转成向量新问题来了先跟缓存里的问题做相似度比较相似度超过阈值就认为问的是同一件事直接返回缓存的答案。importnumpyasnpfromopenaiimportOpenAI clientOpenAI()classSemanticCache:语义缓存基于向量相似度判断是否命中 原理: 把问题转成embedding向量存起来 新问题来了先算跟历史问题的余弦相似度 超过阈值就直接返回缓存的答案 比精确匹配强能识别不同问法的相同问题 def__init__(self,threshold0.92):# 相似度阈值越高越严格# 0.92表示问题很接近才算命中# 太高命中不了太低会误命中需要根据数据调self.thresholdthreshold# 缓存存储实际项目用Redis或向量数据库# 这里用内存列表演示原理self.cache[]# 每项是{question, answer, embedding}def_get_embedding(self,text):获取文本的embedding向量respclient.embeddings.create(modeltext-embedding-3-small,inputtext,)returnresp.data[0].embeddingdef_cosine_similarity(self,vec_a,vec_b):计算两个向量的余弦相似度 公式是 dot(a,b) / (|a| * |b|) 值域[-1, 1]越接近1表示越相似 anp.array(vec_a)bnp.array(vec_b)returnnp.dot(a,b)/(np.linalg.norm(a)*np.linalg.norm(b))defget(self,question):查询缓存 参数: question: 用户问题文本 返回: 命中则返回缓存的答案未命中返回None # 先把问题转成向量query_embself._get_embedding(question)# 遍历缓存找最相似的best_score0best_answerNoneforiteminself.cache:scoreself._cosine_similarity(query_emb,item[embedding])ifscorebest_score:best_scorescore best_answeritem[answer]# 超过阈值才算命中ifbest_scoreself.threshold:print(f缓存命中相似度{best_score:.3f})returnbest_answerreturnNonedefset(self,question,answer):写入缓存embself._get_embedding(question)self.cache.append({question:question,answer:answer,embedding:emb,})批量处理和并发Agent里有很多可以并行的操作。比如RAG检索多个子问题串行查5次要2.5秒并发查只要0.5秒。用asyncio做并发很方便。importasynciofromopenaiimportAsyncOpenAI# 使用异步客户端跟同步客户端API一样# 区别是调用方法前面加awaitasync_clientAsyncOpenAI()asyncdefbatch_llm_calls(prompts):并发调用LLM批量处理多个prompt 参数: prompts: prompt字符串列表 返回: 结果列表顺序跟输入一致 asyncdefcall_one(prompt):单个LLM调用包装成协程respawaitasync_client.chat.completions.create(modelgpt-4o-mini,messages[{role:user,content:prompt}],)returnresp.choices[0].message.content# asyncio.gather并发执行所有调用# 比串行await快N倍N是prompt数量# 总耗时约等于最慢的那个调用resultsawaitasyncio.gather(*[call_one(p)forpinprompts])returnresults# 使用示例asyncdefmain():prompts[什么是Docker,什么是Kubernetes,什么是LangChain,]# 三个调用同时发出总耗时约等于最慢的那个resultsawaitbatch_llm_calls(prompts)forp,rinzip(prompts,results):print(f问题{p}\n回答{r}\n)# 运行异步主函数asyncio.run(main())性能监控优化之前先量化优化之后验证效果。给Agent加一层耗时统计每一步花了多少时间一目了然。importtimeimportfunctoolsdeftimed(func):耗时统计装饰器 用法是在函数上加timed 每次调用会打印函数名和耗时 生产环境可以换成写日志或发到监控系统 functools.wraps(func)defwrapper(*args,**kwargs):starttime.perf_counter()resultfunc(*args,**kwargs)elapsedtime.perf_counter()-startprint(f[耗时]{func.__name__}{elapsed:.3f}s)returnresultreturnwrapper# 使用示例给每个步骤加耗时统计timeddefretrieve(query):模拟检索操作time.sleep(0.3)# 模拟检索耗时return检索结果timeddefllm_call(prompt):模拟LLM调用time.sleep(3)# 模拟LLM推理耗时returnLLM回复timeddefagent_pipeline(question):完整的Agent处理流程# 第一步检索contextretrieve(question)# 第二步LLM推理answerllm_call(context)returnanswer# 跑一次看每步耗时agent_pipeline(测试问题)效果验证实际项目里我做过一轮完整优化记录一下前后对比。优化前用户问一个问题检索知识库0.8秒LLM推理5.2秒后处理0.3秒总共6.3秒加上网络往返差不多8秒。优化后做了三件事。第一加流式输出用户感知响应时间从8秒降到0.5秒因为第一个字0.5秒就出来了。第二加语义缓存重复问题直接返回命中时0.1秒。第三把多步串行LLM调用改成并发3步从15秒降到5秒。再把模型从gpt-4o换成gpt-4o-mini单次推理从5秒降到2秒。最终首次请求从8秒降到2秒缓存命中时0.1秒流式输出让用户0.5秒就看到第一个字。用户满意度直接上去了。踩坑记录第一个坑语义缓存的阈值设太高。我一开始设了0.98结果几乎从来不命中。两个问题的embedding哪怕意思一样余弦相似度也很少到0.98。后来降到0.92命中率上来了偶尔有误命中但概率很低。这个阈值要根据自己的数据测没有万能值拿一批真实问题跑一下统计分布就心里有数了。第二个坑asyncio跟同步代码混用。我把异步的batch_llm_calls放在一个同步函数里直接调用没加asyncio.run()结果协程创建了但没执行函数返回了一堆协程对象。异步代码要么全异步要么用asyncio.run()显式驱动别混着来。混用是Python异步编程里最常见的坑。第三个坑缓存过期策略。知识库更新了但缓存里还是旧答案用户问到的全是过时信息。后来给缓存加了TTL默认1小时过期。知识库更新的时候主动清缓存。不做这步的话缓存反而成了bug源头用户投诉数据不对你查了半天发现是缓存没更新。延伸与判断性能优化是个持续的事没有一劳永逸的方案。用户量上去了新瓶颈又会出现。监控数据是基础没有量化数据就不知道优化了什么、效果怎么样。流式输出是必做的成本几乎为零效果立竿见影。语义缓存适合问题重复率高的场景如果是探索性问答就不太管用。并发优化要注意API的速率限制并发太高会被限流反而更慢。模型选择上别什么事都上最强的模型。简单的问题用小模型复杂的问题才用大模型。搞个路由层根据问题难度选模型成本和速度都能优化。这也是一个值得单独展开的话题。结尾性能优化的核心思路就是找瓶颈、对症下药、量化验证。8秒到2秒靠的是多个手段叠加流式输出、缓存、并发、模型选择缺一不可。先让用户看到响应再慢慢把总时间压下来。下一篇聊安全防护Agent上线后怎么防Prompt注入和数据泄露。