工程实践指南 | 从工具定义到调用链路的全栈优化
—## 工具调用：Agent能力的乘法器一个没有工具的LLM只能"说"，一个有了工具的LLM才能"做"。工具调用（Function Calling/Tool Use）是让AI Agent真正发挥价值的核心机制。但在工程实践中，工具调用的问题比想象中多：- 模型调用了错误的工具- 工具参数提取不完整或格式错误- 工具执行失败但Agent不知道怎么处理- 并行工具调用导致竞争条件- 工具调用链路过长，成本失控本文从工具定义、参数设计、错误处理、到并行优化，系统梳理工具调用的工程最佳实践。—## 工具定义的关键原则### 原则一：名称即意图，描述即用法python# 错误示范：模糊的工具定义bad_tool = { "name": "search", "description": "搜索信息", "parameters": { "type": "object", "properties": { "query": {"type": "string"} } }}# 正确示范：清晰的工具定义good_tool = { "name": "search_knowledge_base", "description": """从内部知识库检索相关文档。 使用场景： - 当需要查找产品功能说明时 - 当需要了解公司政策时 - 当用户问题与内部文档相关时 不适合用于： - 实时信息查询（用search_web代替） - 用户个人数据查询（用get_user_data代替） """, "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "搜索查询词，应该是自然语言描述，如'产品退款政策'" }, "top_k": { "type": "integer", "description": "返回最相关的文档数量，默认3，最大10", "default": 3, "minimum": 1, "maximum": 10 }, "category": { "type": "string", "description": "可选的文档分类过滤", "enum": ["product", "policy", "technical", "faq"] } }, "required": ["query"] }}### 原则二：一个工具做一件事python# 错误：功能过于宽泛的工具bad_db_tool = { "name": "database_operation", "description": "执行数据库操作，支持CRUD", "parameters": { "type": "object", "properties": { "operation": {"type": "string", "enum": ["create", "read", "update", "delete"]}, "table": {"type": "string"}, "data": {"type": "object"} } }}# 正确：每个操作独立的工具tools = [ { "name": "get_user_by_id", "description": "通过用户ID获取用户信息", "parameters": { "type": "object", "properties": { "user_id": {"type": "string", "description": "用户的唯一ID"} }, "required": ["user_id"] } }, { "name": "update_user_profile", "description": "更新用户的个人资料字段", "parameters": { "type": "object", "properties": { "user_id": {"type": "string"}, "fields": { "type": "object", "description": "要更新的字段，如 {'name': '新名字', 'email': '新邮箱'}" } }, "required": ["user_id", "fields"] } }]—## 工具执行层：健壮的实现pythonfrom functools import wrapsfrom typing import Any, Callableimport asyncioimport jsonimport tracebackclass ToolExecutor: """健壮的工具执行器""" def __init__(self): self._tools: dict[str, Callable] = {} self._schemas: dict[str, dict] = {} def register(self, name: str, schema: dict): """装饰器：注册工具""" def decorator(func: Callable): self._tools[name] = func self._schemas[name] = schema @wraps(func) async def wrapper(*args, **kwargs): return await func(*args, **kwargs) return wrapper return decorator async def execute( self, tool_name: str, arguments: dict, timeout: float = 30.0 ) -> dict: """执行工具并返回标准化结果""" if tool_name not in self._tools: return { "success": False, "error": f"工具 '{tool_name}' 不存在", "error_type": "tool_not_found" } # 参数验证 validation_error = self._validate_arguments(tool_name, arguments) if validation_error: return { "success": False, "error": f"参数错误：{validation_error}", "error_type": "invalid_arguments", "hint": "请检查必填参数和参数类型" } # 执行工具（带超时） try: result = await asyncio.wait_for( self._tools[tool_name](**arguments), timeout=timeout ) return {"success": True, "result": result} except asyncio.TimeoutError: return { "success": False, "error": f"工具执行超时（>{timeout}s）", "error_type": "timeout", "hint": "可以尝试减少数据量或稍后重试" } except ValueError as e: return { "success": False, "error": str(e), "error_type": "value_error", "hint": "请检查参数值是否有效" } except Exception as e: return { "success": False, "error": str(e), "error_type": "execution_error", "traceback": traceback.format_exc()[:500] # 截断，避免泄露敏感信息 } def _validate_arguments(self, tool_name: str, arguments: dict) -> str | None: """验证工具参数""" schema = self._schemas[tool_name] required = schema.get("parameters", {}).get("required", []) for field in required: if field not in arguments: return f"缺少必填参数：{field}" return None def get_tools_for_llm(self) -> list[dict]: """获取所有工具的LLM格式定义""" return [ { "type": "function", "function": { "name": name, **schema } } for name, schema in self._schemas.items() ]# 使用示例executor = ToolExecutor()@executor.register("search_web", schema={ "name": "search_web", "description": "在互联网上搜索最新信息", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "搜索关键词"}, "num_results": {"type": "integer", "default": 5} }, "required": ["query"] }})async def search_web(query: str, num_results: int = 5) -> list[dict]: """实际的搜索实现""" # ... 调用搜索API return [{"title": "...", "url": "...", "snippet": "..."}]—## 工具调用循环：Agent的核心逻辑pythonclass ToolCallingAgent: """带工具调用的Agent实现""" def __init__(self, llm_client, tool_executor: ToolExecutor): self.llm = llm_client self.executor = tool_executor self.max_iterations = 10 # 防止无限循环 async def run( self, user_message: str, system_prompt: str = None ) -> str: """运行Agent直到任务完成""" messages = [] if system_prompt: messages.append({"role": "system", "content": system_prompt}) messages.append({"role": "user", "content": user_message}) tools = self.executor.get_tools_for_llm() for iteration in range(self.max_iterations): # 调用LLM response = await self.llm.chat.completions.create( model="gpt-4o", messages=messages, tools=tools, tool_choice="auto" ) assistant_message = response.choices[0].message messages.append(assistant_message.model_dump()) # 检查是否有工具调用 if not assistant_message.tool_calls: # 没有工具调用，任务完成 return assistant_message.content # 执行所有工具调用（支持并行） tool_results = await self._execute_tool_calls( assistant_message.tool_calls ) # 将工具结果添加到消息历史 for tool_call, result in zip(assistant_message.tool_calls, tool_results): messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result, ensure_ascii=False) }) return "任务执行超过最大迭代次数，请检查工具调用逻辑。" async def _execute_tool_calls( self, tool_calls: list ) -> list[dict]: """并行执行多个工具调用""" tasks = [] for tool_call in tool_calls: arguments = json.loads(tool_call.function.arguments) task = self.executor.execute( tool_call.function.name, arguments ) tasks.append(task) # 并行执行所有工具 results = await asyncio.gather(*tasks, return_exceptions=True) # 处理执行异常 processed_results = [] for result in results: if isinstance(result, Exception): processed_results.append({ "success": False, "error": str(result), "error_type": "unexpected_error" }) else: processed_results.append(result) return processed_results—## 工具调用的错误处理策略pythonERROR_HANDLING_SYSTEM_PROMPT = """当工具调用失败时，按以下策略处理：1. **参数错误（invalid_arguments）**： - 检查错误信息，修正参数后重试 - 最多重试2次2. **工具不存在（tool_not_found）**： - 不要重试，使用其他可用工具替代 - 如果无替代方案，告知用户此功能暂不支持3. **超时（timeout）**： - 尝试减少请求数量后重试一次 - 如果仍然超时，告知用户稍后重试4. **执行错误（execution_error）**： - 告知用户发生了什么问题 - 提供可能的解决建议 - 不要重试（可能是持久性错误）在任何情况下，都要清楚地告知用户发生了什么。"""—## 工具调用的安全边界pythonclass SecureToolExecutor(ToolExecutor): """带安全控制的工具执行器""" # 危险工具白名单（需要额外确认） SENSITIVE_TOOLS = { "delete_user_data", "send_email", "execute_sql", "modify_configuration" } # 速率限制 RATE_LIMITS = { "search_web": (10, 60), # 每60秒最多10次 "send_email": (5, 3600), # 每小时最多5次 } async def execute( self, tool_name: str, arguments: dict, require_confirmation: bool = True, **kwargs ) -> dict: """带安全检查的工具执行""" # 检查速率限制 if not await self._check_rate_limit(tool_name): return { "success": False, "error": f"工具 '{tool_name}' 调用过于频繁", "error_type": "rate_limit_exceeded" } # 敏感工具需要确认 if tool_name in self.SENSITIVE_TOOLS and require_confirmation: return { "success": False, "error": f"工具 '{tool_name}' 是敏感操作，需要用户确认", "error_type": "requires_confirmation", "confirmation_required": True, "operation_summary": self._summarize_operation(tool_name, arguments) } return await super().execute(tool_name, arguments, **kwargs) def _summarize_operation(self, tool_name: str, arguments: dict) -> str: """生成操作摘要（用于用户确认）""" summaries = { "delete_user_data": lambda args: f"永久删除用户 {args.get('user_id')} 的全部数据", "send_email": lambda args: f"向 {args.get('to')} 发送邮件：{args.get('subject', '')}", } summarizer = summaries.get(tool_name) return summarizer(arguments) if summarizer else f"执行 {tool_name}" async def _check_rate_limit(self, tool_name: str) -> bool: """检查速率限制（简化实现）""" # 实际应使用Redis计数器 return True—## 工具调用质量监控| 指标 | 描述 | 目标 ||------|------|------|| 工具调用成功率 | 调用成功次数/总次数 | > 95% || 参数错误率 | 参数验证失败的比例 | < 5% || 平均工具调用次数/任务 | 完成一个任务平均调用几次工具 | < 5次 || 工具调用超时率 | 超时次数/总次数 | < 1% || 不必要工具调用率 | 多余工具调用的比例 | < 10% |—## 总结高质量的工具调用系统需要：1. 精心设计的工具定义：名称、描述、参数都要清晰无歧义2. 健壮的执行层：参数验证、超时控制、标准化错误返回3. 合理的Agent循环：最大迭代限制、并行执行支持4. 完善的错误处理：为LLM提供明确的错误处理策略5. 安全边界：敏感操作确认、速率限制、权限控制工具调用设计得好，Agent就能用对工具；设计得不好，再强的LLM也会"手忙脚乱"。