Agent 工具调用的极致性能优化：并行编排、语义缓存与智能预取策略

一、Agent 的响应延迟，90% 耗在了工具调用上

在典型的企业级 Agent 架构中，LLM 推理的延迟相对稳定。真正让用户体验崩溃的，是工具调用环节的延迟累积。

每一次工具调用意味着一次外部 API 请求。查询数据库、调用搜索引擎、读取文件系统、请求第三方服务。一个复杂任务可能触发 5-10 次工具调用。如果串行执行，总延迟 = 各次调用延迟之和。一个 500ms 的 API，10 次调用就是 5 秒——用户早已失去耐心。

更糟糕的是，很多工具调用之间存在独立性。查询用户信息、搜索相关文档、检查权限状态，这三个操作互不依赖，却默认被串行执行。优化工具调用的核心思路只有三个方向：让能并行的并行，让能复用的复用，让可预测的预取。

行业案例：某客服Agent产品上线时工具调用全串行。一个典型对话需调用6个外部API——客户信息查询、订单查询、知识库检索、情绪分析、标签匹配、推荐生成。每步平均400ms，总延迟2.4秒，用户感知为"反应迟钝"。切换到并行编排后，4个无依赖调用并发执行，总延迟降至800ms。用户感知响应速度提升3倍。关键改动仅两处：为每个工具标注依赖关系，调度器按拓扑分层后同层并行。代码改动不到200行，体验提升远超预期。

二、三级加速模型：并行执行、语义缓存与智能预取

该模型的核心流程始于 Agent 接收任务后的工具调用依赖分析。根据分析结果，系统会动态选择三种加速策略之一：对于无依赖的工具调用，直接进入第一级并行执行；对于存在依赖或重复请求的场景，触发第二级语义缓存匹配；而对于可预测的行为模式，则启动第三级智能预取。所有路径的结果最终汇聚至 LLM 推理环节，生成最终响应。

第一级 —— 并行执行：对无数据依赖的工具调用并发执行。总延迟从"累加"变为"取最大值"。这是收益最高、实现成本最低的优化手段。

第二级 —— 语义缓存：不基于精确匹配，而是基于语义相似度匹配。用户问"上周销量"和"7 天前的销售数据"，含义相同但文字不同。语义缓存能命中这类变体请求。

第三级 —— 智能预取：基于用户行为模式预测下一步可能要用的工具，在 Agent 推理期间后台预加载结果。当 LLM 决定要调用该工具时，结果已经在内存中。

三、生产级实现：带语义缓存的工具编排器

"""
---
Agent 工具调用加速器

实现三级加速：
1. 按依赖关系并行调度工具
2. 基于向量相似度的语义缓存
3. 合并关联查询减少调用次数
"""
import asyncio
import hashlib
import time
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Set, Tuple
from enum import Enum

import numpy as np

@dataclass
class ToolDefinition:
    """工具定义"""
    name: str
    description: str
    depends_on: List[str] = field(default_factory=list)
    cache_ttl_seconds: int = 300
    
    async def execute(self, params: dict) -> dict:
        """子类实现具体执行逻辑"""
        raise NotImplementedError

@dataclass
class CacheEntry:
    """缓存条目——存储工具结果和语义向量"""
    result: dict
    embedding: List[float]
    params_hash: str
    timestamp: float = field(default_factory=time.time)
    
    @property
    def expired(self) -> bool:
        return False  # TTL 检查由调用方处理


class SemanticCache:
    """基于余弦相似度的语义缓存。
    
    不依赖精确字符串匹配，而是比较请求的语义向量。
    适合处理用户问法多变但意图相同的场景。
    """
    
    def __init__(self, similarity_threshold: float = 0.92):
        self._store: Dict[str, List[CacheEntry]] = {}
        self._threshold = similarity_threshold
    
    def get(self, tool_name: str, query_embedding: List[float],
            ttl: int = 300) -> Optional[dict]:
        """查找语义相似的缓存结果"""
        entries = self._store.get(tool_name, [])
        now = time.time()
        
        best_score = 0
        best_entry = None
        
        for entry in entries:
            if now - entry.timestamp > ttl:
                continue
            
            # 余弦相似度计算
            score = self._cosine_similarity(
                np.array(query_embedding),
                np.array(entry.embedding),
            )
            if score > best_score:
                best_score = score
                best_entry = entry
        
        if best_entry and best_score >= self._threshold:
            return best_entry.result
        return None
    
    def set(self, tool_name: str, query_embedding: List[float],
            params_hash: str, result: dict):
        """写入缓存"""
        if tool_name not in self._store:
            self._store[tool_name] = []
        
        self._store[tool_name].append(CacheEntry(
            result=result,
            embedding=query_embedding,
            params_hash=params_hash,
        ))
        
        # LRU 淘汰：每工具最多保留 100 条
        if len(self._store[tool_name]) > 100:
            self._store[tool_name] = self._store[tool_name][-100:]
    
    @staticmethod
    def _cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
        dot = np.dot(a, b)
        norm = np.linalg.norm(a) * np.linalg.norm(b)
        return float(dot / norm) if norm > 0 else 0


class ToolOrchestrator:
    """工具编排器——并行调度 + 缓存 + 批处理。
    
    核心设计：
    1. 拓扑排序识别依赖关系
    2. 同层无依赖工具并行执行
    3. 有依赖的工具按序执行
    4. 语义缓存减少重复调用
    """
    
    def __init__(self, cache: SemanticCache):
        self.cache = cache
        self._tools: Dict[str, ToolDefinition] = {}
        # 嵌入模型（简化版，实际应使用专门的 Embedding 服务）
        self._embedding_cache: Dict[str, List[float]] = {}
    
    def register(self, tool: ToolDefinition):
        self._tools[tool.name] = tool
    
    async def execute_plan(self, tool_calls: List[dict]) -> Dict[str, dict]:
        """按依赖关系并行执行工具调用计划。
        
        步骤：
        1. 构建依赖图
        2. 拓扑排序分组
        3. 同组内并行执行
        4. 有缓存时跳过
        """
        # 构建依赖图
        dep_graph: Dict[str, Set[str]] = {}
        for call in tool_calls:
            name = call["name"]
            tool = self._tools.get(name)
            if tool:
                dep_graph[name] = set(tool.depends_on)
        
        # 拓扑分层
        layers = self._topological_layers(dep_graph)
        
        results: Dict[str, dict] = {}
        
        for layer in layers:
            # 同层工具并行执行
            tasks = []
            for tool_name in layer:
                call = next(c for c in tool_calls if c["name"] == tool_name)
                task = self._execute_with_cache(tool_name, call.get("params", {}))
                tasks.append(task)
            
            layer_results = await asyncio.gather(*tasks, return_exceptions=True)
            
            for tool_name, result in zip(layer, layer_results):
                if isinstance(result, Exception):
                    results[tool_name] = {"error": str(result)}
                else:
                    results[tool_name] = result
        
        return results
    
    async def _execute_with_cache(self, tool_name: str, 
                                   params: dict) -> dict:
        """带缓存的工具执行"""
        tool = self._tools.get(tool_name)
        if not tool:
            return {"error": f"未知工具: {tool_name}"}
        
        # 生成参数签名用于精确匹配
        params_str = str(sorted(params.items()))
        params_hash = hashlib.md5(params_str.encode()).hexdigest()
        
        # 语义缓存查找
        embedding = await self._get_embedding(params_str)
        cached = self.cache.get(tool_name, embedding, tool.cache_ttl_seconds)
        if cached is not None:
            return {"cached": True, **cached}
        
        # 执行工具调用
        result = await tool.execute(params)
        
        # 写入缓存
        self.cache.set(tool_name, embedding, params_hash, result)
        
        return result
    
    async def _get_embedding(self, text: str) -> List[float]:
        """获取文本的向量嵌入"""
        if text in self._embedding_cache:
            return self._embedding_cache[text]
        
        # 简化实现：基于字符的轻量向量
        # 生产环境应替换为真实的 Embedding 模型 API 调用
        embedding = [float(ord(c) % 100) / 100 for c in text[:128]]
        # 填充到固定维度
        while len(embedding) < 128:
            embedding.append(0.0)
        
        self._embedding_cache[text] = embedding[:128]
        return self._embedding_cache[text]
    
    def _topological_layers(self, 
                             graph: Dict[str, Set[str]]) -> List[List[str]]:
        """拓扑排序并分层——同层节点可并行执行"""
        in_degree = {n: 0 for n in graph}
        for deps in graph.values():
            for dep in deps:
                if dep in in_degree:
                    in_degree[dep] += 1
        
        layers = []
        remaining = set(graph.keys())
        
        while remaining:
            # 找出入度为 0 的节点（当前层）
            current_layer = [
                n for n in remaining if in_degree.get(n, 0) == 0
            ]
            if not current_layer:
                # 存在循环依赖，取第一个剩余节点
                current_layer = [next(iter(remaining))]
            
            layers.append(current_layer)
            remaining -= set(current_layer)
            
            # 更新入度
            for node in current_layer:
                for dep in graph.get(node, set()):
                    if dep in in_degree:
                        in_degree[dep] -= 1
        
        return layers

四、缓存与预取的风险边界

语义缓存的精度陷阱：语义相似不等于语义相同。"昨天销售额"和"前天销售额"语义极相似，但答案完全不同。缓存键必须包含时间等与答案强相关的参数。不能仅依赖向量相似度，需要混合精确匹配和语义匹配的双层策略。

预取的成本浪费：预测模型不可能 100% 准确。如果预取命中率低于 50%，浪费的计算资源反而拖累系统。建议只在 RTF（Real-Time Factor）敏感的场景启用预取，且限制预取并发数。

不适合缓存的场景：

• 实时数据查询（股票价格、库存数量）——缓存意味着数据过期
• 写操作（创建订单、发送消息）——缓存副作用是灾难性的
• 参数空间爆炸的工具——缓存命中率趋近于零

五、总结

Agent 工具调用的性能优化，本质是一场"延迟消除"的工程实践。三大核心手段各有适用的场景和边界。

优化落地路线：

1. 先做并行执行——收益最大，实现最简单，几乎无风险
2. 再做结果缓存——对高频读操作收益显著，需谨慎处理过期策略
3. 最后考虑预取——只在延迟敏感且模式可预测时启用
4. 依赖分析是并行的前提——拓扑分层决定能并发多少
5. 监控缓存命中率和平均延迟——没有度量就没有优化

要点提炼：

1. 并行执行ROI最高：无依赖调用同层并发，延迟从累加变取最大值。
2. 语义缓存必须混合精确匹配与时间参数，纯向量相似度可能误判同类但不同时间的问题。
3. 预取仅适用于命中率＞50%的场景，低于此阈值计算资源浪费抵消收益。
4. 按序推进：先并行→再缓存→最后预取，每步验证收益再启动下一步。