当AI智能体遇上高并发：我是怎么用Redis+负载均衡干掉推理超时的

结合实际踩坑过程，聊聊大模型推理服务在高并发场景下的调优思路。

一、先说问题：推理超时到底有多烦？

做过AI智能体服务的同学应该都遇到过这个场景：

压测一跑，QPS刚上去，告警就炸了。日志里全是：

TimeoutError: inference request exceeded 30s Connection pool exhausted upstream timed out (110: Operation timed out)

单次推理本来只要2~5秒，并发一高就直接超时。更难受的是，这种问题不稳定，有时候复现，有时候又好好的，排查起来非常头疼。

问题根源在哪？

大模型推理和传统API服务有本质区别：

所以你不能用对待普通微服务的思路来处理这个问题。

二、架构全景：我们的解决方案长什么样

先上整体架构图（文字版）：

用户请求 │ ▼ API Gateway（限流 + 鉴权） │ ▼ Load Balancer（Nginx / 自研调度层） │ ├──────────────────────────┐ ▼ ▼ 推理节点 A 推理节点 B (vLLM / TGI) (vLLM / TGI) │ │ └──────────┬───────────────┘ ▼ Redis 缓存层 （语义缓存 + 结果缓存） │ ▼ 业务逻辑层

两个核心模块：Redis缓存层和负载均衡调度层，缺一不可，但职责完全不同。

三、Redis缓存：不是简单的KV存储

很多人一听到"缓存推理结果"，第一反应是：把prompt做key，把response做value，存Redis，完事。

这个思路方向对，但实际落地会踩很多坑。

3.1 精确匹配缓存（基础方案）

最简单的实现：

importhashlibimportjsonimportredis r=redis.Redis(host='localhost',port=6379,decode_responses=True)defget_cache_key(prompt:str,model:str,params:dict)->str:"""生成缓存key，注意要把模型参数也纳入"""payload={"prompt":prompt,"model":model,"temperature":params.get("temperature",0.7),"max_tokens":params.get("max_tokens",512)}raw=json.dumps(payload,sort_keys=True,ensure_ascii=False)returnf"llm:cache:{hashlib.sha256(raw.encode()).hexdigest()}"defquery_with_cache(prompt:str,model:str,params:dict):key=get_cache_key(prompt,model,params)# 先查缓存cached=r.get(key)ifcached:returnjson.loads(cached),True# True表示命中缓存# 缓存未命中，走推理result=call_inference_api(prompt,model,params)# 写缓存，TTL根据业务设定r.setex(key,3600,json.dumps(result,ensure_ascii=False))returnresult,False

这个方案的命中率很低，只有完全相同的请求才能命中。在智能体场景下，用户输入千变万化，基本没什么效果。

3.2 语义缓存（进阶方案）

问：用户问"今天天气怎么样"和"现在天气如何"，语义上是一样的，为什么不能复用同一个缓存？

答：可以，但需要引入向量相似度检索。

fromsentence_transformersimportSentenceTransformerimportnumpyasnpimportredisfromredis.commands.search.queryimportQuery# 使用Redis Vector Search（需要RedisSearch模块）model_embed=SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')defsemantic_cache_lookup(prompt:str,threshold:float=0.92):""" 语义相似度缓存查找 threshold: 相似度阈值，越高越严格 """query_vec=model_embed.encode(prompt).astype(np.float32).tobytes()# 使用Redis向量检索找最近邻q=(Query("*=>[KNN 3 @embedding $vec AS score]").sort_by("score").return_fields("prompt","response","score").paging(0,3).dialect(2))results=r.ft("idx:llm_cache").search(q,query_params={"vec":query_vec})fordocinresults.docs:similarity=1-float(doc.score)# 余弦距离转相似度ifsimilarity>=threshold:print(f"语义缓存命中，相似度:{similarity:.4f}")returndoc.responsereturnNone

实测数据对比：

阈值调低会提升命中率，但可能返回语义相近但不完全准确的答案，这个trade-off需要根据业务场景决定。

3.3 缓存预热：冷启动问题怎么解

智能体刚上线时，缓存是空的，所有请求都会穿透到推理层，很容易打崩。

importasyncio# 高频问题列表（从历史日志分析得出）HOT_PROMPTS=["你好，请介绍一下你自己","帮我写一份工作总结","解释一下什么是机器学习",# ... 更多高频prompt]asyncdefwarm_up_cache():"""启动时异步预热缓存"""tasks=[]forpromptinHOT_PROMPTS:tasks.append(asyncio.create_task(preload_single(prompt)))# 限制并发，别把推理层压垮semaphore=asyncio.Semaphore(5)asyncwithsemaphore:awaitasyncio.gather(*tasks)print(f"缓存预热完成，共预热{len(HOT_PROMPTS)}条")

四、负载均衡：GPU节点的调度不能照搬CPU那套

问：直接用Nginx轮询不行吗？

不是不行，是不够好。

Nginx的轮询/加权轮询是基于连接数的，它不知道每个推理节点当前的GPU显存占用、推理队列深度、平均响应时间。结果就是：有的节点队列已经堆满了，新请求还在往里怼；有的节点闲着，没人分配。

4.1 基于节点健康度的动态调度

importasyncioimportaiohttpfromdataclassesimportdataclassfromtypingimportListimporttime@dataclassclassInferenceNode:host:strport:intweight:float=1.0queue_depth:int=0# 当前队列深度avg_latency:float=0.0# 近期平均延迟（秒）gpu_memory_used:float=0.0# GPU显存占用率is_healthy:bool=Truelast_check:float=0.0classSmartLoadBalancer:def__init__(self,nodes:List[InferenceNode]):self.nodes=nodes self.check_interval=5# 秒defcompute_score(self,node:InferenceNode)->float:""" 综合打分，分数越低越优先分配 综合考虑：队列深度、延迟、显存占用 """ifnotnode.is_healthy:returnfloat('inf')score=(node.queue_depth*0.5+node.avg_latency*0.3+node.gpu_memory_used*0.2)returnscoredefpick_node(self)->InferenceNode:"""选出当前最优节点"""healthy_nodes=[nforninself.nodesifn.is_healthy]ifnothealthy_nodes:raiseRuntimeError("所有推理节点不可用")returnmin(healthy_nodes,key=self.compute_score)asyncdefhealth_check_loop(self):"""后台定期拉取各节点指标"""whileTrue:awaitasyncio.gather(*[self._check_node(node)fornodeinself.nodes])awaitasyncio.sleep(self.check_interval)asyncdef_check_node(self,node:InferenceNode):url=f"http://{node.host}:{node.port}/metrics"try:asyncwithaiohttp.ClientSession()assession:asyncwithsession.get(url,timeout=aiohttp.ClientTimeout(total=3))asresp:ifresp.status==200:metrics=awaitresp.json()node.queue_depth=metrics.get("queue_depth",0)node.avg_latency=metrics.get("avg_latency_seconds",0)node.gpu_memory_used=metrics.get("gpu_memory_utilization",0)node.is_healthy=Trueelse:node.is_healthy=FalseexceptException:node.is_healthy=Falsenode.last_check=time.time()

4.2 请求重试与熔断

节点偶尔抖动是正常的，不能因为一次超时就放弃整个请求：

importasynciofromfunctoolsimportwrapsdefwith_retry(max_retries=3,backoff_base=0.5):""" 带指数退避的重试装饰器 注意：重试要换节点，不能打同一个节点 """defdecorator(func):@wraps(func)asyncdefwrapper(self,prompt,*args,**kwargs):last_exception=Nonetried_nodes=set()forattemptinrange(max_retries):node=self.pick_node_excluding(tried_nodes)ifnodeisNone:breaktried_nodes.add(node.host)try:returnawaitfunc(self,prompt,node,*args,**kwargs)exceptasyncio.TimeoutErrorase:last_exception=e wait_time=backoff_base*(2**attempt)print(f"节点{node.host}超时，{wait_time}s 后重试第{attempt+1}次")awaitasyncio.sleep(wait_time)# 临时降低该节点权重node.weight=max(0.1,node.weight*0.5)raiselast_exceptionorRuntimeError("所有重试均失败")returnwrapperreturndecorator

五、两者协同：请求进来后完整链路是这样的

收到请求 │ ▼ ① 语义缓存查询（Redis，< 50ms） │ ├── 命中 ──────────────────► 直接返回，结束 │ └── 未命中 │ ▼ ② 限流检查（令牌桶） │ ├── 超限 ──────────► 返回 429，排队或拒绝 │ └── 通过 │ ▼ ③ 负载均衡选节点（综合评分） │ ▼ ④ 推理请求（带超时+重试） │ ▼ ⑤ 结果写入Redis缓存 │ ▼ ⑥ 返回结果

这条链路里，缓存是第一道防线，能挡掉30%~50%的请求；负载均衡是第二道，保证推理层不被压垮。

六、上线前后的数据对比

在某智能客服项目中，接入方案前后的对比：

七、几个容易忽视的细节

1. 缓存Key要包含系统提示词（system prompt）

很多同学只对用户输入做hash，忘了system prompt不同会导致完全不同的输出，这是低级错误。

2. 流式输出（Streaming）的缓存策略

流式返回时不能直接缓存，需要在服务端收全响应后再写入缓存，对用户侧保持流式体验。

3. 多模态请求不要无脑缓存

图片、文件类请求，缓存意义不大，还占显存，建议只对纯文本prompt做语义缓存。

4. Redis内存要监控，设好淘汰策略

推荐使用allkeys-lru策略，避免缓存把Redis内存撑爆。

八、总结

推理超时本质上是资源供给和请求压力的错配，Redis缓存解决的是"重复请求的无效消耗"，负载均衡解决的是"有效请求的分发不均"。两者一起上，才能在不无限堆卡的前提下，把推理服务的吞吐和稳定性都做起来。

如果你的场景QPS还不高（比如< 50），优先把语义缓存做好，性价比最高。QPS上来之后，再考虑推理节点的动态调度和熔断机制。

有问题欢迎评论区交流，踩过的坑越多，越值得聊。

— 本文由喜爱AIclaude-sonnet- 4.6 辅助完成