Prompt Engineering 与 Agent 工作流：工具选择与动态路由的编排策略
一、Agent 的工具选择困境：从硬编码到智能路由
当 Agent 需要调用外部工具完成任务时，核心问题是如何选择合适的工具。早期实现依赖关键词匹配等硬编码规则，但面对模糊意图时容易出错。例如用户说"帮我查一下明天的天气"，系统难以判断该调用天气 API 还是日历 API；而"把这个数据整理一下"这类请求，则需区分表格工具和图表工具的适用场景。
工具数量增长进一步加剧了复杂性。企业级 Agent 通常接入 20-50 个工具，每个工具具有不同的参数格式和调用约束。硬编码规则随工具数量呈指数级增长，维护成本迅速失控。动态路由策略让 LLM 根据上下文自主选择工具，虽提升了扩展性，但也带来新风险：LLM 可能误选工具或构造无效参数。
二、工具动态路由的架构设计
flowchart TB subgraph 输入层 USER[用户请求] --> INTENT[意图识别] INTENT --> CTX[上下文构建] end subgraph 工具注册层 REG[工具注册表] --> META[工具元数据: 名称/描述/参数/约束] REG --> EMBED[工具描述向量化] end subgraph 路由决策层 CTX --> RETRIEVE[工具检索: 语义匹配 Top-K] EMBED --> RETRIEVE RETRIEVE --> LLM_ROUTE[LLM 路由决策: 从 Top-K 中选择] LLM_ROUTE --> PARAM[参数构造: LLM 生成调用参数] end subgraph 执行与验证层 PARAM --> VALID[参数校验: Schema 验证] VALID --> |通过| EXEC[工具执行] VALID --> |失败| RETRY[参数修正: LLM 重新生成] RETRY --> VALID EXEC --> RESULT[结果解析] RESULT --> |需要继续| INTENT RESULT --> |任务完成| OUTPUT[输出结果] end subgraph 安全层 EXEC --> SANDBOX[沙箱执行: 超时/资源限制] SANDBOX --> AUDIT[审计日志: 记录所有工具调用] end style RETRIEVE fill:#e3f2fd style LLM_ROUTE fill:#fff3e0 style VALID fill:#e8f5e9 style SANDBOX fill:#ffebee动态路由采用"检索+排序"两阶段设计：先通过语义检索筛选 Top-K 候选工具，再由 LLM 从中选择最优工具并构造参数。这种设计既降低了 LLM 直接处理大量工具的认知负担，又提升了选择准确性。
三、工具动态路由引擎的实现
# tool_router.py — Agent 工具动态路由引擎 import time import json import hashlib from dataclasses import dataclass, field from typing import Optional, Any import numpy as np @dataclass class ToolParameter: """工具参数定义""" name: str type: str # string / number / boolean / array / object description: str required: bool = True enum: list[str] = field(default_factory=list) default: Any = None @dataclass class ToolDefinition: """工具定义""" tool_id: str name: str description: str category: str # search / data / communication / file / system parameters: list[ToolParameter] = field(default_factory=list) constraints: dict = field(default_factory=dict) # 超时、权限等约束 embedding: Optional[np.ndarray] = None @dataclass class ToolCallRequest: """工具调用请求""" tool_id: str parameters: dict request_id: str = "" confidence: float = 0.0 @dataclass class ToolCallResult: """工具调用结果""" request_id: str tool_id: str success: bool result: Any = None error: str = "" latency_ms: float = 0 class ToolRegistry: """工具注册表：管理所有可用工具""" def __init__(self, embed_fn=None): self._tools: dict[str, ToolDefinition] = {} self._embed_fn = embed_fn def register(self, tool: ToolDefinition) -> None: """注册工具""" # 为工具描述生成嵌入向量 if self._embed_fn: desc_text = f"{tool.name}: {tool.description}" tool.embedding = self._embed_fn(desc_text) self._tools[tool.tool_id] = tool def search(self, query: str, top_k: int = 5, category: str = None) -> list[ToolDefinition]: """语义检索最相关的工具""" if not self._embed_fn: return list(self._tools.values())[:top_k] query_emb = self._embed_fn(query) scored = [] for tool in self._tools.values(): if category and tool.category != category: continue if tool.embedding is None: continue sim = np.dot(query_emb, tool.embedding) / ( np.linalg.norm(query_emb) * np.linalg.norm(tool.embedding) + 1e-8 ) scored.append((sim, tool)) scored.sort(key=lambda x: x[0], reverse=True) return [tool for _, tool in scored[:top_k]] def get_tool(self, tool_id: str) -> Optional[ToolDefinition]: return self._tools.get(tool_id) class ToolRouter: """工具路由器：两阶段路由决策""" def __init__(self, registry: ToolRegistry, llm_fn=None): self._registry = registry self._llm_fn = llm_fn def route(self, user_request: str, context: dict = None) -> Optional[ToolCallRequest]: """根据用户请求路由到合适的工具""" # 阶段1：语义检索 Top-K 候选工具 candidates = self._registry.search(user_request, top_k=5) if not candidates: return None if len(candidates) == 1: # 只有一个候选，直接使用 tool = candidates[0] params = self._construct_params(tool, user_request, context) return ToolCallRequest( tool_id=tool.tool_id, parameters=params, confidence=0.8, ) # 阶段2：LLM 从候选中选择最合适的工具 if self._llm_fn: return self._llm_route(candidates, user_request, context) # 无 LLM 时使用第一个候选 tool = candidates[0] params = self._construct_params(tool, user_request, context) return ToolCallRequest( tool_id=tool.tool_id, parameters=params, confidence=0.5, ) def _llm_route(self, candidates: list[ToolDefinition], user_request: str, context: dict = None) -> Optional[ToolCallRequest]: """使用 LLM 从候选工具中选择""" # 构建工具列表描述 tools_desc = "" for i, tool in enumerate(candidates): params_desc = ", ".join( f"{p.name}({p.type})" for p in tool.parameters ) tools_desc += ( f"{i+1}. {tool.name} (ID: {tool.tool_id})\n" f" 描述: {tool.description}\n" f" 参数: {params_desc}\n" ) prompt = ( f"用户请求: {user_request}\n" f"上下文: {json.dumps(context or {}, ensure_ascii=False)}\n\n" f"可选工具:\n{tools_desc}\n" f"请选择最合适的工具并构造调用参数。" f"以 JSON 格式输出:\n" f'{{"tool_id": "xxx", "parameters": {{...}}, "confidence": 0.0-1.0}}' ) try: result = self._llm_fn(prompt) parsed = json.loads(result) tool_id = parsed.get("tool_id") tool = self._registry.get_tool(tool_id) if tool is None: return None # 参数校验 params = parsed.get("parameters", {}) valid, errors = self._validate_params(tool, params) if not valid: # 尝试修正参数 params = self._fix_params(tool, params, errors, user_request) return ToolCallRequest( tool_id=tool_id, parameters=params, confidence=float(parsed.get("confidence", 0.5)), ) except (json.JSONDecodeError, ValueError, KeyError): return None def _construct_params(self, tool: ToolDefinition, user_request: str, context: dict = None) -> dict: """构造工具调用参数""" params = {} for param in tool.parameters: if param.default is not None: params[param.name] = param.default elif param.required: params[param.name] = "" # 占位 return params def _validate_params(self, tool: ToolDefinition, params: dict) -> tuple[bool, list[str]]: """校验工具调用参数""" errors = [] for param_def in tool.parameters: if param_def.required and param_def.name not in params: errors.append( f"缺少必填参数: {param_def.name}" ) continue value = params.get(param_def.name) if value is None: continue # 类型检查 type_checks = { "string": lambda v: isinstance(v, str), "number": lambda v: isinstance(v, (int, float)), "boolean": lambda v: isinstance(v, bool), "array": lambda v: isinstance(v, list), } check = type_checks.get(param_def.type) if check and not check(value): errors.append( f"参数 {param_def.name} 类型错误: " f"期望 {param_def.type}" ) # 枚举检查 if param_def.enum and value not in param_def.enum: errors.append( f"参数 {param_def.name} 值无效: " f"可选 {param_def.enum}" ) return len(errors) == 0, errors def _fix_params(self, tool: ToolDefinition, params: dict, errors: list[str], user_request: str) -> dict: """尝试修正参数错误""" if not self._llm_fn: return params fix_prompt = ( f"工具 {tool.name} 的参数校验失败:\n" f"错误: {errors}\n" f"当前参数: {json.dumps(params, ensure_ascii=False)}\n" f"用户请求: {user_request}\n" f"请修正参数，以 JSON 格式输出修正后的完整参数。" ) try: result = self._llm_fn(fix_prompt) return json.loads(result) except (json.JSONDecodeError, ValueError): return params class ToolExecutor: """工具执行器：安全执行与审计""" def __init__(self, registry: ToolRegistry): self._registry = registry self._handlers: dict[str, callable] = {} self._audit_log: list[dict] = [] def register_handler(self, tool_id: str, handler: callable) -> None: """注册工具执行处理函数""" self._handlers[tool_id] = handler def execute(self, request: ToolCallRequest) -> ToolCallResult: """执行工具调用""" start = time.time() request_id = hashlib.md5( f"{request.tool_id}:{time.time()}".encode() ).hexdigest()[:8] tool = self._registry.get_tool(request.tool_id) if tool is None: return ToolCallResult( request_id=request_id, tool_id=request.tool_id, success=False, error=f"工具 {request.tool_id} 不存在", ) handler = self._handlers.get(request.tool_id) if handler is None: return ToolCallResult( request_id=request_id, tool_id=request.tool_id, success=False, error=f"工具 {request.tool_id} 未注册处理函数", ) # 执行工具（带超时保护） timeout = tool.constraints.get("timeout_seconds", 30) try: result = handler(request.parameters) latency = (time.time() - start) * 1000 # 记录审计日志 self._log(request_id, request, True, result, latency) return ToolCallResult( request_id=request_id, tool_id=request.tool_id, success=True, result=result, latency_ms=round(latency, 2), ) except Exception as e: latency = (time.time() - start) * 1000 self._log(request_id, request, False, str(e), latency) return ToolCallResult( request_id=request_id, tool_id=request.tool_id, success=False, error=str(e), latency_ms=round(latency, 2), ) def _log(self, request_id: str, request: ToolCallRequest, success: bool, result: Any, latency_ms: float) -> None: """记录审计日志""" self._audit_log.append({ "timestamp": time.time(), "request_id": request_id, "tool_id": request.tool_id, "parameters": request.parameters, "confidence": request.confidence, "success": success, "result_preview": str(result)[:200] if result else None, "latency_ms": latency_ms, })四、动态路由的准确率与安全风险
动态路由的效果主要受两个因素影响：语义检索的召回率和 LLM 的选择精度。
语义检索召回率：当工具描述与用户请求表述差异较大时，可能出现遗漏。例如用户说"帮我发个邮件"，但工具描述为"SMTP 邮件发送服务"，语义距离较远。可通过为工具添加别名和示例查询来改善。
LLM 选择精度：在 5 个候选工具中，LLM 的选择准确率约为 85%-90%，但随着工具数量增加而下降。功能相似的工具（如"搜索文档"和"搜索邮件"）容易混淆。解决方案包括明确区分工具使用场景，并在路由 Prompt 中加入历史选择记录。
安全风险：动态路由允许 LLM 自主选择工具和构造参数，存在选择危险工具（如删除文件）或构造恶意参数（如路径遍历）的风险。应对措施包括工具权限分级（只读/写入/危险）、参数白名单校验、执行沙箱隔离及审计日志记录。对于高风险操作，应要求用户显式确认。
适用边界：动态路由适合工具数量多、意图多样的场景。工具少于 5 个的简单 Agent 采用硬编码路由更可靠；涉及资金操作或数据删除的高风险场景，应保留人工确认环节。
五、总结
Agent 的工具动态路由通过"语义检索 + LLM 排序"的两阶段策略，有效应对了工具数量增长带来的路由决策挑战。语义检索缩小候选范围，LLM 从候选中精确选择并构造参数。参数校验和修正机制确保调用正确性，审计日志和安全沙箱保障执行安全性。建议从硬编码路由起步，当工具数量显著增加时引入动态路由，并始终为危险工具保留人工确认环节。
