工具调用机制
High Contrast
Dark Mode
Light Mode
Sepia
Forest
2 min read457 words

工具调用机制

工具调用(Function Calling / Tool Use)是 Agent 与外部世界交互的关键能力。没有工具,Agent 只是一个聊天机器人。

工具调用原理

sequenceDiagram participant U as 用户 participant A as Agent (LLM) participant T as 工具 U->>A: 查一下北京天气 Note over A: 分析: 需要调用天气 API A->>T: get_weather(city="北京") T-->>A: {"temp": 22, "condition": "晴"} Note over A: 整合结果生成回答 A->>U: 北京今天晴,气温 22°C

LLM 并不真的「执行」工具。它只是输出一段结构化的调用指令,由应用层执行后将结果返回给 LLM。

OpenAI Function Calling

"""
OpenAI 原生工具调用
"""
from openai import OpenAI
import json
client = OpenAI()
# 1. 定义工具
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"]
}
}
},
{
"type": "function",
"function": {
"name": "search_docs",
"description": "在知识库中搜索文档",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
},
"top_k": {
"type": "integer",
"description": "返回结果数量"
}
},
"required": ["query"]
}
}
}
]
# 2. 工具实现
def get_weather(city: str, unit: str = "celsius") -> dict:
"""模拟天气 API"""
return {"city": city, "temp": 22, "condition": "晴", "unit": unit}
def search_docs(query: str, top_k: int = 3) -> list:
"""模拟知识库搜索"""
return [{"title": f"关于{query}的文档", "content": f"{query}的详细说明..."}]
TOOL_MAP = {
"get_weather": get_weather,
"search_docs": search_docs,
}
# 3. Agent 循环
def agent_loop(user_message: str) -> str:
"""完整的工具调用循环"""
messages = [
{"role": "system", "content": "你是一个有用的助手,可以使用提供的工具。"},
{"role": "user", "content": user_message},
]
while True:
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
)
message = response.choices[0].message
# 无工具调用 → 返回最终答案
if not message.tool_calls:
return message.content
# 将 assistant 消息加入历史
messages.append(message)
# 执行每个工具调用
for tool_call in message.tool_calls:
func_name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
print(f"🔧 调用: {func_name}({args})")
# 执行工具
result = TOOL_MAP[func_name](**args)
# 将结果返回给 LLM
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False),
})
# 使用
answer = agent_loop("北京今天天气怎么样?")
print(answer)

构建工具注册系统

实际项目中需要一个优雅的工具注册和管理系统:

"""
通用工具注册系统
"""
from typing import Callable
import inspect
class ToolRegistry:
"""工具注册中心"""
def __init__(self):
self._tools: dict[str, dict] = {}
def tool(self, description: str):
"""装饰器:注册工具"""
def decorator(func: Callable):
schema = self._build_schema(func, description)
self._tools[func.__name__] = {
"function": func,
"schema": schema,
}
return func
return decorator
def _build_schema(self, func: Callable, description: str) -> dict:
"""自动从函数签名生成 JSON Schema"""
sig = inspect.signature(func)
properties = {}
required = []
type_map = {
str: "string",
int: "integer",
float: "number",
bool: "boolean",
list: "array",
}
for name, param in sig.parameters.items():
annotation = param.annotation
prop_type = type_map.get(annotation, "string")
properties[name] = {"type": prop_type, "description": name}
if param.default is inspect.Parameter.empty:
required.append(name)
return {
"type": "function",
"function": {
"name": func.__name__,
"description": description,
"parameters": {
"type": "object",
"properties": properties,
"required": required,
},
},
}
def get_schemas(self) -> list[dict]:
"""获取所有工具的 Schema"""
return [t["schema"] for t in self._tools.values()]
def execute(self, name: str, args: dict):
"""执行工具"""
if name not in self._tools:
raise ValueError(f"未知工具: {name}")
return self._tools[name]["function"](**args)
def list_tools(self) -> list[str]:
"""列出所有已注册工具"""
return list(self._tools.keys())
# ==================
# 使用装饰器注册工具
# ==================
registry = ToolRegistry()
@registry.tool("执行 Python 代码并返回结果")
def run_python(code: str) -> str:
"""安全执行 Python 代码"""
try:
result = {}
exec(code, {"__builtins__": {}}, result)
return str(result)
except Exception as e:
return f"执行错误: {e}"
@registry.tool("读取文件内容")
def read_file(filepath: str) -> str:
"""读取文件"""
try:
with open(filepath, "r", encoding="utf-8") as f:
return f.read()
except FileNotFoundError:
return f"文件不存在: {filepath}"
@registry.tool("向文件写入内容")
def write_file(filepath: str, content: str) -> str:
"""写入文件"""
with open(filepath, "w", encoding="utf-8") as f:
f.write(content)
return f"已写入文件: {filepath}"
# 查看已注册工具
print(registry.list_tools())
# ['run_python', 'read_file', 'write_file']
# 获取 OpenAI 格式的 Schema
schemas = registry.get_schemas()
# 执行工具
result = registry.execute("read_file", {"filepath": "README.md"})

常见工具类型

工具类型 用途 示例
搜索工具 获取外部信息 Web 搜索、知识库检索
代码执行 数据处理和计算 Python 执行器、SQL 客户端
文件操作 读写文件 文件读取、创建、修改
API 调用 集成外部服务 天气 API、邮件、Slack
数据库 持久化存储 SQL 查询、向量数据库

工具调用的最佳实践

graph TB A[最佳实践] --> B[设计] A --> C[安全] A --> D[性能] B --> B1[工具名称要直观] B --> B2[参数描述要详细] B --> B3[一个工具只做一件事] C --> C1[输入验证] C --> C2[权限控制] C --> C3[沙箱执行代码] D --> D1[缓存结果] D --> D2[超时设置] D --> D3[并行执行] style A fill:#e3f2fd,stroke:#1976d2,stroke-width:3px
  1. 描述清晰:工具描述决定了 LLM 能否正确选择工具
  2. 参数简单:尽量用基本类型,避免复杂嵌套
  3. 返回值标准化:始终返回字符串,包含足够上下文
  4. 错误处理:工具应优雅处理错误,而不是抛出异常
  5. 安全沙箱:代码执行类工具必须在沙箱中运行

本章小结

下一章:开始构建完整的单一 Agent,包含工具、记忆和规划的完整系统。