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
总结
工具调用是 AI Agent 实现复杂任务的核心能力。设计一个好的工具调用系统需要考虑：
希望这些经验对大家有帮助。
