Agent系统的性能优化：从Prompt压缩到工具调用的延迟削减策略

Agent系统的性能优化从Prompt压缩到工具调用的延迟削减策略一、Agent系统的延迟构成Agent系统的延迟可以分解为四个主要组成部分Prompt构建与传输延迟、LLM推理延迟、工具调用执行延迟、以及结果后处理延迟。对于典型的ReAct模式Agent单次交互通常需要3-5轮迭代——每一轮都包含完整的思考→行动→观察循环。多轮迭代的串行执行使得端到端延迟很容易超过10秒。理解延迟构成是优化的前提。在一个典型的RAGTool-Use Agent中各阶段延迟占比大致为LLM推理占50-60%工具调用占20-30%Prompt处理占10-15%后处理占5-10%。这种分布决定了优化工作的优先级——压缩LLM推理和工具调用延迟的ROI最高。graph LR A[用户输入] -- B[Prompt构建] B -- C[LLM推理] C -- D{需要工具调用?} D --|是| E[工具调用执行] E -- F[结果整合] F -- C D --|否| G[结果后处理] G -- H[返回用户] subgraph 优化策略 B -.- B1[压缩技术] C -.- C1[缓存命中] E -.- E1[并行执行] D -.- D1[提前退出] end style C fill:#E53935,color:#fff style E fill:#FF8F00,color:#fff style B fill:#3949AB,color:#fff二、Prompt压缩技术Prompt压缩是降低LLM推理延迟最直接的手段——推理延迟与输入token数大致呈线性关系减少Prompt长度就是减少推理时间。LLMLingua是微软提出的Prompt压缩框架核心原理是通过一个小型语言模型如LLaMA-2-7B评估每个token对任务的重要程度保留高重要性token移除低重要性token。在RAG场景中LLMLingua可以将检索到的上下文从2000 token压缩到500 token以内同时保持语义完整性。实测压缩比在3-5倍时对下游任务准确率的影响通常小于2%。选择性截断是更轻量但同样有效的策略。它的核心思想是不是所有上下文对当前推理步骤同等重要。对于长对话历史保留最近3轮对话system prompt通常足够。对于检索到的文档片段通过相似度排序只保留Top-K。对于工具的输出结果只保留关键字段而非完整返回。 Prompt压缩策略实现 - 选择性截断根据重要性对上下文排序并截断 - Token预算管理为不同context类型分配token配额 class PromptCompressor: Prompt压缩器 —— 在质量与速度之间的平衡 def __init__(self, max_total_tokens: int 4096): self.max_total_tokens max_total_tokens # Token预算分配策略 self.budget_allocation { system_prompt: 0.15, # 系统提示15% tools_schema: 0.20, # 工具定义20% conversation_history: 0.25, # 对话历史25% retrieved_context: 0.30, # 检索上下文30% user_query: 0.10, # 用户输入10% } def compress(self, prompt_components: dict) - str: 对Prompt各组件分配预算并压缩 compressed {} # 系统提示通常较短保留原始 compressed[system_prompt] prompt_components[system_prompt] # 工具定义只保留名称参数描述去除使用示例和长文档 compressed[tools_schema] self._trim_tool_definitions( prompt_components[tools_schema], budgetint(self.max_total_tokens * self.budget_allocation[tools_schema]) ) # 对话历史保留最近N轮 compressed[conversation_history] self._truncate_conversation( prompt_components[conversation_history], budgetint(self.max_total_tokens * self.budget_allocation[conversation_history]) ) # 检索上下文按相似度排序后截断 compressed[retrieved_context] self._rank_and_truncate( prompt_components[retrieved_context], budgetint(self.max_total_tokens * self.budget_allocation[retrieved_context]) ) # 用户查询直接保留 compressed[user_query] prompt_components[user_query] return self._assemble(compressed) def _truncate_conversation(self, messages: list, budget: int) - list: 截断对话历史保留最近轮次但确保总token在预算内 if not messages: return [] truncated [] token_count 0 # 从最新消息向前遍历 for msg in reversed(messages): msg_tokens self._estimate_tokens(msg[content]) if token_count msg_tokens budget: # 检查是否至少保留了1轮用户-助手对 if len(truncated) 2: break # 不够1轮强制保留该消息 truncated.insert(0, msg) token_count msg_tokens return truncated def _rank_and_truncate(self, documents: list, budget: int) - list: 按相关性排序后截断文档列表 if not documents: return [] # 按相似度分数降序排列 ranked sorted(documents, keylambda d: d.get(relevance_score, 0), reverseTrue) selected [] token_count 0 for doc in ranked: content doc.get(content, ) doc_tokens self._estimate_tokens(content) if token_count doc_tokens budget: # 最后一个文档部分截断 remaining budget - token_count if remaining 50: # 至少保留50个有意义的token doc[content] self._smart_truncate(content, remaining) selected.append(doc) break selected.append(doc) token_count doc_tokens return selected三、工具调用的并行化执行ReAct循环的传统实现是严格串行的思考→调用工具→等待结果→再思考→调用下一个工具。但很多工具调用之间并不存在数据依赖——例如同时查询天气API和新闻API两个调用完全可以并行执行。识别并行化机会的关键是依赖分析。在执行每个ReAct步骤之前分析LLM输出的工具调用列表检查每个调用的参数是否引用了其他调用的输出。没有引用关系的调用可以安全地并行执行有引用关系的必须串行等待被引用调用完成后传递结果。 工具调用并行化引擎 核心思路分析工具调用间的依赖关系 构建DAG有向无环图并行执行无依赖的调用 import asyncio from typing import List, Dict, Set class ParallelToolExecutor: 并行工具调用执行器 def __init__(self, max_parallel: int 5, timeout_seconds: float 30.0): self.max_parallel max_parallel self.timeout timeout_seconds async def execute(self, tool_calls: List[ToolCall]) - Dict[str, ToolResult]: 主入口分析依赖 → 并行执行 → 返回结果 # 步骤1构建依赖图 dep_graph self._build_dependency_graph(tool_calls) # 步骤2拓扑排序 → 确定执行层级 execution_levels self._topological_sort(dep_graph, tool_calls) # 步骤3按层级并行执行 results: Dict[str, ToolResult] {} for level in execution_levels: # 同一层级的所有调用无相互依赖可以并行 if len(level) 1: # 单调用直接执行 call level[0] results[call.id] await self._execute_single(call) else: # 多调用并行执行用信号量控制并发数 semaphore asyncio.Semaphore(self.max_parallel) async def bounded_execute(call: ToolCall): async with semaphore: return call.id, await self._execute_single(call) level_results await asyncio.gather( *[bounded_execute(call) for call in level], return_exceptionsTrue ) for result in level_results: if isinstance(result, Exception): call_id unknown results[call_id] ToolResult( errorstr(result), successFalse) else: call_id, tool_result result results[call_id] tool_result return results def _build_dependency_graph( self, tool_calls: List[ToolCall]) - Dict[str, Set[str]]: 构建依赖图 规则如果工具B的参数引用了工具A的输出通过$A.id语法 则B依赖A graph: Dict[str, Set[str]] {} # 收集所有工具的输出标识符 output_ids {tc.id for tc in tool_calls} for tc in tool_calls: graph[tc.id] set() for param_value in tc.arguments.values(): if isinstance(param_value, str): # 检查参数是否引用了其他工具的输出 for ref_id in output_ids: if ref_id ! tc.id and f${ref_id} in param_value: graph[tc.id].add(ref_id) return graph def _topological_sort( self, dep_graph: Dict[str, Set[str]], tool_calls: List[ToolCall] ) - List[List[ToolCall]]: 拓扑排序 → 分为多个执行层级 id_to_call {tc.id: tc for tc in tool_calls} in_degree {tc.id: len(deps) for tc, deps in dep_graph.items()} levels [] while in_degree: # 当前入度为0的所有节点不依赖任何未执行的调用 current_level_ids [ tid for tid, deg in in_degree.items() if deg 0 ] if not current_level_ids: # 存在循环依赖 —— 按定义顺序串行执行 remaining list(in_degree.keys()) current_level_ids [remaining[0]] current_level [id_to_call[tid] for tid in current_level_ids] levels.append(current_level) # 移出当前层级节点更新后继节点的入度 for tid in current_level_ids: del in_degree[tid] for tid, deps in dep_graph.items(): if tid in in_degree: in_degree[tid] - len( set(current_level_ids) deps) return levels async def _execute_single(self, call: ToolCall) - ToolResult: 执行单个工具调用带超时控制 try: return await asyncio.wait_for( call.execute(), timeoutself.timeout) except asyncio.TimeoutError: return ToolResult( errorfTool execution timed out after {self.timeout}s, successFalse)四、上下文缓存与提前退出上下文缓存Context Caching是降低LLM推理延迟的另一个关键手段。多个连续请求共享相同的System Prompt和工具定义时这些固定部分的KV Cache可以被复用。Google的Gemini和Anthropic的Claude都已原生支持上下文缓存——将缓存的Prefix部分保存在服务端后续请求只需处理增量部分。在工程实现上缓存策略分两层第一层是请求级缓存——通过请求的Hash指纹判断是否与之前的请求共享相同的Prefix第二层是Token级缓存——利用推理引擎的Prefix Caching能力如vLLM的Automatic Prefix Caching自动识别并复用已计算的KV Cache。提前退出Early Exit策略针对的是在ReAct循环中LLM可能在中间步骤就确定了最终答案不需要继续后面的工具调用。通过分析LLM输出的置信度logprobs和工具调用的必要性可以在满足条件时直接返回结果跳过不必要的后续迭代。 ReAct循环的提前退出与缓存策略 核心优化 1. 当LLM在思考中表明已获得足够信息时跳过后续工具调用 2. System Prompt和工具定义的KV Cache在多轮中复用 class OptimizedReActLoop: 优化版ReAct循环 —— 提前退出 缓存复用 def __init__(self, llm_client, tool_executor, max_iterations: int 10): self.llm llm_client self.tool_executor tool_executor self.max_iterations max_iterations # 缓存system prompt 工具定义 的 cache key self._prefix_cache_id: Optional[str] None async def run(self, user_query: str) - AgentResponse: 执行ReAct循环带提前退出优化 messages self._build_initial_messages(user_query) # 尝试从缓存加载固定的Prefix KV Cache cache_id self._get_or_create_prefix_cache() for iteration in range(self.max_iterations): # LLM推理 —— 如果有缓存则指定cache_id response await self.llm.chat( messagesmessages, prefix_cache_idcache_id, temperature0.1, logprobsTrue # 需要logprobs用于置信度判断 ) # 检查是否可以直接返回最终答案 if self._has_final_answer(response): return AgentResponse( answerresponse.content, iterationsiteration 1, early_exitTrue ) # 提取工具调用 tool_calls self._extract_tool_calls(response) if not tool_calls: # 没有工具调用 → 视作最终答案 return AgentResponse( answerresponse.content, iterationsiteration 1 ) # 并行执行工具调用 tool_results await self.tool_executor.execute(tool_calls) # 将工具结果追加到对话历史 messages self._append_tool_results( messages, response, tool_results) # 检查提前退出条件 if self._should_early_exit(response, tool_results): # 工具已返回足够信息生成最终答案 final_response await self.llm.chat( messagesmessages [{ role: user, content: 基于以上信息请直接给出最终答案。 }], prefix_cache_idcache_id ) return AgentResponse( answerfinal_response.content, iterationsiteration 1, early_exitTrue ) # 达到最大迭代次数 return AgentResponse( answerresponse.content, iterationsself.max_iterations, max_iterations_reachedTrue ) def _has_final_answer(self, response) - bool: 判断LLM是否已给出最终答案不再需要工具调用 content response.content.strip().lower() # 显式终止标记 final_markers [final answer:, 最终答案, answer:] if any(marker in content for marker in final_markers): return True # 没有 Action/工具调用 → 可能已是答案 tool_call_markers [action:, tool_call:, 使用工具] if not any(marker in content for marker in tool_call_markers): return True return False def _should_early_exit( self, response, tool_results: dict) - bool: 判断是否应触发提前退出 # 条件1所有工具调用都成功 结果置信度高 all_success all( r.success for r in tool_results.values()) # 条件2工具结果中已包含明确的最终答案 has_definitive_info any( final in str(r.data).lower() or answer in str(r.data).lower() for r in tool_results.values() ) return all_success and has_definitive_info五、总结Agent系统的性能优化需要从延迟构成的全局视角出发在Prompt压缩、推理加速、工具并行化和缓存复用四个方向同时发力。Prompt压缩减少输入token数效果直接体现在TTFT上工具并行化减少等待轮次线性加速3个并行工具调用节省60%以上的工具等待时间上下文缓存减少重复计算Prefix命中可将TTFT降至接近零提前退出避免不必要的迭代尤其是在简单查询场景下可以节省2-4轮额外迭代。这些优化手段的组合效果是相乘而非相加的。一个经过Prompt压缩节省30% token 工具并行化节省50%工具等待 上下文缓存节省40%推理时间的Agent系统端到端延迟可以从12秒降到3-4秒。但需要注意的是每一种优化都带来了一定程度的工程复杂度——在一个简单场景下全量实施可能得不偿失。从Profile数据出发优先优化延迟占比最高的环节才是最务实的策略。