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

目录

1. Agent 服务的性能瓶颈
2. 缓存策略全览
3. 异步与并发优化
4. LLM 层优化
5. 架构层优化
6. 完整高并发架构

---

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, TTL=5min)
    ↓ 未命中
L2: 语义相似缓存 (向量库, TTL=30min)
    ↓ 未命中  
L3: 部分结果缓存 (工具调用结果, TTL=10min)
    ↓ 未命中
L4: 实际执行

2.2 L1: 精确匹配缓存

import hashlib
import redis
import json

class ExactMatchCache:
    """精确匹配缓存 —— 完全相同的输入直接返回缓存"""
    
    def __init__(self, redis_client, ttl=300):
        self.redis = redis_client
        self.ttl = ttl
    
    def _normalize(self, text: str) -> str:
        """标准化输入"""
        # 去除多余空格、统一大小写（中文不需）、去除标点变化
        text = re.sub(r'\s+', ' ', text.strip())
        return text
    
    def get(self, user_input: str) -> Optional[str]:
        """获取缓存"""
        key = f"cache:exact:{hashlib.md5(self._normalize(user_input).encode()).hexdigest()}"
        cached = self.redis.get(key)
        return json.loads(cached) if cached else None
    
    def set(self, user_input: str, response: str):
        """写入缓存"""
        key = f"cache:exact:{hashlib.md5(self._normalize(user_input).encode()).hexdigest()}"
        self.redis.setex(key, self.ttl, json.dumps(response, ensure_ascii=False))
    
    def invalidate(self, pattern: str = "*"):
        """批量失效缓存"""
        keys = self.redis.keys(f"cache:exact:{pattern}")
        if keys:
            self.redis.delete(*keys)

# 不缓存的场景
NO_CACHE_PATTERNS = [
    r"我的.*订单",       # 个性化查询
    r"ORD-\d{4}-\d{5}",  # 具体订单号
    r"查询.*积分|余额",   # 实时数据
]

2.3 L2: 语义相似缓存

class SemanticCache:
    """语义相似缓存 —— 意思相同的不同问法命中缓存"""
    
    def __init__(self, vector_store, redis_client, 
                 similarity_threshold=0.92, ttl=1800):
        self.vector_store = vector_store
        self.redis = redis_client
        self.threshold = similarity_threshold
        self.ttl = ttl
    
    async def get(self, user_input: str, session_id: str) -> Optional[str]:
        """语义检索缓存"""
        
        query_embedding = await self._get_embedding(user_input)
        
        # 搜索相似的已缓存问题
        results = self.vector_store.search(
            query_embedding,
            k=1,
            filter={"type": "semantic_cache"}
        )
        
        if results and results[0]["score"] > self.threshold:
            cache_key = results[0]["metadata"]["cache_key"]
            cached = self.redis.get(cache_key)
            if cached:
                # 更新访问时间
                self.redis.expire(cache_key, self.ttl)
                return json.loads(cached)
        
        return None
    
    async def set(self, user_input: str, response: str):
        """写入语义缓存"""
        
        embedding = await self._get_embedding(user_input)
        cache_key = f"cache:semantic:{hashlib.md5(user_input.encode()).hexdigest()}"
        
        # 存储响应
        self.redis.setex(cache_key, self.ttl, json.dumps(response, ensure_ascii=False))
        
        # 存储向量用于语义检索
        self.vector_store.add(
            id=cache_key,
            vector=embedding,
            metadata={
                "type": "semantic_cache",
                "cache_key": cache_key,
                "original_query": user_input[:200],
                "timestamp": time.time()
            }
        )
    
    async def _get_embedding(self, text):
        # 实际的 embedding 调用
        return await self.embedding_client.encode(text)

2.4 L3: 工具调用结果缓存

class ToolResultCache:
    """工具调用结果缓存 —— 同一工具+参数的结果缓存"""
    
    def __init__(self, redis_client, default_ttl=600):
        self.redis = redis_client
        self.default_ttl = default_ttl
        
        # 不同工具的缓存 TTL
        self.ttl_map = {
            "search_knowledge_base": 1800,  # 知识库30分钟
            "query_product": 300,            # 商品信息5分钟
            "query_inventory": 60,           # 库存1分钟
            "query_order": 0,                # 订单不缓存（实时变化）
            "query_user": 0,                 # 用户信息不缓存
        }
    
    def get(self, tool_name: str, args: dict) -> Optional[dict]:
        """获取工具结果缓存"""
        
        ttl = self.ttl_map.get(tool_name, -1)
        if ttl == 0:
            return None  # 该工具不缓存
        
        # 构建缓存 key
        cache_key = self._build_key(tool_name, args)
        
        cached = self.redis.get(cache_key)
        return json.loads(cached) if cached else None
    
    def set(self, tool_name: str, args: dict, result: dict):
        """缓存工具结果"""
        ttl = self.ttl_map.get(tool_name, self.default_ttl)
        if ttl <= 0:
            return
        
        cache_key = self._build_key(tool_name, args)
        self.redis.setex(
            cache_key,
            ttl,
            json.dumps(result, ensure_ascii=False)
        )
    
    def _build_key(self, tool_name: str, args: dict) -> str:
        """构建缓存 key"""
        args_str = json.dumps(args, sort_keys=True, ensure_ascii=False)
        args_hash = hashlib.md5(args_str.encode()).hexdigest()
        return f"tool_cache:{tool_name}:{args_hash}"
    
    def invalidate_tool(self, tool_name: str):
        """使某个工具的所有缓存失效"""
        keys = self.redis.keys(f"tool_cache:{tool_name}:*")
        if keys:
            self.redis.delete(*keys)

2.5 缓存集成到 Agent

class CachedAgent:
    """集成缓存的 Agent"""
    
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
        self.exact_cache = ExactMatchCache(redis_client)
        self.semantic_cache = SemanticCache(vector_store, redis_client)
        self.tool_cache = ToolResultCache(redis_client)
        
        self.stats = {"cache_hit": 0, "cache_miss": 0}
    
    async def run(self, user_input: str, session_id: str):
        # 1. 精确缓存
        exact = self.exact_cache.get(user_input)
        if exact:
            self.stats["cache_hit"] += 1
            return exact
        
        self.stats["cache_miss"] += 1
        
        # 2. 语义缓存
        semantic = await self.semantic_cache.get(user_input, session_id)
        if semantic:
            self.stats["cache_hit"] += 1
            return semantic
        
        # 3. 实际执行（带工具缓存）
        response = await self._execute_with_tool_cache(user_input)
        
        # 4. 写入缓存
        self.exact_cache.set(user_input, response)
        await self.semantic_cache.set(user_input, response)
        
        return response
    
    async def _execute_with_tool_cache(self, user_input):
        """带工具缓存的执行"""
        messages = [{"role": "user", "content": user_input}]
        
        for step in range(15):
            response = await self.llm.chat(messages, tools=self.tools)
            
            if response.finish_reason == "stop":
                return response.content
            
            for tc in response.tool_calls:
                # 检查工具缓存
                cached_result = self.tool_cache.get(tc.name, tc.args)
                
                if cached_result:
                    result = cached_result
                    logger.info(f"工具缓存命中: {tc.name}")
                else:
                    # 执行工具
                    result = await self._execute_tool(tc)
                    # 缓存结果
                    self.tool_cache.set(tc.name, tc.args, result)
                
                messages.append({
                    "role": "tool",
                    "content": json.dumps(result, ensure_ascii=False)
                })

3. 异步与并发优化

3.1 会话内工具并行

async def execute_parallel_tools(self, tool_calls: list) -> list:
    """并行执行互不依赖的工具调用"""
    
    # 检查工具之间有无依赖
    independent = self._filter_independent(tool_calls)
    
    if len(independent) > 1:
        # 并行执行
        tasks = [
            self._execute_tool(tc) for tc in independent
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        for tc, result in zip(independent, results):
            if isinstance(result, Exception):
                logger.error(f"工具 {tc.name} 执行失败: {result}")
        
        return results
    else:
        # 单个工具直接执行
        return [await self._execute_tool(tool_calls[0])]

def _filter_independent(self, tool_calls):
    """过滤出互不依赖的工具调用"""
    # 简单规则: 如果工具类别不同，就认为可并行
    # 实际可根据工具间数据依赖判断
    categories = [self.tools[tc.name].category for tc in tool_calls]
    if len(set(categories)) == len(categories):
        return tool_calls  # 全部可并行
    return tool_calls[:1]  # 保守：只并行第一个

3.2 跨会话异步处理

class AsyncAgentService:
    """异步 Agent 服务"""
    
    def __init__(self, max_concurrent=50):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.agent_pool = AgentPool(min_size=5, max_size=20)
    
    async def chat(self, user_input: str, session_id: str):
        """异步聊天接口"""
        
        async with self.semaphore:  # 并发控制
            agent = await self.agent_pool.acquire()
            try:
                response = await agent.run(user_input, session_id)
                return response
            finally:
                await self.agent_pool.release(agent)

class AgentPool:
    """Agent 实例池（复用 LLM 连接）"""
    
    def __init__(self, min_size=5, max_size=20):
        self.pool = asyncio.Queue()
        self.min_size = min_size
        self.max_size = max_size
        self.current_size = 0
    
    async def acquire(self):
        """获取 Agent 实例"""
        try:
            return self.pool.get_nowait()
        except asyncio.QueueEmpty:
            if self.current_size < self.max_size:
                self.current_size += 1
                return self._create_agent()
            return await self.pool.get()
    
    async def release(self, agent):
        """归还 Agent 实例"""
        await self.pool.put(agent)

4. LLM 层优化

class LLMOptimizer:
    """LLM 层优化"""
    
    @staticmethod
    def use_small_model_for_simple_tasks(user_input, intent):
        """简单任务用小模型，省钱+快"""
        
        if intent in ["chitchat", "simple_fact"]:
            return "gpt-4o-mini"   # 快 5x，便宜 20x
        elif intent in ["procedure", "faq"]:
            return "gpt-4o"        # 平衡
        else:
            return "gpt-4o"        # 复杂任务才用最好的
    
    @staticmethod
    def streaming_response(generator):
        """流式输出 —— 用户更快看到第一个字"""
        for chunk in generator:
            yield chunk.content
    
    @staticmethod
    def prompt_compression(messages, max_tokens=8000):
        """压缩 Prompt —— 移除冗余内容"""
        # 保留 system prompt + 最近 N 轮 + 工具结果摘要
        compressed = [messages[0]]  # system prompt
        
        # 最近的消息
        recent = messages[-10:]
        
        # 压缩过长的工具返回
        for msg in recent:
            if msg["role"] == "tool" and len(str(msg["content"])) > 1000:
                msg["content"] = str(msg["content"])[:1000] + "...[已截断]"
            compressed.append(msg)
        
        return compressed
    
    @staticmethod
    def estimate_and_control_tokens(messages):
        """Token 预估与控制"""
        total = sum(len(str(m.get("content", ""))) // 4 for m in messages)
        
        if total > 12000:
            # 触发压缩
            return LLMOptimizer.prompt_compression(messages)
        
        return messages

5. 架构层优化

# ===== 推荐的高并发架构 =====

"""
              ┌──────────┐
              │  Nginx    │  (负载均衡 + 限流)
              └─────┬────┘
                    │
        ┌───────────┼───────────┐
        ▼           ▼           ▼
   ┌─────────┐ ┌─────────┐ ┌─────────┐
   │Agent服务 │ │Agent服务 │ │Agent服务 │  (水平扩展)
   │ 实例 1  │ │ 实例 2  │ │ 实例 3  │
   └────┬────┘ └────┬────┘ └────┬────┘
        │           │           │
        └───────────┼───────────┘
                    │
        ┌───────────┼───────────┐
        ▼           ▼           ▼
   ┌─────────┐ ┌─────────┐ ┌─────────┐
   │ Redis   │ │ Milvus  │ │ LLM API │
   │ (缓存)  │ │ (向量库)│ │ (多Key) │
   └─────────┘ └─────────┘ └─────────┘
"""

class ScaledAgentService:
    """水平扩展的 Agent 服务"""
    
    def __init__(self):
        # 多 LLM API Key 轮转（突破 RPM 限制）
        self.llm_keys = [
            "sk-key1", "sk-key2", "sk-key3"
        ]
        self.current_key_idx = 0
        
        # 限流器
        self.rate_limiter = TokenBucket(rate=100, capacity=200)
    
    def get_llm_client(self):
        """轮转 API Key"""
        key = self.llm_keys[self.current_key_idx]
        self.current_key_idx = (self.current_key_idx + 1) % len(self.llm_keys)
        return LLMClient(api_key=key)
    
    async def handle_request(self, user_input, session_id):
        """处理请求（带限流）"""
        
        if not self.rate_limiter.consume():
            return {"error": "服务繁忙，请稍后重试", "retry_after": 5}
        
        return await self.agent.chat(user_input, session_id)

class TokenBucket:
    """令牌桶限流器"""
    
    def __init__(self, rate: int, capacity: int):
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last_refill = time.time()
    
    def consume(self, tokens=1):
        self._refill()
        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False
    
    def _refill(self):
        now = time.time()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.last_refill = now

6. 完整高并发架构

关键数值总结

优化手段	延迟降低	并发提升
精确缓存	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 压缩

---