工具调用实战
工具（Tools）是 AI Agent 与外部世界交互的桥梁。通过工具，Agent 可以搜索信息、执行代码、调用 API，真正"行动"起来。
为什么需要工具
大模型本身有局限性：
- 知识有截止日期
- 无法访问实时数据
- 不能执行任意代码
- 无法操作外部系统
工具让 AI 能够：
- 获取实时信息（搜索、API）
- 执行计算（代码解释器）
- 操作文件系统
- 与第三方服务交互
工具设计原则
1. 清晰的描述
工具的名称和描述要明确，让模型知道何时使用：
# ❌ 不好：描述模糊
def search(query):
"""搜索"""
pass
# ✅ 好：描述清晰
def search_weather(query):
"""搜索天气信息。输入：城市名称（如"北京"），输出：该城市的当前天气"""
pass2. 简洁的输入输出
# ❌ 复杂参数
def send_email(to, cc, bcc, subject, body, attachments, priority, ...):
pass
# ✅ 简单直接
def send_email(to, subject, body):
"""发送邮件。参数：收件人、主题、正文"""
pass3. 单一职责
每个工具只做一件事：
# ❌ 一个工具做多件事
def handle_request(request):
if request.type == "weather":
return get_weather(request.city)
elif request.type == "news":
return get_news(request.topic)
...
# ✅ 拆分为独立工具
def get_weather(city):
"""查询城市天气"""
def search_news(topic):
"""搜索新闻"""常用工具类型
1. 搜索工具
def web_search(query):
"""搜索互联网信息。用于获取实时数据、新闻等。"""
import requests
# 实际实现可接入 Google、Bing API 等
return f"搜索结果：{query}"
def wiki_search(query):
"""搜索维基百科。用于获取权威词条信息。"""
pass2. 计算工具
def calculate(expression):
"""数学计算。用于精确的数学运算。"""
import ast
import operator
ops = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
}
def eval_expr(node):
if isinstance(node, ast.Constant):
return node.value
elif isinstance(node, ast.BinOp):
left = eval_expr(node.left)
right = eval_expr(node.right)
return ops[type(node.op)](left, right)
tree = ast.parse(expression, mode='eval')
return str(eval_expr(tree.body))3. 代码执行工具
def execute_code(code):
"""执行 Python 代码。用于运行计算、分析数据等。"""
import subprocess
result = subprocess.run(
["python", "-c", code],
capture_output=True,
text=True
)
return result.stdout or result.stderr4. API 调用工具
def get_weather(city):
"""查询天气。输入城市名，返回天气信息。"""
import requests
# 示例：调用天气 API
api_key = "your-api-key"
url = f"https://api.weather.com/v3/weather"
response = requests.get(url, params={"city": city, "key": api_key})
return response.json()
def fetch_stock_price(symbol):
"""查询股票价格。输入股票代码（如 AAPL）。"""
pass
def send_slack_message(channel, message):
"""发送 Slack 消息。用于通知。"""
pass5. 文件操作工具
def read_file(path):
"""读取文件内容。输入：文件路径。"""
with open(path, 'r', encoding='utf-8') as f:
return f.read()
def write_file(path, content):
"""写入文件。参数：文件路径、内容。"""
with open(path, 'w', encoding='utf-8') as f:
f.write(content)
return "写入成功"
def list_files(directory):
"""列出目录下的文件。"""
import os
return os.listdir(directory)完整示例：天气预报 Agent
from openai import OpenAI
client = OpenAI()
# 定义工具
def get_weather(city):
"""获取城市天气。输入：城市名称（如北京、上海）"""
# 模拟天气数据
weather_data = {
"北京": "晴，15°C，PM2.5: 45",
"上海": "多云，18°C，PM2.5: 62",
"广州": "小雨，22°C，PM2.5: 88"
}
return weather_data.get(city, "未知城市")
def search_city_info(city):
"""搜索城市信息。用于获取城市的基本介绍。"""
info = {
"北京": "中国的首都，政治、文化中心",
"上海": "中国的经济、金融中心",
"广州": "中国南方的重要城市，商贸中心"
}
return info.get(city, "未知")
TOOLS = {
"get_weather": get_weather,
"search_city_info": search_city_info
}
TOOL_DESCRIPTIONS = """
- get_weather: 获取城市天气。输入：城市名称
- search_city_info: 搜索城市信息。输入：城市名称
"""
def weather_agent(user_input):
messages = [
{"role": "system", "content": f"""你是一个天气助手，可以使用工具来回答用户问题。
可用工具：
{TOOL_DESCRIPTIONS}
回答格式要求：
1. 先思考需要什么信息
2. 调用必要的工具
3. 根据结果回答用户"""},
{"role": "user", "content": user_input}
]
for _ in range(10):
response = client.chat.completions.create(
model="gpt-4",
messages=messages
)
content = response.choices[0].message.content
messages.append({"role": "assistant", "content": content})
# 解析工具调用
tool_calls = parse_tool_calls(content)
if not tool_calls:
return content
# 执行工具
for tool_name, tool_input in tool_calls:
if tool_name in TOOLS:
result = TOOLS[tool_name](tool_input)
messages.append({
"role": "user",
"content": f"工具 {tool_name} 返回：{result}"
})
return "已达到最大迭代次数"
def parse_tool_calls(content):
"""从模型输出中解析工具调用"""
import re
# 简单解析：查找 [TOOL:tool_name]input[/TOOL] 格式
pattern = r'\[TOOL:(\w+)\](.*?)\[/TOOL\]'
matches = re.findall(pattern, content, re.DOTALL)
if matches:
return matches
# 检查是否直接返回答案
return None
# 测试
print(weather_agent("北京今天天气怎么样？"))工具调用进阶
1. 工具选择策略
def select_best_tool(query, tools):
"""让模型选择最合适的工具"""
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "根据用户问题，选择最合适的工具。"},
{"role": "user", "content": f"问题：{query}\n工具：{list(tools.keys())}"}
]
)
return response.choices[0].message.content2. 并行工具调用
import concurrent.futures
def execute_tools_parallel(tool_calls):
"""并行执行多个工具"""
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = {
executor.submit(tool, input): (name, input)
for name, tool in TOOLS.items()
for input in tool_calls.get(name, [])
}
results = {}
for future in concurrent.futures.as_completed(futures):
name, input = futures[future]
results[f"{name}:{input}"] = future.result()
return results3. 工具链
def tool_chain(query, chain):
"""顺序执行工具链"""
result = query
for tool_name in chain:
result = TOOLS[tool_name](result)
return result安全注意事项
- 输入验证：验证工具输入，防止注入攻击
- 权限控制：限制工具的操作范围
- 日志记录：记录工具调用历史
- 超时处理：设置工具调用超时
- 错误处理：优雅处理工具异常
def safe_execute(tool, input_data, timeout=30):
"""安全执行工具"""
import signal
def timeout_handler(signum, frame):
raise TimeoutError("工具执行超时")
try:
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(timeout)
result = tool(input_data)
signal.alarm(0)
return result
except TimeoutError:
return "执行超时"
except Exception as e:
return f"执行错误：{str(e)}"下一步
- 多代理系统 - 了解多个 Agent 如何协作