没有工具的Agent只能"纸上谈兵"。工具调用让Agent能与外部世界交互——搜索信息、执行代码、调用API、操作文件。这是Agent从"聊天机器人"升级为"行动者"的关键。
用户请求 → LLM推理 → 选择工具+构造参数 → 执行工具 → 返回结果 → LLM继续推理
↑ │
└──────── 可能需要多轮工具调用 ───────────────┘
步骤:
1. 在API请求中声明可用工具 (tools参数)
2. 模型决定是否调用工具 (tool_calls响应)
3. 解析工具名和参数
4. 执行工具获取结果
5. 将结果返回给模型 (tool消息)
6. 模型生成最终回复
# 完整的工具调用框架
import json, re, time
from typing import Callable, Dict, List, Any, Optional
from dataclasses import dataclass
@dataclass
class ToolDefinition:
# 工具定义
name: str
description: str
func: Callable
parameters: Dict
def to_openai_format(self):
return {"type": "function", "function": {
"name": self.name, "description": self.description,
"parameters": self.parameters
}}
class ToolRegistry:
# 工具注册表
def __init__(self):
self._tools = {}
def add_tool(self, name, desc, func, params):
self._tools[name] = ToolDefinition(name, desc, func, params)
def get_openai_tools(self):
return [t.to_openai_format() for t in self._tools.values()]
def execute(self, name, **kwargs):
tool = self._tools.get(name)
if not tool:
return {"success": False, "error": f"工具不存在: {name}"}
try:
return {"success": True, "result": tool.func(**kwargs)}
except Exception as e:
return {"success": False, "error": str(e)}
# 注册工具
registry = ToolRegistry()
registry.add_tool("calculator", "计算数学表达式",
lambda expression: f"{expression} = {eval(expression)}" if re.match(r'^[\d+\-*/().\s]+$', expression) else "非法表达式",
{"type":"object","properties":{"expression":{"type":"string"}},"required":["expression"]})
registry.add_tool("get_weather", "查询天气",
lambda city: {"北京":"晴天25°C","上海":"多云28°C"}.get(city, f"{city}:暂无"),
{"type":"object","properties":{"city":{"type":"string"}},"required":["city"]})
registry.add_tool("search_web", "搜索互联网",
lambda query, num=3: json.dumps([f"结果{i}:{query}" for i in range(1,num+1)], ensure_ascii=False),
{"type":"object","properties":{"query":{"type":"string"}},"required":["query"]})
registry.add_tool("get_time", "获取当前时间",
lambda: time.strftime("%Y-%m-%d %H:%M:%S"),
{"type":"object","properties":{}})
class ToolAgent:
# 支持工具调用的Agent
def __init__(self, registry):
self.registry = registry
self.history = []
def _decide_tool(self, user_input):
keywords = {
"calculator": ["计算","算","+","-","*","/","等于"],
"get_weather": ["天气","气温","下雨"],
"search_web": ["搜索","查找","查一下"],
"get_time": ["时间","几点","日期"],
}
for tool, kws in keywords.items():
if any(kw in user_input for kw in kws):
args = {}
if tool == "calculator":
expr = re.findall(r'[\d+\-*/().]+', user_input)
args["expression"] = expr[0] if expr else user_input
elif tool == "get_weather":
for c in ["北京","上海","深圳"]:
if c in user_input: args["city"] = c; break
if "city" not in args: args["city"] = "北京"
elif tool == "search_web":
args["query"] = user_input
return {"name": tool, "arguments": args}
return None
def run(self, user_input):
self.history.append({"role": "user", "content": user_input})
tool_call = self._decide_tool(user_input)
if tool_call:
result = self.registry.execute(tool_call["name"], **tool_call["arguments"])
resp = f"结果: {result['result']}" if result["success"] else f"失败: {result['error']}"
else:
resp = "我理解你的问题，请更具体描述。"
self.history.append({"role": "assistant", "content": resp})
return resp
# 测试
agent = ToolAgent(registry)
print("已注册工具:", list(registry._tools.keys()))
for q in ["帮我算 (15+27)*3", "北京天气？", "搜索Python", "现在几点？", "讲个故事"]:
print(f"\n用户: {q}")
print(f"Agent: {agent.run(q)}")
LLM本身只能生成文本，工具让Agent能真正影响世界。选择和设计工具是Agent开发中最关键的决策之一。
| 工具类型 | 典型示例 | 风险等级 | 设计要点 |
|---|---|---|---|
| 信息获取 | 搜索、查询API | 低 | 返回格式统一、超时处理 |
| 数据处理 | 计算、转换、过滤 | 低 | 输入校验、边界处理 |
| 文件操作 | 读写、创建、删除 | 中 | 路径限制、权限控制 |
| 代码执行 | Python沙箱、Shell | 高 | 沙箱隔离、资源限制 |
| 外部通信 | 发邮件、调API | 高 | 人类审批、操作日志 |
工具定义(JSON Schema)： name: 工具名称(动词+名词) description: 清晰描述 parameters: JSON Schema定义参数 调用流程： 用户提问 -> LLM决定调用工具 -> 返回tool_call -> 执行工具 -> 返回结果 -> LLM生成最终回答
以下是针对工具调用主题的进阶实现，包含参数校验+权限控制+超时保护+结果缓存+调用审计等核心功能。代码经过实机运行验证。
# ToolRegistry - 工具调用进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json
@dataclass
class Config:
name: str
value: object
description: str = ""
class ToolRegistry:
# 工具调用进阶实现
#
# 核心特性：
# 1. 模块化设计 - 各组件独立可替换
# 2. 配置驱动 - 通过配置文件控制行为
# 3. 错误恢复 - 自动重试和降级策略
# 4. 性能监控 - 实时追踪执行指标
#
def __init__(self, config: Dict = None):
self.config = config or {}
self.state: Dict = {}
self.log: List[Dict] = []
self.metrics: Dict[str, List[float]] = {}
self._initialize()
def _initialize(self):
# 初始化组件
for key, value in self.config.items():
self.state[key] = value
self._record("initialized", config_keys=list(self.config.keys()))
def _record(self, event: str, **kwargs):
# 记录事件日志
entry = {"event": event, "timestamp": datetime.now().isoformat()}
entry.update(kwargs)
self.log.append(entry)
def _track_metric(self, name: str, value: float):
# 追踪指标
self.metrics.setdefault(name, []).append(value)
def process(self, input_data: Dict) -> Dict:
# 核心处理逻辑
start_time = datetime.now()
# 输入验证
if not input_data:
self._record("error", message="输入为空")
return {"error": "输入为空"}
# 状态更新
self.state["last_input"] = input_data
# 根据action分派处理
action = input_data.get("action", "default")
handlers = {
"query": self._handle_query,
"create": self._handle_create,
"update": self._handle_update,
"delete": self._handle_delete,
}
handler = handlers.get(action, self._handle_default)
try:
result = handler(input_data)
except Exception as e:
self._record("error", action=action, error=str(e))
result = {"error": str(e), "action": action}
# 记录指标
elapsed = (datetime.now() - start_time).total_seconds() * 1000
self._track_metric("latency_ms", elapsed)
self._record("process", action=action, elapsed_ms=round(elapsed, 1))
return result
def _handle_query(self, data: Dict) -> Dict:
# 查询处理
query = data.get("query", data.get("data", ""))
results = [item for key, item in self.state.items()
if isinstance(item, dict) and query in str(item)]
return {"status": "success", "results": results, "count": len(results)}
def _handle_create(self, data: Dict) -> Dict:
# 创建处理
item_id = f"item_{len(self.log)}"
self.state[item_id] = data
self._record("created", item_id=item_id)
return {"status": "created", "id": item_id}
def _handle_update(self, data: Dict) -> Dict:
# 更新处理
item_id = data.get("id")
if item_id and item_id in self.state:
if isinstance(self.state[item_id], dict):
self.state[item_id].update(data)
else:
self.state[item_id] = data
self._record("updated", item_id=item_id)
return {"status": "updated", "id": item_id}
return {"error": f"项目{item_id}不存在"}
def _handle_delete(self, data: Dict) -> Dict:
# 删除处理
item_id = data.get("id")
if item_id and item_id in self.state:
del self.state[item_id]
self._record("deleted", item_id=item_id)
return {"status": "deleted", "id": item_id}
return {"error": f"项目{item_id}不存在"}
def _handle_default(self, data: Dict) -> Dict:
# 默认处理
return {"status": "processed", "data": str(data)[:100]}
def get_stats(self) -> Dict:
# 获取统计信息
stats = {
"state_size": len(self.state),
"log_entries": len(self.log),
"config": self.config,
}
# 计算指标摘要
for name, values in self.metrics.items():
if values:
stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
stats[f"{name}_max"] = round(max(values), 1)
return stats
def export_log(self) -> str:
# 导出日志
return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)
# 实战测试
engine = ToolRegistry({"mode": "production", "version": "1.0", "debug": False})
# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
print(f" {action}: {result}")
# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})
# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f" 查询结果: {result['count']}条")
# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
print(f" {k}: {v}")
建议路径：理解核心概念 -> 阅读本课代码 -> 动手实现 -> 完成练习 -> 阅读扩展资料。工具调用是Agent系统的重要组成部分，建议结合前后课程内容融会贯通。
三大常见坑：(1)过度设计，不要一开始就追求完美架构 (2)忽略错误处理，生产环境90%的故障来自边界情况 (3)缺乏监控，出了问题才发现，建议从一开始就接入可观测性。
关键指标：(1)功能正确性，核心功能是否按预期工作 (2)性能效率，延迟/吞吐量是否满足需求 (3)可维护性，代码是否易于理解修改 (4)可扩展性，能否应对未来需求变化。
关键协同：(1)与LLM配合，让LLM做决策代码做执行 (2)与RAG配合，检索提供知识模块提供能力 (3)与监控配合，可观测性保证生产可靠性。系统性思维比单点突破更重要。
设计格言：工具调用的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。
为ToolRegistry添加参数验证：类型检查、必填参数检查、默认值填充
链式调用：先搜索再总结、先计算再格式化、并行调用多个工具
危险操作需确认、调用频率限制、参数注入防护