AI Agent 工具调用系统设计：让大模型掌控世界

AI Agent 工具调用系统设计让大模型掌控世界前言工具调用Tool Use / Function Calling是 AI Agent 实现复杂任务的关键能力。通过工具调用大模型可以与外部世界交互执行计算、查询数据库、调用 API真正变成一个能够行动的智能体。我之前设计的代码审查 Agent 就有完善的工具调用能力可以搜索文档、读写文件、执行代码。工具调用系统设计得好不好直接决定了 Agent 的能力边界。今天分享一些我在实践中总结的设计模式和经验。工具调用的基本原理什么是 Function CallingFunction Calling 是让大模型生成结构化调用指令的能力# OpenAI API 的 Function Calling 示例 response client.chat.completions.create( modelgpt-4, messages[{role: user, content: 北京今天天气怎么样}], tools[ { type: function, function: { name: get_weather, description: 获取指定城市的天气信息, parameters: { type: object, properties: { city: { type: string, description: 城市名称 }, unit: { type: string, enum: [celsius, fahrenheit], description: 温度单位 } }, required: [city] } } } ] ) # 模型可能返回 # { # tool_calls: [{ # id: call_xxx, # function: { # name: get_weather, # arguments: {city: 北京, unit: celsius} # } # }] # }工具调用的工作流程用户输入 → LLM 判断是否需要工具 → 是 → 解析工具和参数 → 执行工具 → 返回结果 → LLM 生成最终回答 ↓ 否 直接生成回答工具描述设计Schema 设计原则工具的 JSON Schema 描述直接决定模型能否正确理解和使用工具# 好的工具描述示例 good_tool_schema { name: search_documents, description: 在企业知识库中搜索相关文档。适用于查找技术文档、API说明、最佳实践等。, parameters: { type: object, properties: { query: { type: string, description: 搜索查询语句建议使用完整的问句或关键词组合。例如如何使用 REST API 或 REST API 使用方法 }, top_k: { type: integer, description: 返回的最多文档数量默认 5, default: 5 } }, required: [query] } } # 不好的工具描述示例 bad_tool_schema { name: search, description: 搜索, parameters: { type: object, properties: { q: {type: string} }, required: [q] } }工具分组策略当工具数量很多时应该进行分组TOOL_GROUPS { information: { description: 信息查询类工具, tools: [search_documents, get_weather, get_time] }, code: { description: 代码处理类工具, tools: [execute_code, read_file, write_file] }, communication: { description: 通信类工具, tools: [send_email, send_message] } } def get_tools_for_scenario(scenario: str) - List[dict]: 根据场景选择合适的工具组 if scenario technical_support: return [get_tool_schema(information), get_tool_schema(code)] elif scenario business: return [get_tool_schema(information), get_tool_schema(communication)] else: return get_all_tools()工具执行框架基础执行器from dataclasses import dataclass from typing import Dict, List, Callable, Any import json dataclass class ToolCall: 工具调用请求 id: str name: str arguments: dict dataclass class ToolResult: 工具执行结果 call_id: str success: bool result: Any error: str None class ToolExecutor: 工具执行器 def __init__(self): self.tools: Dict[str, Callable] {} self.schemas: Dict[str, dict] {} def register(self, name: str, schema: dict, func: Callable): 注册工具 self.tools[name] func self.schemas[name] schema def execute(self, call: ToolCall) - ToolResult: 执行工具调用 if call.name not in self.tools: return ToolResult( call_idcall.id, successFalse, resultNone, errorfUnknown tool: {call.name} ) try: func self.tools[call.name] result func(**call.arguments) return ToolResult( call_idcall.id, successTrue, resultresult ) except Exception as e: return ToolResult( call_idcall.id, successFalse, resultNone, errorstr(e) ) def execute_batch(self, calls: List[ToolCall]) - List[ToolResult]: 批量执行工具调用 return [self.execute(call) for call in calls]实际工具实现# 注册实际工具 executor ToolExecutor() # 搜索引擎 def search_web(query: str, num_results: int 5) - dict: 搜索网页 # 实际实现调用搜索引擎 API results google_search(query, num_results) return { query: query, results: [ { title: r.title, snippet: r.snippet, url: r.url } for r in results ] } executor.register( search_web, { name: search_web, description: 搜索网页获取最新信息。适用于查询实时新闻、未知问题等。, parameters: { type: object, properties: { query: {type: string, description: 搜索查询}, num_results: { type: integer, description: 返回结果数量, default: 5 } }, required: [query] } }, search_web ) # 计算器 def calculator(expression: str) - dict: 安全计算数学表达式 # 只允许基本运算 allowed_chars set(0123456789-*/.() ) if not all(c in allowed_chars for c in expression): return {error: 不允许的字符} try: result eval(expression) return {expression: expression, result: result} except Exception as e: return {error: str(e)} executor.register( calculator, { name: calculator, description: 计算数学表达式的值。只支持基本运算符加()、减(-)、乘(*)、除(/)。, parameters: { type: object, properties: { expression: { type: string, description: 数学表达式例如23*4 或 (105)/3 } }, required: [expression] } }, calculator )Agent 工具调用循环完整的工具调用循环class ToolUsingAgent: 支持工具调用的 Agent def __init__(self, llm, executor: ToolExecutor, max_iterations: int 10): self.llm llm self.executor executor self.max_iterations max_iterations def run(self, user_input: str) - str: 运行 Agent 处理用户输入 messages [ {role: system, content: self._get_system_prompt()}, {role: user, content: user_input} ] for iteration in range(self.max_iterations): # 1. 获取 LLM 响应 response self.llm.chat(messages, toolsself.executor.get_all_schemas()) # 2. 检查是否有工具调用 if not response.tool_calls: # 没有工具调用直接返回 return response.content # 3. 执行工具调用 tool_results [] for call in response.tool_calls: tool_call ToolCall( idcall.id, namecall.function.name, argumentsjson.loads(call.function.arguments) ) result self.executor.execute(tool_call) tool_results.append(result) # 4. 将结果添加到消息 messages.append(response.to_message()) for result in tool_results: messages.append({ role: tool, tool_call_id: result.call_id, content: json.dumps(result.result) if result.success else result.error }) return 达到最大迭代次数 def _get_system_prompt(self) - str: return 你是一个智能助手可以通过调用工具来完成任务。 Available tools: self.executor.get_tools_description()结果验证与重试class VerifiedToolExecutor(ToolExecutor): 带验证的工具执行器 def execute_with_verification( self, call: ToolCall, expected_format: dict None ) - ToolResult: 执行并验证结果 result self.execute(call) if not result.success: return result if expected_format: # 验证结果格式 is_valid, error self._verify_format(result.result, expected_format) if not is_valid: return ToolResult( call_idcall.id, successFalse, resultNone, errorfResult format error: {error} ) return result def _verify_format(self, result: Any, expected: dict) - tuple: 验证结果格式 if expected.get(type) array: if not isinstance(result, list): return False, Expected array elif expected.get(type) object: if not isinstance(result, dict): return False, Expected object for key in expected.get(required, []): if key not in result: return False, fMissing required field: {key} return True, None工具调用的安全考虑输入验证class SecureToolExecutor(ToolExecutor): 安全的工具执行器 def execute(self, call: ToolCall) - ToolResult: # 1. 参数验证 schema self.schemas.get(call.name) if not schema: return ToolResult(call.id, False, None, Unknown tool) # 检查必需参数 required schema.get(parameters, {}).get(required, []) for param in required: if param not in call.arguments: return ToolResult( call.id, False, None, fMissing required parameter: {param} ) # 2. 值域检查 properties schema.get(parameters, {}).get(properties, {}) for param, spec in properties.items(): if param in call.arguments: value call.arguments[param] if not self._validate_param_value(value, spec): return ToolResult( call.id, False, None, fInvalid value for parameter {param} ) # 3. 敏感操作检查 if self._is_sensitive_operation(call): # 记录审计日志 self._audit_log(call) return super().execute(call) def _validate_param_value(self, value, spec) - bool: 验证参数值 param_type spec.get(type) if param_type string: if maxLength in spec and len(value) spec[maxLength]: return False if enum in spec and value not in spec[enum]: return False elif param_type integer: if not isinstance(value, int): return False if minimum in spec and value spec[minimum]: return False if maximum in spec and value spec[maximum]: return False return True def _is_sensitive_operation(self, call: ToolCall) - bool: 检查是否为敏感操作 sensitive_tools {delete_file, send_email, execute_sql} return call.name in sensitive_tools异步工具调用import asyncio from typing import List class AsyncToolExecutor: 异步工具执行器 def __init__(self): self.tools: Dict[str, Callable] {} def register(self, name: str, func: Callable): self.tools[name] func async def execute_async(self, call: ToolCall) - ToolResult: 异步执行单个工具 if call.name not in self.tools: return ToolResult(call.id, False, None, Unknown tool) try: func self.tools[call.name] # 如果是异步函数 if asyncio.iscoroutinefunction(func): result await func(**call.arguments) else: # 在线程池中运行同步函数 loop asyncio.get_event_loop() result await loop.run_in_executor(None, lambda: func(**call.arguments)) return ToolResult(call.id, True, result) except Exception as e: return ToolResult(call.id, False, None, str(e)) async def execute_batch_async(self, calls: List[ToolCall]) - List[ToolResult]: 批量异步执行 tasks [self.execute_async(call) for call in calls] return await asyncio.gather(*tasks)工具调用优化并行 vs 串行class SmartToolExecutor(ToolExecutor): 智能工具执行器 def execute_with_plan(self, calls: List[ToolCall]) - List[ToolResult]: 分析依赖关系并优化执行顺序 # 1. 构建依赖图 dependencies self._build_dependency_graph(calls) # 2. 找出可以并行的调用 independent_calls [ call for call in calls if not dependencies.get(call.id) ] # 3. 并行执行独立的调用 parallel_results self.execute_batch(independent_calls) # 4. 串行执行有依赖的调用 remaining_calls [c for c in calls if c.id in dependencies] serial_results [] for call in remaining_calls: deps dependencies[call.id] # 等待依赖完成 for dep_id in deps: self._wait_for_result(dep_id) result self.execute(call) serial_results.append(result) return parallel_results serial_results总结工具调用是 AI Agent 实现复杂任务的核心能力。设计一个好的工具调用系统需要考虑清晰的工具描述让模型准确理解工具用途和参数健壮的执行框架错误处理、验证、重试机制安全保障输入验证、敏感操作审计性能优化并行执行、依赖分析希望这些经验对大家有帮助。