Tool Calling（工具调用）是现代 AI Agent 的核心能力，让大模型能够主动使用外部工具来完成复杂任务。本文深入探讨 Tool Calling 的工程实践，包括设计模式、错误处理、安全机制等关键议题。
一、Tool Calling 是什么？为什么重要？
1.1 定义与原理
Tool Calling（工具调用） 是一种让大语言模型（LLM）能够调用外部函数或 API 的技术能力。当模型识别到用户请求需要特定操作时，它会生成结构化的函数调用请求，而非普通的文本回复。
┌─────────────────────────────────────────────────────────────────────────────┐
│ Tool Calling 工作原理 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ 用户请求 │
│ ┌─────────────┐ │
│ │ 帮我查下北京 │ │
│ │ 明天的天气 │ │
│ └─────────────┘ │
│ ↓ │
│ ┌─────────────────┐ │
│ │ 大模型推理 │ │
│ │ 识别需要工具 │ │
│ └─────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────┐ │
│ │ 模型输出结构化调用请求 │ │
│ │ { │ │
│ │ "name": "get_weather", │ │
│ │ "arguments": { │ │
│ │ "city": "北京", │ │
│ │ "date": "明天" │ │
│ │ } │ │
│ │ } │ │
│ └─────────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────┐ │
│ │ 工具执行器 │ │
│ │ 调用天气 API │ │
│ └─────────────────┘ │
│ ↓ │
│ ┌─────────────────┐ │
│ │ 结果注入 │ │
│ │ 天气: 晴, 15°C │ │
│ └─────────────────┘ │
│ ↓ │
│ ┌─────────────────┐ │
│ │ 最终生成 │ │
│ │ "北京明天天气 │ │
│ │ 晴朗，温度 │ │
│ │ 15°C左右..." │ │
│ └─────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘1.2 为什么 Tool Calling 如此重要？
核心价值：突破 LLM 的能力边界
1.3 OpenAI Function Calling 协议详解
OpenAI 在 2023 年 6 月正式引入 Function Calling，为行业提供了标准化参考。
{
// 工具定义 schema
"tools": [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市和日期的天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，需要使用中文，如：北京、上海",
"enum": ["北京", "上海", "广州", "深圳"]
},
"date": {
"type": "string",
"description": "查询日期，格式为 YYYY-MM-DD",
"pattern": "^\\d{4}-\\d{2}-\\d{2}$"
},
"unit": {
"type": "string",
"description": "温度单位",
"enum": ["celsius", "fahrenheit"],
"default": "celsius"
}
},
"required": ["city", "date"]
}
}
}
]
}API 调用示例
import openai
from typing import List, Dict, Any
def call_with_tools():
"""带工具调用的 API 调用"""
response = openai.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "user",
"content": "北京明天天气怎么样？"
}
],
tools=[
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市和日期的天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，需要使用中文"
},
"date": {
"type": "string",
"description": "查询日期，格式为 YYYY-MM-DD"
}
},
"required": ["city", "date"]
}
}
}
],
tool_choice="auto" # auto 让模型决定是否调用工具
)
# 获取模型响应
message = response.choices[0].message
# 检查是否有工具调用
if message.tool_calls:
for tool_call in message.tool_calls:
print(f"调用的工具: {tool_call.function.name}")
print(f"参数: {tool_call.function.arguments}")
# 解析 JSON 参数
args = json.loads(tool_call.function.arguments)
return args
# 没有工具调用，直接返回文本
return {"text": message.content}二、工具定义设计规范
2.1 参数 Schema 设计原则
好的工具定义需要遵循以下原则：
"""
工具定义的最佳实践
"""
# 示例：一个订单查询工具的完整定义
ORDER_QUERY_TOOL = {
"name": "query_order",
"description": "查询用户的订单信息，包括订单状态、物流进度、金额明细等",
"parameters": {
"type": "object",
"properties": {
# 订单 ID：必填，精确查询
"order_id": {
"type": "string",
"description": "订单号，格式为 18 位数字，如：202401151234567890",
"pattern": "^\\d{18}$",
"examples": ["202401151234567890"]
},
# 用户 ID：必填，确保查询权限
"user_id": {
"type": "string",
"description": "用户 ID，从当前会话上下文获取"
},
# 查询类型：可选，控制返回信息范围
"query_type": {
"type": "string",
"description": "查询的信息类型",
"enum": ["basic", "detail", "logistics", "all"],
"default": "all",
"examples": ["basic", "detail", "logistics"]
},
# 时间范围：可选，用于物流查询
"date_range": {
"type": "object",
"description": "时间范围筛选",
"properties": {
"start": {
"type": "string",
"description": "开始日期，YYYY-MM-DD 格式"
},
"end": {
"type": "string",
"description": "结束日期，YYYY-MM-DD 格式"
}
}
}
},
"required": ["order_id", "user_id"] # 只强制必填最小集
},
# 安全相关配置
"security": {
"required_permissions": ["order:read"],
"rate_limit": "100/minute",
"timeout": 5
}
}2.2 工具描述规范
工具描述是模型理解何时该调用工具的关键依据。
"""
工具描述设计规范
"""
# ❌ 不好：过于模糊
BAD_DESCRIPTIONS = [
"查询数据",
"获取信息",
"处理请求",
"执行操作"
]
# ✅ 好：清晰描述用途、触发场景、限制条件
GOOD_DESCRIPTIONS = {
"get_weather": """
获取指定城市和日期的天气预报信息。
适用场景：
- 用户询问"明天天气怎么样"
- 用户询问某地气候特征
- 用户询问出行建议
限制条件：
- 仅支持国内主要城市
- 仅支持未来 7 天内的预报
- 不提供历史天气查询
""",
"search_products": """
在商品库中搜索符合条件的产品。
适用场景：
- 用户想找特定类型的产品
- 用户询问某产品的价格、规格
- 用户进行产品比较
搜索支持：
- 关键词搜索
- 分类筛选
- 价格区间筛选
- 排序（价格、销量、评分）
""",
"create_order": """
创建新订单，完成商品购买流程。
适用场景：
- 用户确认购买并下单
- 用户修改订单后重新提交
重要提示：
- 会产生真实扣款，必须用户明确确认
- 需要校验库存和价格变动
- 下单前应展示完整订单信息供用户确认
"""
}2.3 类型约束与验证
from typing import Any, Dict, List, Optional, Union
from pydantic import BaseModel, Field, field_validator
from datetime import datetime
import json
class ToolParameterValidator:
"""
工具参数验证器：确保模型生成的参数符合 schema 约束
"""
def __init__(self, schema: Dict[str, Any]):
self.schema = schema
self.properties = schema.get("parameters", {}).get("properties", {})
self.required = schema.get("parameters", {}).get("required", [])
def validate(self, arguments: Union[str, Dict]) -> tuple[bool, Optional[Dict], Optional[str]]:
"""
验证参数
Args:
arguments: 模型生成的参数（JSON 字符串或字典）
Returns:
(是否有效, 验证后的参数, 错误信息)
"""
# 解析 JSON 字符串
if isinstance(arguments, str):
try:
arguments = json.loads(arguments)
except json.JSONDecodeError as e:
return False, None, f"JSON 解析失败: {e}"
# 检查必填参数
for field in self.required:
if field not in arguments:
return False, None, f"缺少必填参数: {field}"
# 验证每个字段
validated = {}
for key, value in arguments.items():
if key not in self.properties:
# 忽略未知字段
continue
prop = self.properties[key]
is_valid, validated_value, error = self._validate_field(
key, value, prop
)
if not is_valid:
return False, None, f"参数 {key} 验证失败: {error}"
validated[key] = validated_value
return True, validated, None
def _validate_field(
self,
name: str,
value: Any,
schema: Dict
) -> tuple[bool, Any, Optional[str]]:
"""验证单个字段"""
expected_type = schema.get("type")
# 类型检查
type_checks = {
"string": lambda v: isinstance(v, str),
"number": lambda v: isinstance(v, (int, float)),
"integer": lambda v: isinstance(v, int),
"boolean": lambda v: isinstance(v, bool),
"array": lambda v: isinstance(v, list),
"object": lambda v: isinstance(v, dict)
}
if expected_type in type_checks:
if not type_checks[expected_type](value):
return False, None, f"类型错误，期望 {expected_type}"
# 枚举检查
if "enum" in schema:
if value not in schema["enum"]:
return False, None, f"值必须在 {schema['enum']} 中"
# 正则表达式检查
if "pattern" in schema and isinstance(value, str):
import re
if not re.match(schema["pattern"], value):
return False, None, f"格式不符合要求: {schema['pattern']}"
# 范围检查
if expected_type in ("number", "integer"):
if "minimum" in schema and value < schema["minimum"]:
return False, None, f"值不能小于 {schema['minimum']}"
if "maximum" in schema and value > schema["maximum"]:
return False, None, f"值不能大于 {schema['maximum']}"
return True, value, None
# Pydantic 模型方式（推荐）
class GetWeatherParams(BaseModel):
"""天气查询参数"""
city: str = Field(
..., # 必填
description="城市名称，必须使用中文",
min_length=2,
max_length=10
)
date: str = Field(
...,
description="查询日期，YYYY-MM-DD 格式"
)
unit: Literal["celsius", "fahrenheit"] = Field(
default="celsius",
description="温度单位"
)
@field_validator("date")
@classmethod
def validate_date(cls, v: str) -> str:
"""验证日期格式"""
try:
datetime.strptime(v, "%Y-%m-%d")
except ValueError:
raise ValueError("日期格式必须是 YYYY-MM-DD")
return v
@field_validator("city")
@classmethod
def validate_city(cls, v: str) -> str:
"""验证城市名称"""
valid_cities = ["北京", "上海", "广州", "深圳", "杭州", "成都"]
if v not in valid_cities:
raise ValueError(f"暂不支持该城市，支持的城市有：{', '.join(valid_cities)}")
return v三、工具注册与发现机制
3.1 工具注册中心
from typing import Dict, List, Optional, Callable, Any
from dataclasses import dataclass, field
from enum import Enum
import inspect
class ToolCategory(Enum):
"""工具分类"""
SEARCH = "search" # 搜索类
DATA = "data" # 数据类
ACTION = "action" # 操作类
UTILITY = "utility" # 工具类
@dataclass
class ToolMetadata:
"""工具元数据"""
name: str
description: str
category: ToolCategory
parameters_schema: Dict[str, Any]
handler: Callable
permissions: List[str] = field(default_factory=list)
rate_limit: str = "100/minute"
timeout: int = 30
tags: List[str] = field(default_factory=list)
# 依赖关系
dependencies: List[str] = field(default_factory=list)
# 执行约束
max_retries: int = 3
require_confirmation: bool = False
class ToolRegistry:
"""
工具注册中心：统一管理所有可用工具
功能：
1. 工具注册与注销
2. 工具发现与查询
3. 权限管理
4. 动态加载
"""
def __init__(self):
self._tools: Dict[str, ToolMetadata] = {}
self._categories: Dict[ToolCategory, List[str]] = {}
self._tags: Dict[str, List[str]] = {}
def register(
self,
name: str,
description: str,
category: ToolCategory,
parameters_schema: Dict[str, Any],
handler: Callable,
permissions: Optional[List[str]] = None,
tags: Optional[List[str]] = None,
**kwargs
) -> None:
"""
注册工具
Args:
name: 工具名称（唯一标识）
description: 工具描述
category: 工具分类
parameters_schema: 参数 JSON Schema
handler: 处理函数
permissions: 所需权限列表
tags: 标签列表，用于搜索
"""
if name in self._tools:
raise ValueError(f"工具 {name} 已存在")
metadata = ToolMetadata(
name=name,
description=description,
category=category,
parameters_schema=parameters_schema,
handler=handler,
permissions=permissions or [],
tags=tags or [],
**kwargs
)
self._tools[name] = metadata
# 更新分类索引
if category not in self._categories:
self._categories[category] = []
self._categories[category].append(name)
# 更新标签索引
for tag in (tags or []):
if tag not in self._tags:
self._tags[tag] = []
self._tags[tag].append(name)
def get(self, name: str) -> Optional[ToolMetadata]:
"""获取工具元数据"""
return self._tools.get(name)
def list_by_category(self, category: ToolCategory) -> List[ToolMetadata]:
"""按分类列出工具"""
tool_names = self._categories.get(category, [])
return [self._tools[name] for name in tool_names]
def search_by_tag(self, tag: str) -> List[ToolMetadata]:
"""按标签搜索工具"""
tool_names = self._tags.get(tag, [])
return [self._tools[name] for name in tool_names]
def search_by_keywords(self, keywords: List[str]) -> List[ToolMetadata]:
"""
按关键词搜索工具
搜索范围：名称、描述、标签
"""
results = []
for tool in self._tools.values():
text = f"{tool.name} {tool.description} {' '.join(tool.tags)}"
# 检查是否所有关键词都匹配
if all(kw.lower() in text.lower() for kw in keywords):
results.append(tool)
return results
def get_tools_schema(self) -> List[Dict[str, Any]]:
"""获取所有工具的 schema，用于 LLM 调用"""
return [
{
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.parameters_schema
}
}
for tool in self._tools.values()
]
def unregister(self, name: str) -> bool:
"""注销工具"""
if name not in self._tools:
return False
tool = self._tools[name]
# 从分类索引移除
if tool.category in self._categories:
self._categories[tool.category].remove(name)
# 从标签索引移除
for tag in tool.tags:
if tag in self._tags:
self._tags[tag].remove(name)
del self._tools[name]
return True
# 全局注册中心实例
_global_registry = ToolRegistry()
def register_tool(
category: ToolCategory,
permissions: Optional[List[str]] = None,
tags: Optional[List[str]] = None,
**kwargs
):
"""
装饰器：简化工具注册
用法：
@register_tool(category=ToolCategory.SEARCH, tags=["weather", "查询"])
def get_weather(city: str, date: str):
...
"""
def decorator(func: Callable) -> Callable:
# 从函数签名生成 schema
sig = inspect.signature(func)
params = sig.parameters
properties = {}
required = []
for name, param in params.items():
param_type = "string"
if param.annotation == int:
param_type = "integer"
elif param.annotation == float:
param_type = "number"
elif param.annotation == bool:
param_type = "boolean"
properties[name] = {
"type": param_type,
"description": param.name # TODO: 从 docstring 提取
}
if param.default is inspect.Parameter.empty:
required.append(name)
schema = {
"type": "object",
"properties": properties,
"required": required
}
# 获取函数描述
description = func.__doc__ or func.__name__
# 注册
_global_registry.register(
name=func.__name__,
description=description,
category=category,
parameters_schema=schema,
handler=func,
permissions=permissions,
tags=tags,
**kwargs
)
return func
return decorator3.2 动态工具加载
class DynamicToolLoader:
"""
动态工具加载器
支持从模块、配置、远程服务加载工具
"""
def __init__(self, registry: ToolRegistry):
self.registry = registry
def load_from_module(self, module_path: str) -> int:
"""
从 Python 模块加载工具
扫描模块中所有被 @register_tool 装饰的函数
"""
import importlib
module = importlib.import_module(module_path)
# 已经在装饰器中注册，只需验证
return len([
name for name in dir(module)
if not name.startswith('_')
])
def load_from_yaml(self, yaml_path: str) -> int:
"""
从 YAML 配置加载外部工具
适用于：API 包装为工具、HTTP 服务等
"""
import yaml
with open(yaml_path, 'r') as f:
config = yaml.safe_load(f)
count = 0
for tool_config in config.get("tools", []):
self._load_http_tool(tool_config)
count += 1
return count
def _load_http_tool(self, config: Dict[str, Any]) -> None:
"""将 HTTP API 包装为工具"""
import httpx
async def http_handler(**kwargs) -> Any:
async with httpx.AsyncClient() as client:
response = await client.request(
method=config["method"],
url=config["url"],
headers=config.get("headers", {}),
json=kwargs if config.get("body_type") == "json" else None,
params=kwargs if config.get("body_type") == "query" else None
)
return response.json()
self.registry.register(
name=config["name"],
description=config["description"],
category=ToolCategory(config["category"]),
parameters_schema=config["parameters"],
handler=http_handler,
tags=config.get("tags", [])
)四、调用链路与编排
4.1 完整调用链路
┌─────────────────────────────────────────────────────────────────────────────┐
│ Tool Calling 完整调用链路 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 用户请求入口 │ │
│ │ "帮我查下我的订单状态" │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 1. 意图识别与工具匹配 │ │
│ │ 判断用户请求需要调用哪些工具，构建调用计划 │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 2. 参数提取与验证 │ │
│ │ 从用户请求中提取参数，验证类型和约束 │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 3. 权限检查 │ │
│ │ 验证用户是否有权限调用该工具 │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 4. 工具执行 │ │
│ │ 调用实际的处理函数，支持异步和重试 │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 5. 结果处理 │ │
│ │ 处理执行结果，可能触发二次调用或进入生成阶段 │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 6. 结果注入与生成 │ │
│ │ 将工具执行结果注入上下文，触发 LLM 生成最终回答 │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘4.2 工具调用框架代码
import asyncio
import json
from typing import Any, Dict, List, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
import logging
logger = logging.getLogger(__name__)
class ExecutionStatus(Enum):
"""执行状态"""
PENDING = "pending"
RUNNING = "running"
SUCCESS = "success"
FAILED = "failed"
TIMEOUT = "timeout"
@dataclass
class ToolCall:
"""工具调用请求"""
id: str # 调用唯一 ID
name: str # 工具名称
arguments: Dict[str, Any] # 调用参数
created_at: datetime = field(default_factory=datetime.now)
@dataclass
class ToolResult:
"""工具执行结果"""
call_id: str # 对应的调用 ID
status: ExecutionStatus # 执行状态
result: Any = None # 返回结果
error: Optional[str] = None # 错误信息
execution_time: float = 0.0 # 执行耗时（秒）
class ToolCallError(Exception):
"""工具调用相关错误"""
def __init__(self, message: str, code: str, recoverable: bool = True):
self.message = message
self.code = code
self.recoverable = recoverable
super().__init__(self.message)
class ToolExecutor:
"""
工具执行器：负责工具的注册、调用、错误处理
核心功能：
1. 工具注册与管理
2. 参数验证
3. 权限检查
4. 执行与重试
5. 超时控制
"""
def __init__(self, registry: ToolRegistry):
self.registry = registry
self._execution_history: List[ToolResult] = []
async def execute(
self,
tool_call: ToolCall,
context: Dict[str, Any]
) -> ToolResult:
"""
执行工具调用
Args:
tool_call: 工具调用请求
context: 执行上下文（包含用户信息、会话状态等）
Returns:
执行结果
"""
start_time = datetime.now()
try:
# 1. 获取工具元数据
tool_meta = self.registry.get(tool_call.name)
if not tool_meta:
raise ToolCallError(
f"工具不存在: {tool_call.name}",
code="TOOL_NOT_FOUND",
recoverable=False
)
# 2. 权限检查
self._check_permissions(tool_meta, context)
# 3. 参数验证
validated_args = self._validate_arguments(
tool_call.arguments,
tool_meta.parameters_schema
)
# 4. 执行（带重试）
result = await self._execute_with_retry(
tool_meta.handler,
validated_args,
tool_meta.max_retries,
tool_meta.timeout
)
# 5. 记录历史
execution_time = (datetime.now() - start_time).total_seconds()
tool_result = ToolResult(
call_id=tool_call.id,
status=ExecutionStatus.SUCCESS,
result=result,
execution_time=execution_time
)
self._execution_history.append(tool_result)
logger.info(f"工具 {tool_call.name} 执行成功，耗时 {execution_time}s")
return tool_result
except ToolCallError as e:
logger.error(f"工具执行错误: {e.message}")
return ToolResult(
call_id=tool_call.id,
status=ExecutionStatus.FAILED,
error=e.message,
execution_time=(datetime.now() - start_time).total_seconds()
)
except asyncio.TimeoutError:
logger.error(f"工具 {tool_call.name} 执行超时")
return ToolResult(
call_id=tool_call.id,
status=ExecutionStatus.TIMEOUT,
error="执行超时",
execution_time=tool_meta.timeout
)
except Exception as e:
logger.exception(f"工具 {tool_call.name} 执行异常")
return ToolResult(
call_id=tool_call.id,
status=ExecutionStatus.FAILED,
error=str(e),
execution_time=(datetime.now() - start_time).total_seconds()
)
def _check_permissions(
self,
tool_meta: ToolMetadata,
context: Dict[str, Any]
) -> None:
"""权限检查"""
user_permissions = context.get("user_permissions", [])
for required in tool_meta.permissions:
if required not in user_permissions:
raise ToolCallError(
f"缺少必要权限: {required}",
code="PERMISSION_DENIED",
recoverable=False
)
def _validate_arguments(
self,
arguments: Dict[str, Any],
schema: Dict[str, Any]
) -> Dict[str, Any]:
"""参数验证"""
validator = ToolParameterValidator(schema)
is_valid, validated, error = validator.validate(arguments)
if not is_valid:
raise ToolCallError(
f"参数验证失败: {error}",
code="INVALID_ARGUMENTS",
recoverable=True # 可恢复，模型可能重新生成参数
)
return validated
async def _execute_with_retry(
self,
handler: Callable,
arguments: Dict[str, Any],
max_retries: int,
timeout: int
) -> Any:
"""带重试的执行"""
last_error = None
for attempt in range(max_retries + 1):
try:
# 异步执行
if asyncio.iscoroutinefunction(handler):
return await asyncio.wait_for(
handler(**arguments),
timeout=timeout
)
else:
# 同步函数在线程池执行
loop = asyncio.get_event_loop()
return await asyncio.wait_for(
loop.run_in_executor(None, lambda: handler(**arguments)),
timeout=timeout
)
except Exception as e:
last_error = e
if attempt < max_retries:
logger.warning(
f"执行失败，重试中 ({attempt + 1}/{max_retries}): {e}"
)
await asyncio.sleep(2 ** attempt) # 指数退避
else:
raise
raise last_error
def get_execution_stats(self) -> Dict[str, Any]:
"""获取执行统计"""
if not self._execution_history:
return {"total": 0}
return {
"total": len(self._execution_history),
"success": len([
r for r in self._execution_history
if r.status == ExecutionStatus.SUCCESS
]),
"failed": len([
r for r in self._execution_history
if r.status == ExecutionStatus.FAILED
]),
"avg_execution_time": sum(r.execution_time for r in self._execution_history) / len(self._execution_history)
}
class ToolCallingOrchestrator:
"""
工具调用编排器：协调 LLM 和工具执行
负责：
1. 多轮工具调用编排
2. 结果注入与上下文管理
3. 调用链路追踪
"""
def __init__(
self,
llm, # LLM 客户端
executor: ToolExecutor,
max_turns: int = 10 # 最大调用轮次，防止无限循环
):
self.llm = llm
self.executor = executor
self.max_turns = max_turns
async def run(
self,
user_message: str,
context: Dict[str, Any],
conversation_history: Optional[List[Dict]] = None
) -> Dict[str, Any]:
"""
运行工具调用对话
Args:
user_message: 用户消息
context: 执行上下文
conversation_history: 对话历史（用于多轮对话）
Returns:
{
"final_message": str, # 最终回复
"tool_calls": List[Dict], # 所有工具调用记录
"turns": int # 使用的轮次
}
"""
# 构建消息历史
messages = conversation_history or []
messages.append({"role": "user", "content": user_message})
tool_calls_history = []
turns = 0
while turns < self.max_turns:
turns += 1
# 1. 调用 LLM（携带工具定义）
response = await self._call_llm(messages)
# 2. 检查是否有工具调用
if not response.tool_calls:
# 没有工具调用，返回 LLM 回复
messages.append({
"role": "assistant",
"content": response.content
})
return {
"final_message": response.content,
"tool_calls": tool_calls_history,
"turns": turns
}
# 3. 执行工具调用
for tool_call_data in response.tool_calls:
tool_call = ToolCall(
id=f"call_{turns}_{tool_call_data.id}",
name=tool_call_data.function.name,
arguments=json.loads(tool_call_data.function.arguments)
)
tool_calls_history.append({
"id": tool_call.id,
"name": tool_call.name,
"arguments": tool_call.arguments
})
# 执行工具
result = await self.executor.execute(tool_call, context)
# 4. 将结果注入上下文
messages.append({
"role": "assistant",
"tool_calls": [tool_call_data],
"content": None
})
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result.result) if result.result else result.error
})
tool_calls_history[-1]["result"] = result.result
tool_calls_history[-1]["status"] = result.status.value
tool_calls_history[-1]["execution_time"] = result.execution_time
# 检查是否需要继续
if not self._should_continue(messages):
break
# 超过最大轮次，强制结束
logger.warning(f"达到最大轮次 {self.max_turns}，强制结束")
return {
"final_message": "抱歉，任务过于复杂，已达到最大处理轮次。请简化您的问题或分步操作。",
"tool_calls": tool_calls_history,
"turns": turns,
"truncated": True
}
async def _call_llm(self, messages: List[Dict]) -> Any:
"""调用 LLM"""
tools = self.executor.registry.get_tools_schema()
response = await self.llm.ainvoke(
messages,
tools=tools if tools else None,
tool_choice="auto"
)
return response
def _should_continue(self, messages: List[Dict]) -> bool:
"""判断是否需要继续调用"""
# 检查最后一条消息是否有工具调用
last_msg = messages[-1]
# 如果最后是用户消息或 tool 消息的回复，可能需要继续
if last_msg.get("role") in ("user", "tool"):
return True
# 如果最后是助手消息且没有工具调用，结束
if last_msg.get("role") == "assistant" and not last_msg.get("tool_calls"):
return False
return True五、错误处理策略
5.1 错误分类与处理
from enum import Enum
from typing import Optional, Callable, Any
from dataclasses import dataclass
class ErrorCategory(Enum):
"""错误分类"""
# 参数错误
INVALID_ARGUMENT = "invalid_argument" # 参数格式/类型错误
MISSING_ARGUMENT = "missing_argument" # 缺少必填参数
# 执行错误
TOOL_NOT_FOUND = "tool_not_found" # 工具不存在
PERMISSION_DENIED = "permission_denied" # 权限不足
RATE_LIMIT = "rate_limit" # 限流
TIMEOUT = "timeout" # 超时
# 业务错误
BUSINESS_ERROR = "business_error" # 业务逻辑错误
EXTERNAL_SERVICE_ERROR = "external_error" # 外部服务错误
# 系统错误
SYSTEM_ERROR = "system_error" # 系统异常
@dataclass
class ErrorHandlingStrategy:
"""错误处理策略"""
category: ErrorCategory
retryable: bool # 是否可重试
recoverable: bool # 是否可恢复
fallback: Optional[str] # 降级方案
user_message: str # 用户友好的错误消息
ERROR_HANDLING_MATRIX = {
ErrorCategory.INVALID_ARGUMENT: ErrorHandlingStrategy(
category=ErrorCategory.INVALID_ARGUMENT,
retryable=False,
recoverable=True,
fallback=None,
user_message="参数格式有误，请重新描述您的需求"
),
ErrorCategory.MISSING_ARGUMENT: ErrorHandlingStrategy(
category=ErrorCategory.MISSING_ARGUMENT,
retryable=False,
recoverable=True,
fallback=None,
user_message="缺少必要信息，请补充完整您的请求"
),
ErrorCategory.PERMISSION_DENIED: ErrorHandlingStrategy(
category=ErrorCategory.PERMISSION_DENIED,
retryable=False,
recoverable=False,
fallback="escalate_to_human", # 转人工
user_message="您没有权限执行此操作，请联系管理员"
),
ErrorCategory.RATE_LIMIT: ErrorHandlingStrategy(
category=ErrorCategory.RATE_LIMIT,
retryable=True,
recoverable=True,
fallback=None,
user_message="操作过于频繁，请稍后再试"
),
ErrorCategory.TIMEOUT: ErrorHandlingStrategy(
category=ErrorCategory.TIMEOUT,
retryable=True,
recoverable=True,
fallback=None,
user_message="服务响应超时，请稍后重试"
),
ErrorCategory.EXTERNAL_SERVICE_ERROR: ErrorHandlingStrategy(
category=ErrorCategory.EXTERNAL_SERVICE_ERROR,
retryable=True,
recoverable=True,
fallback="cached_result", # 返回缓存结果
user_message="服务暂时不可用，已使用缓存数据"
),
}
class ToolErrorHandler:
"""
工具错误处理器
功能：
1. 错误分类
2. 策略选择
3. 错误恢复
"""
def __init__(self, orchestrator: ToolCallingOrchestrator):
self.orchestrator = orchestrator
def classify_error(self, error: Exception) -> ErrorCategory:
"""分类错误"""
if isinstance(error, ToolCallError):
code_mapping = {
"INVALID_ARGUMENTS": ErrorCategory.INVALID_ARGUMENT,
"TOOL_NOT_FOUND": ErrorCategory.TOOL_NOT_FOUND,
"PERMISSION_DENIED": ErrorCategory.PERMISSION_DENIED,
"RATE_LIMIT": ErrorCategory.RATE_LIMIT,
"TIMEOUT": ErrorCategory.TIMEOUT
}
return code_mapping.get(error.code, ErrorCategory.SYSTEM_ERROR)
# 根据异常类型分类
if isinstance(error, asyncio.TimeoutError):
return ErrorCategory.TIMEOUT
if "timeout" in str(error).lower():
return ErrorCategory.TIMEOUT
if "permission" in str(error).lower():
return ErrorCategory.PERMISSION_DENIED
if "rate limit" in str(error).lower():
return ErrorCategory.RATE_LIMIT
return ErrorCategory.EXTERNAL_SERVICE_ERROR
def get_strategy(self, error: Exception) -> ErrorHandlingStrategy:
"""获取错误处理策略"""
category = self.classify_error(error)
return ERROR_HANDLING_MATRIX.get(category)
async def handle(
self,
error: Exception,
context: Dict[str, Any]
) -> Dict[str, Any]:
"""
处理错误
Returns:
{
"recovered": bool,
"message": str,
"action": Optional[str] # 后续动作
}
"""
strategy = self.get_strategy(error)
if not strategy:
return {
"recovered": False,
"message": "发生未知错误",
"action": "escalate_to_human"
}
# 尝试降级
if strategy.fallback and strategy.recoverable:
result = await self._execute_fallback(
strategy.fallback,
context
)
if result:
return {
"recovered": True,
"message": strategy.user_message,
"action": "use_fallback",
"fallback_data": result
}
return {
"recovered": strategy.recoverable,
"message": strategy.user_message,
"action": "escalate_to_human" if not strategy.recoverable else "retry"
}
async def _execute_fallback(
self,
fallback_type: str,
context: Dict[str, Any]
) -> Optional[Any]:
"""执行降级方案"""
if fallback_type == "cached_result":
# 返回缓存结果
cache_key = context.get("cache_key")
if cache_key:
return self._get_from_cache(cache_key)
elif fallback_type == "escalate_to_human":
# 记录并通知人工处理
self._notify_human(context)
return None
def _get_from_cache(self, key: str) -> Optional[Any]:
"""从缓存获取数据"""
# 实现缓存查询逻辑
pass
def _notify_human(self, context: Dict[str, Any]) -> None:
"""通知人工处理"""
# 实现通知逻辑（消息队列、工单系统等）
pass5.2 循环调用防护
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, List
@dataclass
class LoopDetectionConfig:
"""循环检测配置"""
max_consecutive_calls: int = 5 # 同一工具最大连续调用次数
max_total_calls: int = 20 # 单次对话最大总调用次数
similar_action_threshold: float = 0.8 # 相似动作判定阈值
time_window_seconds: int = 60 # 检测时间窗口
class LoopDetector:
"""
循环调用检测器
检测以下情况：
1. 同一工具反复调用
2. 相似参数反复调用
3. 调用次数超出限制
"""
def __init__(self, config: LoopDetectionConfig):
self.config = config
self._call_counts: Dict[str, List[Dict]] = defaultdict(list)
def record_call(self, tool_name: str, arguments: Dict) -> None:
"""记录一次调用"""
import time
self._call_counts[tool_name].append({
"arguments": arguments,
"timestamp": time.time()
})
def check_and_raise_if_loop(
self,
tool_name: str,
arguments: Dict
) -> None:
"""
检查是否形成循环，循环则抛出异常
Raises:
ToolCallError: 检测到循环调用
"""
import time
# 获取该工具的调用历史
history = self._call_counts.get(tool_name, [])
# 清理过期的记录
current_time = time.time()
history = [
h for h in history
if current_time - h["timestamp"] < self.config.time_window_seconds
]
self._call_counts[tool_name] = history
# 检查1：连续调用次数
if len(history) >= self.config.max_consecutive_calls:
raise ToolCallError(
f"检测到工具 {tool_name} 连续调用超过限制",
code="LOOP_DETECTED",
recoverable=False
)
# 检查2：相似参数
for h in history:
similarity = self._compute_similarity(arguments, h["arguments"])
if similarity > self.config.similar_action_threshold:
raise ToolCallError(
f"检测到相似参数重复调用 {tool_name}",
code="LOOP_DETECTED",
recoverable=True
)
# 检查3：总调用次数
total_calls = sum(len(calls) for calls in self._call_counts.values())
if total_calls >= self.config.max_total_calls:
raise ToolCallError(
"已达到最大工具调用次数限制",
code="MAX_CALLS_EXCEEDED",
recoverable=False
)
def _compute_similarity(
self,
args1: Dict,
args2: Dict
) -> float:
"""计算参数相似度"""
if not args1 or not args2:
return 0.0
common_keys = set(args1.keys()) & set(args2.keys())
if not common_keys:
return 0.0
matches = sum(1 for k in common_keys if args1[k] == args2[k])
return matches / len(common_keys)
def reset(self) -> None:
"""重置检测状态"""
self._call_counts.clear()六、安全机制
# 工具调用安全配置
security_config:
# 权限控制
permissions:
# 权限级别定义
levels:
- name: "read"
description: "只读权限"
allowed_operations: ["search", "query", "get"]
- name: "write"
description: "读写权限"
allowed_operations: ["read", "create", "update"]
- name: "admin"
description: "管理员权限"
allowed_operations: ["*"]
# 敏感工具定义
sensitive_tools:
- name: "delete_order"
require_confirmation: true
require_secondary_auth: true
log_level: "critical"
- name: "refund"
require_confirmation: true
audit_trail: true
- name: "send_email"
whitelist_recipients: true
# 沙箱执行
sandbox:
enabled: true
isolation_level: "process" # process / container / vm
resource_limits:
cpu_time: 5 # 秒
memory: 256 # MB
disk_io: 100 # MB
network: false # 是否允许网络请求
allowed_modules:
- "math"
- "json"
- "re"
- "datetime"
# 审计日志
audit:
enabled: true
log_all_calls: true
log_payloads: true
retention_days: 90
alert_on_anomaly: trueclass SecurityMiddleware:
"""
安全中间件：权限检查、沙箱执行、审计日志
"""
def __init__(
self,
executor: ToolExecutor,
permission_service,
audit_logger
):
self.executor = executor
self.permission_service = permission_service
self.audit_logger = audit_logger
async def execute_secure(
self,
tool_call: ToolCall,
context: Dict[str, Any]
) -> ToolResult:
"""
安全执行工具调用
"""
tool_meta = self.executor.registry.get(tool_call.name)
# 1. 审计日志
self.audit_logger.log_tool_call(
tool_name=tool_call.name,
arguments=tool_call.arguments,
user_id=context.get("user_id"),
session_id=context.get("session_id")
)
# 2. 权限检查
user_permissions = context.get("user_permissions", [])
self._verify_permissions(tool_meta, user_permissions)
# 3. 敏感操作确认
if tool_meta.require_confirmation:
if not context.get("confirmed"):
raise ToolCallError(
"此操作需要用户确认",
code="CONFIRMATION_REQUIRED",
recoverable=True
)
# 4. 沙箱执行（如果启用）
if self._should_sandbox(tool_meta):
return await self._execute_in_sandbox(tool_call, tool_meta)
# 5. 正常执行
return await self.executor.execute(tool_call, context)
def _verify_permissions(
self,
tool_meta: ToolMetadata,
user_permissions: List[str]
) -> None:
"""验证权限"""
for required in tool_meta.permissions:
if required not in user_permissions:
self.audit_logger.log_security_event(
event_type="permission_denied",
tool_name=tool_meta.name,
required_permission=required
)
raise ToolCallError(
f"缺少必要权限: {required}",
code="PERMISSION_DENIED",
recoverable=False
)
def _should_sandbox(self, tool_meta: ToolMetadata) -> bool:
"""判断是否需要沙箱执行"""
# 代码执行类工具需要沙箱
return tool_meta.category == ToolCategory.UTILITY
async def _execute_in_sandbox(
self,
tool_call: ToolCall,
tool_meta: ToolMetadata
) -> ToolResult:
"""在沙箱中执行"""
# 实现沙箱逻辑（可用 subprocess、docker 等）
pass七、完整示例：天气 + 提醒助手
"""
完整的 Tool Calling 示例：天气 + 提醒助手
"""
import asyncio
import json
from typing import List, Dict, Any, Optional
from openai import OpenAI
# ============== 工具定义 ==============
TOOLS = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": """获取指定城市的天气信息。
适用场景：
- 用户询问某地的天气
- 用户询问是否需要带伞/添衣
- 为出行提供天气建议
返回信息包括：温度、湿度、天气状况、紫外线指数、穿衣建议等。""",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，必须使用中文，如：北京、上海、杭州",
"enum": ["北京", "上海", "广州", "深圳", "杭州", "成都", "南京", "武汉"]
},
"date": {
"type": "string",
"description": "查询日期，YYYY-MM-DD 格式，或'今天'/'明天'/'后天'"
}
},
"required": ["city", "date"]
}
}
},
{
"type": "function",
"function": {
"name": "set_reminder",
"description": """创建日程提醒。
适用场景：
- 用户要求设置提醒
- 用户想在特定时间做某事
- 会议、约会、服药等提醒
设置成功后，会在指定时间通过系统通知用户。""",
"parameters": {
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "提醒标题，简明扼要描述事项"
},
"datetime": {
"type": "string",
"description": "提醒时间，YYYY-MM-DD HH:MM 格式"
},
"repeat": {
"type": "string",
"description": "重复规则",
"enum": ["不重复", "每天", "每周", "每月"],
"default": "不重复"
},
"notes": {
"type": "string",
"description": "备注信息，可选"
}
},
"required": ["title", "datetime"]
}
}
}
]
# ============== 工具实现 ==============
def get_weather(city: str, date: str) -> Dict[str, Any]:
"""获取天气数据（模拟实现）"""
weather_data = {
"北京": {"temp": 15, "condition": "晴", "humidity": 45, "suggestion": "适宜户外活动"},
"上海": {"temp": 22, "condition": "多云", "humidity": 65, "suggestion": "记得带伞"},
"广州": {"temp": 28, "condition": "雷阵雨", "humidity": 80, "suggestion": "避免外出"},
"深圳": {"temp": 26, "condition": "晴", "humidity": 55, "suggestion": "紫外线较强"},
"杭州": {"temp": 20, "condition": "阴", "humidity": 60, "suggestion": "适合漫步西湖"},
"成都": {"temp": 18, "condition": "小雨", "humidity": 75, "suggestion": "建议穿外套"},
}
data = weather_data.get(city, {"temp": 20, "condition": "晴", "humidity": 50, "suggestion": "天气宜人"})
return {
"city": city,
"date": date,
"temperature": f"{data['temp']}°C",
"condition": data["condition"],
"humidity": f"{data['humidity']}%",
"suggestion": data["suggestion"]
}
def set_reminder(title: str, datetime: str, repeat: str = "不重复", notes: str = "") -> Dict[str, Any]:
"""创建提醒（模拟实现）"""
return {
"success": True,
"reminder_id": f"rem_{hash(title + datetime) % 100000}",
"title": title,
"datetime": datetime,
"repeat": repeat,
"notes": notes,
"message": f"已设置提醒：{title}，时间是 {datetime}"
}
# 工具函数映射
TOOL_IMPLEMENTATIONS = {
"get_weather": get_weather,
"set_reminder": set_reminder
}
# ============== 执行器 ==============
class WeatherReminderExecutor:
"""天气+提醒执行器"""
def __init__(self):
self.tools = {t["function"]["name"]: t["function"] for t in TOOLS}
def execute(self, name: str, arguments: Dict[str, Any]) -> Any:
"""执行工具调用"""
if name not in self.TOOL_IMPLEMENTATIONS:
raise ValueError(f"未知工具: {name}")
handler = TOOL_IMPLEMENTATIONS[name]
return handler(**arguments)
async def run_conversation():
"""运行对话示例"""
client = OpenAI()
executor = WeatherReminderExecutor()
messages = [
{
"role": "system",
"content": """你是一个智能助手，可以查询天气和设置提醒。
当用户询问天气时，先调用 get_weather 获取信息，然后结合信息给出建议。
当用户要求设置提醒时，先调用 set_reminder 完成设置。
始终使用中文回复，保持友好专业的语气。"""
}
]
# 用户输入
user_input = "北京明天天气怎么样？如果不下雨的话，帮我设置一个明天上午9点去健身的提醒"
messages.append({"role": "user", "content": user_input})
print(f"👤 用户: {user_input}\n")
# 多轮对话循环
max_turns = 5
for turn in range(max_turns):
# 调用 LLM
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=TOOLS,
tool_choice="auto"
)
assistant_msg = response.choices[0].message
# 检查工具调用
if assistant_msg.tool_calls:
messages.append({
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": tc.id,
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments
}
}
for tc in assistant_msg.tool_calls
]
})
# 执行工具并注入结果
for tc in assistant_msg.tool_calls:
tool_name = tc.function.name
args = json.loads(tc.function.arguments)
print(f"🔧 调用工具: {tool_name}")
print(f" 参数: {args}")
result = executor.execute(tool_name, args)
print(f" 结果: {result}\n")
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": json.dumps(result, ensure_ascii=False)
})
else:
# 最终回复
messages.append({
"role": "assistant",
"content": assistant_msg.content
})
print(f"🤖 助手: {assistant_msg.content}")
break
if __name__ == "__main__":
asyncio.run(run_conversation())运行结果示例
👤 用户: 北京明天天气怎么样？如果不下雨的话，帮我设置一个明天上午9点去健身的提醒
🔧 调用工具: get_weather
参数: {'city': '北京', 'date': '明天'}
结果: {'city': '北京', 'date': '明天', 'temperature': '15°C', 'condition': '晴', 'humidity': '45%', 'suggestion': '适宜户外活动'}
🔧 调用工具: set_reminder
参数: {'title': '去健身', 'datetime': '明天 09:00'}
结果: {'success': True, 'reminder_id': 'rem_12345', 'message': '已设置提醒：去健身，时间是 明天 09:00'}
🤖 助手: 明天北京的天气非常不错！☀️
🌡️ 温度：15°C
🌤️ 天气状况：晴
💧 湿度：45%
👕 穿衣建议：适宜户外活动
好消息是明天天气晴朗，非常适合去健身！我已经帮您设置好了提醒：
📅 提醒详情
- 事项：去健身
- 时间：明天上午 9:00
- 提醒ID：rem_12345
祝您健身愉快！💪 记得做好热身运动哦~八、总结
Tool Calling 是构建复杂 AI Agent 的核心技术能力，其工程化水平直接影响系统的可靠性和用户体验。
核心要点回顾：
工具定义是关键：好的 schema + 描述 = 模型正确调用的基础
参数验证不可少：防止模型生成无效参数导致执行失败
错误处理要分级：不同错误采用不同策略，提升用户体验
安全机制要完善：权限控制、沙箱执行、审计日志缺一不可
循环调用要防护：防止 Agent 陷入无限循环
最佳实践清单：
工具描述要详细，包含适用场景和限制条件
参数使用 enum 约束可能的值，减少歧义
必填参数只放最小集，降低调用门槛
敏感操作需要用户二次确认
始终记录审计日志，便于问题排查
实现重试机制，但设置最大重试次数
检测循环调用，防止系统卡死
评论区