当前位置：首页 > news >正文

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

news 2026/5/22 12:18:58

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

前言

工具调用（Tool Use / Function Calling）是 AI Agent 实现复杂任务的关键能力。通过工具调用，大模型可以与外部世界交互，执行计算、查询数据库、调用 API，真正变成一个能够"行动"的智能体。

我之前设计的代码审查 Agent 就有完善的工具调用能力，可以搜索文档、读写文件、执行代码。工具调用系统设计得好不好，直接决定了 Agent 的能力边界。今天分享一些我在实践中总结的设计模式和经验。

工具调用的基本原理

什么是 Function Calling

Function Calling 是让大模型生成结构化调用指令的能力：

# OpenAI API 的 Function Calling 示例 response = client.chat.completions.create( model="gpt-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_id=call.id, success=False, result=None, error=f"Unknown tool: {call.name}" ) try: func = self.tools[call.name] result = func(**call.arguments) return ToolResult( call_id=call.id, success=True, result=result ) except Exception as e: return ToolResult( call_id=call.id, success=False, result=None, error=str(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": "数学表达式，例如：2+3*4 或 (10+5)/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, tools=self.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( id=call.id, name=call.function.name, arguments=json.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_id=call.id, success=False, result=None, error=f"Result 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, f"Missing 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, f"Missing 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, f"Invalid 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