第三章：多工具协作¶
多工具场景¶
在实际应用中，往往需要多个工具协作完成复杂任务：
定义多个工具¶
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"},
"date": {"type": "string", "description": "日期，如：今天、明天"}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "send_email",
"description": "发送电子邮件",
"parameters": {
"type": "object",
"properties": {
"to": {"type": "string", "description": "收件人邮箱"},
"subject": {"type": "string", "description": "邮件主题"},
"body": {"type": "string", "description": "邮件正文"}
},
"required": ["to", "subject", "body"]
}
}
},
{
"type": "function",
"function": {
"name": "search_web",
"description": "搜索互联网信息",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "执行数学计算",
"parameters": {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式"}
},
"required": ["expression"]
}
}
}
]
工具执行器¶
创建一个统一的工具执行框架：
import json
from typing import Dict, Callable, Any
from openai import OpenAI
client = OpenAI()
class ToolExecutor:
"""工具执行器"""
def __init__(self):
self.tools: Dict[str, Callable] = {}
self.tool_definitions: list = []
def register(self, definition: dict, func: Callable):
"""注册工具"""
name = definition["function"]["name"]
self.tools[name] = func
self.tool_definitions.append(definition)
return self # 支持链式调用
def execute(self, tool_call) -> Any:
"""执行单个工具调用"""
name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
if name not in self.tools:
return {"error": f"未知工具: {name}"}
try:
return self.tools[name](**args)
except Exception as e:
return {"error": str(e)}
def execute_all(self, tool_calls: list) -> list:
"""执行多个工具调用"""
return [self.execute(tc) for tc in tool_calls]
# 使用示例
executor = ToolExecutor()
# 注册工具
executor.register(
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string"}
},
"required": ["city"]
}
}
},
get_weather # 实际函数
)
executor.register(
{
"type": "function",
"function": {
"name": "send_email",
"description": "发送邮件",
"parameters": {
"type": "object",
"properties": {
"to": {"type": "string"},
"subject": {"type": "string"},
"body": {"type": "string"}
},
"required": ["to", "subject", "body"]
}
}
},
send_email
)
多轮对话管理¶
class ConversationManager:
"""对话管理器"""
def __init__(self, tool_executor: ToolExecutor, model: str = "gpt-4o"):
self.client = OpenAI()
self.executor = tool_executor
self.model = model
self.messages: list = []
def chat(self, user_message: str, max_turns: int = 10) -> str:
"""处理用户消息"""
self.messages.append({"role": "user", "content": user_message})
for _ in range(max_turns):
# 调用模型
response = self.client.chat.completions.create(
model=self.model,
messages=self.messages,
tools=self.executor.tool_definitions,
tool_choice="auto"
)
message = response.choices[0].message
self.messages.append(message)
# 检查是否需要调用工具
if not message.tool_calls:
return message.content
# 执行所有工具调用
for tool_call in message.tool_calls:
result = self.executor.execute(tool_call)
self.messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False)
})
return "抱歉，处理您的请求时超出了最大轮次限制。"
def reset(self):
"""重置对话"""
self.messages = []
# 使用示例
manager = ConversationManager(executor)
# 复杂查询
answer = manager.chat(
"查一下北京明天的天气，如果会下雨就发邮件到 user@example.com 提醒我带伞"
)
print(answer)
工具依赖处理¶
有些场景下，工具调用之间存在依赖关系：
# 用户：查一下苹果公司的股价，然后计算如果我买100股需要多少钱
# 工具调用顺序：
# 1. get_stock_price("AAPL") → 180.5
# 2. calculate("180.5 * 100") → 18050
模型会自动处理这种依赖：
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "user", "content": "查一下苹果公司的股价，然后计算如果我买100股需要多少钱"}
],
tools=tools
)
# 第一轮：模型返回 get_stock_price 调用
# 执行后，第二轮：模型返回 calculate 调用（使用第一轮的结果）
条件执行¶
def smart_tool_conversation(user_message: str) -> str:
"""智能工具调用对话"""
messages = [{"role": "user", "content": user_message}]
while True:
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto"
)
message = response.choices[0].message
messages.append(message)
if not message.tool_calls:
return message.content
# 处理工具调用
for tool_call in message.tool_calls:
result = executor.execute(tool_call)
# 添加条件判断逻辑
if tool_call.function.name == "get_weather":
weather_data = result
# 可以在这里添加额外逻辑
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False)
})
工具调用日志¶
import logging
from datetime import datetime
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class LoggedToolExecutor(ToolExecutor):
"""带日志的工具执行器"""
def execute(self, tool_call) -> Any:
start_time = datetime.now()
name = tool_call.function.name
args = tool_call.function.arguments
logger.info(f"[工具调用] {name}({args})")
try:
result = super().execute(tool_call)
duration = (datetime.now() - start_time).total_seconds()
logger.info(f"[工具结果] {name} 完成，耗时 {duration:.2f}s")
return result
except Exception as e:
logger.error(f"[工具错误] {name} 失败: {e}")
raise
# 使用示例
executor = LoggedToolExecutor()
# ... 注册工具 ...
工具权限控制¶
class PermissionControlledExecutor(ToolExecutor):
"""带权限控制的工具执行器"""
def __init__(self):
super().__init__()
self.permissions: Dict[str, set] = {} # 用户 -> 允许的工具
def set_permissions(self, user_id: str, allowed_tools: set):
"""设置用户权限"""
self.permissions[user_id] = allowed_tools
def execute(self, tool_call, user_id: str = None) -> Any:
"""执行时检查权限"""
name = tool_call.function.name
if user_id and user_id in self.permissions:
if name not in self.permissions[user_id]:
return {"error": f"您没有权限使用工具: {name}"}
return super().execute(tool_call)
# 使用示例
executor = PermissionControlledExecutor()
# 用户 A 只能查询天气
executor.set_permissions("user_a", {"get_weather", "search_web"})
# 用户 B 可以使用所有工具
executor.set_permissions("user_b", {"get_weather", "send_email", "search_web", "calculate"})
小结¶
本章学习了：
- ✅ 多工具定义和管理
- ✅ 工具执行器设计
- ✅ 多轮对话管理
- ✅ 工具依赖和条件执行
- ✅ 日志和权限控制
下一章¶
第四章：自定义工具开发 - 学习如何开发自己的工具库。