【AI应用开发】用户会话多、并发高，Agent 服务如何做性能优化？缓存怎么做？

【AI应用开发】用户会话多、并发高Agent 服务如何做性能优化缓存怎么做目录Agent 服务的性能瓶颈缓存策略全览异步与并发优化LLM 层优化架构层优化完整高并发架构1. Agent 服务的性能瓶颈一次 Agent 调用 N 次 LLM API 调用 M 次工具调用 典型耗时组成: LLM 推理: 1-3s × 2-5次 2-15s 工具执行: 0.1-2s × 2-5次 0.2-10s RAG 检索: 0.2-0.5s × 1-3次 0.2-1.5s 总延迟: 3-25 秒/次对话用户感知非常慢 高并发瓶颈: 1. LLM API 限流: 多数 API 有 RPM/TPM 限制 2. 长连接占用: Agent 执行时间长连接不释放 3. 内存膨胀: 每会话上下文消耗内存 4. 无缓存: 同样问题每次重新完整执行2. 缓存策略全览2.1 多层缓存架构L1: 精确匹配缓存 (Redis, TTL5min) ↓ 未命中 L2: 语义相似缓存 (向量库, TTL30min) ↓ 未命中 L3: 部分结果缓存 (工具调用结果, TTL10min) ↓ 未命中 L4: 实际执行2.2 L1: 精确匹配缓存importhashlibimportredisimportjsonclassExactMatchCache:精确匹配缓存 —— 完全相同的输入直接返回缓存def__init__(self,redis_client,ttl300):self.redisredis_client self.ttlttldef_normalize(self,text:str)-str:标准化输入# 去除多余空格、统一大小写中文不需、去除标点变化textre.sub(r\s, ,text.strip())returntextdefget(self,user_input:str)-Optional[str]:获取缓存keyfcache:exact:{hashlib.md5(self._normalize(user_input).encode()).hexdigest()}cachedself.redis.get(key)returnjson.loads(cached)ifcachedelseNonedefset(self,user_input:str,response:str):写入缓存keyfcache:exact:{hashlib.md5(self._normalize(user_input).encode()).hexdigest()}self.redis.setex(key,self.ttl,json.dumps(response,ensure_asciiFalse))definvalidate(self,pattern:str*):批量失效缓存keysself.redis.keys(fcache:exact:{pattern})ifkeys:self.redis.delete(*keys)# 不缓存的场景NO_CACHE_PATTERNS[r我的.*订单,# 个性化查询rORD-\d{4}-\d{5},# 具体订单号r查询.*积分|余额,# 实时数据]2.3 L2: 语义相似缓存classSemanticCache:语义相似缓存 —— 意思相同的不同问法命中缓存def__init__(self,vector_store,redis_client,similarity_threshold0.92,ttl1800):self.vector_storevector_store self.redisredis_client self.thresholdsimilarity_threshold self.ttlttlasyncdefget(self,user_input:str,session_id:str)-Optional[str]:语义检索缓存query_embeddingawaitself._get_embedding(user_input)# 搜索相似的已缓存问题resultsself.vector_store.search(query_embedding,k1,filter{type:semantic_cache})ifresultsandresults[0][score]self.threshold:cache_keyresults[0][metadata][cache_key]cachedself.redis.get(cache_key)ifcached:# 更新访问时间self.redis.expire(cache_key,self.ttl)returnjson.loads(cached)returnNoneasyncdefset(self,user_input:str,response:str):写入语义缓存embeddingawaitself._get_embedding(user_input)cache_keyfcache:semantic:{hashlib.md5(user_input.encode()).hexdigest()}# 存储响应self.redis.setex(cache_key,self.ttl,json.dumps(response,ensure_asciiFalse))# 存储向量用于语义检索self.vector_store.add(idcache_key,vectorembedding,metadata{type:semantic_cache,cache_key:cache_key,original_query:user_input[:200],timestamp:time.time()})asyncdef_get_embedding(self,text):# 实际的 embedding 调用returnawaitself.embedding_client.encode(text)2.4 L3: 工具调用结果缓存classToolResultCache:工具调用结果缓存 —— 同一工具参数的结果缓存def__init__(self,redis_client,default_ttl600):self.redisredis_client self.default_ttldefault_ttl# 不同工具的缓存 TTLself.ttl_map{search_knowledge_base:1800,# 知识库30分钟query_product:300,# 商品信息5分钟query_inventory:60,# 库存1分钟query_order:0,# 订单不缓存实时变化query_user:0,# 用户信息不缓存}defget(self,tool_name:str,args:dict)-Optional[dict]:获取工具结果缓存ttlself.ttl_map.get(tool_name,-1)ifttl0:returnNone# 该工具不缓存# 构建缓存 keycache_keyself._build_key(tool_name,args)cachedself.redis.get(cache_key)returnjson.loads(cached)ifcachedelseNonedefset(self,tool_name:str,args:dict,result:dict):缓存工具结果ttlself.ttl_map.get(tool_name,self.default_ttl)ifttl0:returncache_keyself._build_key(tool_name,args)self.redis.setex(cache_key,ttl,json.dumps(result,ensure_asciiFalse))def_build_key(self,tool_name:str,args:dict)-str:构建缓存 keyargs_strjson.dumps(args,sort_keysTrue,ensure_asciiFalse)args_hashhashlib.md5(args_str.encode()).hexdigest()returnftool_cache:{tool_name}:{args_hash}definvalidate_tool(self,tool_name:str):使某个工具的所有缓存失效keysself.redis.keys(ftool_cache:{tool_name}:*)ifkeys:self.redis.delete(*keys)2.5 缓存集成到 AgentclassCachedAgent:集成缓存的 Agentdef__init__(self,llm,tools):self.llmllm self.toolstools self.exact_cacheExactMatchCache(redis_client)self.semantic_cacheSemanticCache(vector_store,redis_client)self.tool_cacheToolResultCache(redis_client)self.stats{cache_hit:0,cache_miss:0}asyncdefrun(self,user_input:str,session_id:str):# 1. 精确缓存exactself.exact_cache.get(user_input)ifexact:self.stats[cache_hit]1returnexact self.stats[cache_miss]1# 2. 语义缓存semanticawaitself.semantic_cache.get(user_input,session_id)ifsemantic:self.stats[cache_hit]1returnsemantic# 3. 实际执行带工具缓存responseawaitself._execute_with_tool_cache(user_input)# 4. 写入缓存self.exact_cache.set(user_input,response)awaitself.semantic_cache.set(user_input,response)returnresponseasyncdef_execute_with_tool_cache(self,user_input):带工具缓存的执行messages[{role:user,content:user_input}]forstepinrange(15):responseawaitself.llm.chat(messages,toolsself.tools)ifresponse.finish_reasonstop:returnresponse.contentfortcinresponse.tool_calls:# 检查工具缓存cached_resultself.tool_cache.get(tc.name,tc.args)ifcached_result:resultcached_result logger.info(f工具缓存命中:{tc.name})else:# 执行工具resultawaitself._execute_tool(tc)# 缓存结果self.tool_cache.set(tc.name,tc.args,result)messages.append({role:tool,content:json.dumps(result,ensure_asciiFalse)})3. 异步与并发优化3.1 会话内工具并行asyncdefexecute_parallel_tools(self,tool_calls:list)-list:并行执行互不依赖的工具调用# 检查工具之间有无依赖independentself._filter_independent(tool_calls)iflen(independent)1:# 并行执行tasks[self._execute_tool(tc)fortcinindependent]resultsawaitasyncio.gather(*tasks,return_exceptionsTrue)fortc,resultinzip(independent,results):ifisinstance(result,Exception):logger.error(f工具{tc.name}执行失败:{result})returnresultselse:# 单个工具直接执行return[awaitself._execute_tool(tool_calls[0])]def_filter_independent(self,tool_calls):过滤出互不依赖的工具调用# 简单规则: 如果工具类别不同就认为可并行# 实际可根据工具间数据依赖判断categories[self.tools[tc.name].categoryfortcintool_calls]iflen(set(categories))len(categories):returntool_calls# 全部可并行returntool_calls[:1]# 保守只并行第一个3.2 跨会话异步处理classAsyncAgentService:异步 Agent 服务def__init__(self,max_concurrent50):self.semaphoreasyncio.Semaphore(max_concurrent)self.agent_poolAgentPool(min_size5,max_size20)asyncdefchat(self,user_input:str,session_id:str):异步聊天接口asyncwithself.semaphore:# 并发控制agentawaitself.agent_pool.acquire()try:responseawaitagent.run(user_input,session_id)returnresponsefinally:awaitself.agent_pool.release(agent)classAgentPool:Agent 实例池复用 LLM 连接def__init__(self,min_size5,max_size20):self.poolasyncio.Queue()self.min_sizemin_size self.max_sizemax_size self.current_size0asyncdefacquire(self):获取 Agent 实例try:returnself.pool.get_nowait()exceptasyncio.QueueEmpty:ifself.current_sizeself.max_size:self.current_size1returnself._create_agent()returnawaitself.pool.get()asyncdefrelease(self,agent):归还 Agent 实例awaitself.pool.put(agent)4. LLM 层优化classLLMOptimizer:LLM 层优化staticmethoddefuse_small_model_for_simple_tasks(user_input,intent):简单任务用小模型省钱快ifintentin[chitchat,simple_fact]:returngpt-4o-mini# 快 5x便宜 20xelifintentin[procedure,faq]:returngpt-4o# 平衡else:returngpt-4o# 复杂任务才用最好的staticmethoddefstreaming_response(generator):流式输出 —— 用户更快看到第一个字forchunkingenerator:yieldchunk.contentstaticmethoddefprompt_compression(messages,max_tokens8000):压缩 Prompt —— 移除冗余内容# 保留 system prompt 最近 N 轮 工具结果摘要compressed[messages[0]]# system prompt# 最近的消息recentmessages[-10:]# 压缩过长的工具返回formsginrecent:ifmsg[role]toolandlen(str(msg[content]))1000:msg[content]str(msg[content])[:1000]...[已截断]compressed.append(msg)returncompressedstaticmethoddefestimate_and_control_tokens(messages):Token 预估与控制totalsum(len(str(m.get(content,)))//4forminmessages)iftotal12000:# 触发压缩returnLLMOptimizer.prompt_compression(messages)returnmessages5. 架构层优化# 推荐的高并发架构 ┌──────────┐ │ Nginx │ (负载均衡 限流) └─────┬────┘ │ ┌───────────┼───────────┐ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │Agent服务 │ │Agent服务 │ │Agent服务 │ (水平扩展) │ 实例 1 │ │ 实例 2 │ │ 实例 3 │ └────┬────┘ └────┬────┘ └────┬────┘ │ │ │ └───────────┼───────────┘ │ ┌───────────┼───────────┐ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ Redis │ │ Milvus │ │ LLM API │ │ (缓存) │ │ (向量库)│ │ (多Key) │ └─────────┘ └─────────┘ └─────────┘ classScaledAgentService:水平扩展的 Agent 服务def__init__(self):# 多 LLM API Key 轮转突破 RPM 限制self.llm_keys[sk-key1,sk-key2,sk-key3]self.current_key_idx0# 限流器self.rate_limiterTokenBucket(rate100,capacity200)defget_llm_client(self):轮转 API Keykeyself.llm_keys[self.current_key_idx]self.current_key_idx(self.current_key_idx1)%len(self.llm_keys)returnLLMClient(api_keykey)asyncdefhandle_request(self,user_input,session_id):处理请求带限流ifnotself.rate_limiter.consume():return{error:服务繁忙请稍后重试,retry_after:5}returnawaitself.agent.chat(user_input,session_id)classTokenBucket:令牌桶限流器def__init__(self,rate:int,capacity:int):self.raterate self.capacitycapacity self.tokenscapacity self.last_refilltime.time()defconsume(self,tokens1):self._refill()ifself.tokenstokens:self.tokens-tokensreturnTruereturnFalsedef_refill(self):nowtime.time()elapsednow-self.last_refill self.tokensmin(self.capacity,self.tokenselapsed*self.rate)self.last_refillnow6. 完整高并发架构关键数值总结优化手段延迟降低并发提升精确缓存100% (命中时)极高语义缓存80-90%极高工具缓存20-40%中异步并行30-50%50%小模型分流60-80%100%流式输出感知 50%不变水平扩展不变N倍线性API Key 轮转不变N倍落地建议第一阶段立即可做: ✅ Redis 精确缓存 → 热点问题零延迟 ✅ 异步并行工具调用 → 减少 30% 延迟 ✅ 流式输出 → 用户感知更快 第二阶段一周内: ✅ 工具结果缓存 → 减少重复 API 调用 ✅ 语义缓存 → 覆盖相似问法 ✅ 小模型分流 → 降低 LLM 成本 第三阶段生产加固: ✅ 多 API Key 轮转 → 突破 RPM 限制 ✅ 水平扩展 负载均衡 ✅ Token 预算 Prompt 压缩