深入理解 Function Calling：让大模型连接真实世界
7251 字
36 分钟
深入理解 Function Calling：让大模型连接真实世界
1. Function Calling 的引入
1.1 为什么需要 Function Calling？
纯粹的大语言模型本质上是一个封闭的文字接龙机器：
- 只能输出文本
- 不能查询实时天气
- 不能访问数据库
- 不能执行任何代码
- 不能与外部系统交互
这就像一个博学但与世隔绝的图书馆管理员——他知道很多陈旧的知识，却无法帮你做实事。
Function Calling 改变了这一切：
让 LLM 能够”调用”外部函数，将自然语言意图转化为结构化的函数调用，从而连接真实的数字世界。
1.2 Function Calling 的革命性意义
| 维度 | 纯 LLM | LLM + Function Calling |
|---|---|---|
| 数据时效性 | 训练截止 | 实时 |
| 知识范围 | 训练数据 | 任何可调用的服务 |
| 操作能力 | 仅生成文本 | 触发任何 API |
| 准确性 | 易幻觉 | 基于真实数据 |
| 个性化 | 通用 | 可定制业务逻辑 |
1.3 Function Calling 发展简史
2023.06: OpenAI 发布 Function Calling (GPT-3.5/4) │ 标志 LLM 工具调用时代的开始 │2023.11: GPTs 与 Assistants API │ 标准化工具集成 │2023-2024: Anthropic Claude Tool Use │ 类似机制，更精细控制 │2024.05: Google Gemini Function Calling │ 多模态 + 工具调用 │2024-2025: 标准化趋势 │ - OpenAI GPT-4o/4-turbo 优化调用 │ - Anthropic 推出 Prompt Caching for tools │ - JSON Schema 成为事实标准 │ - 多模态工具（Vision, Audio） │2025+: 完全 Agent 时代 │ - Agent 自主调用多工具 │ - 工具调用 + 长上下文 │ - 工具自我发现与组合1.4 与之前手写解析的对比
# 以前: 字符串解析（脆弱）prompt = """可用工具:- Search[查询关键词]- Calculator[数学表达式]
请使用 Search 查询北京天气..."""response = llm.generate(prompt)# LLM 可能返回:# "Search[北京今天天气]"# "Action: Search[\"北京天气\"]"# "我建议使用 Search 工具，输入是: 北京天气" ← 多样!# 需要用正则表达式艰难解析
# 现在: Function Calling（标准）tools = [{ "type": "function", "function": { "name": "get_weather", "description": "获取指定城市的天气", "parameters": { "type": "object", "properties": { "city": {"type": "string"} }, "required": ["city"] } }}]response = client.chat.completions.create( model="gpt-4", messages=messages, tools=tools)# 返回标准 JSON, 立即可用!2. Function Calling 基础
2.1 核心概念
Function Calling 的三大要素：
1. Tool Schema (工具模式) ↓ 用 JSON Schema 定义工具 ↓2. LLM Reasoning (LLM 推理) ↓ LLM 决定是否需要调用工具、调用哪个 ↓3. Tool Execution (工具执行) ↓ 程序执行工具，将结果返回给 LLM2.2 基本工作流程
┌─────────────────────────────────────────────────────────────────┐│ Function Calling 流程 │├─────────────────────────────────────────────────────────────────┤│ ││ 用户: "北京今天天气怎么样？" ││ ││ ↓ ││ ││ 1. 应用程序准备消息和工具定义 ││ messages = [{"role": "user", "content": "..."}] ││ tools = [{工具定义}] ││ ││ ↓ ││ ││ 2. 调用 LLM API ││ response = client.chat(messages, tools) ││ ││ ↓ ││ ││ 3. LLM 决定: 需要调用 get_weather(city="北京") ││ response.message.tool_calls = [...] ││ ││ ↓ ││ ││ 4. 应用程序执行工具 ││ weather = get_weather(city="北京") ││ ││ ↓ ││ ││ 5. 将工具结果返回给 LLM ││ messages.append({role: "tool", content: weather}) ││ ││ ↓ ││ ││ 6. LLM 根据工具结果生成最终回答 ││ response = client.chat(messages) ││ response.message.content = "北京今天是晴天..." ││ │└─────────────────────────────────────────────────────────────────┘2.3 完整代码示例
import jsonfrom openai import OpenAI
# 1. 定义工具tools = [ { "type": "function", "function": { "name": "get_weather", "description": "获取指定城市的天气信息", "parameters": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称，例如 '北京', 'Shanghai'" }, "unit": { "type": "string", "enum": ["celsius", "fahrenheit"], "description": "温度单位" } }, "required": ["city"] } } }, { "type": "function", "function": { "name": "search_restaurants", "description": "搜索餐厅信息", "parameters": { "type": "object", "properties": { "location": {"type": "string"}, "cuisine": {"type": "string"}, "price_range": {"type": "string"} }, "required": ["location"] } } }]
# 2. 实现工具函数def get_weather(city: str, unit: str = "celsius") -> str: """模拟天气查询""" weather_data = { "北京": {"temp": 25, "condition": "晴"}, "上海": {"temp": 28, "condition": "多云"}, } data = weather_data.get(city, {"temp": "未知", "condition": "未知"}) temp = data["temp"] if unit == "fahrenheit" and isinstance(temp, int): temp = temp * 9/5 + 32 return json.dumps({ "city": city, "temperature": temp, "condition": data["condition"], "unit": unit }, ensure_ascii=False)
def search_restaurants(location: str, cuisine: str = "") -> str: """模拟餐厅搜索""" restaurants = [ {"name": "北京烤鸭店", "rating": 4.8, "cuisine": "京菜"}, {"name": "海底捞", "rating": 4.7, "cuisine": "火锅"}, ] if cuisine: restaurants = [r for r in restaurants if cuisine in r["cuisine"]] return json.dumps(restaurants, ensure_ascii=False)
# 3. 注册工具TOOL_FUNCTIONS = { "get_weather": get_weather, "search_restaurants": search_restaurants,}
# 4. 主对话循环client = OpenAI(api_key="your-key")
def chat_with_tools(user_message: str) -> str: """带 Function Calling 的对话""" messages = [{"role": "user", "content": user_message}]
# 第一轮: 让 LLM 决定调用哪些工具 response = client.chat.completions.create( model="gpt-4", messages=messages, tools=tools, tool_choice="auto" # LLM 自动决定 )
assistant_message = response.choices[0].message messages.append(assistant_message)
# 检查是否需要调用工具 if assistant_message.tool_calls: # 执行所有工具调用 for tool_call in assistant_message.tool_calls: function_name = tool_call.function.name function_args = json.loads(tool_call.function.arguments)
# 调用工具 function_response = TOOL_FUNCTIONS[function_name](**function_args)
# 将结果加入消息 messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": function_response, })
# 第二轮: 让 LLM 基于工具结果生成回答 second_response = client.chat.completions.create( model="gpt-4", messages=messages ) return second_response.choices[0].message.content
# 不需要工具调用 return assistant_message.content
# 5. 使用result = chat_with_tools("北京今天天气怎么样？")print(result)3. JSON Schema 详解
3.1 JSON Schema 基础
JSON Schema 是 JSON 数据的结构化定义语言，是 Function Calling 的通用语言：
{ "type": "object", "properties": { "name": {"type": "string"}, "age": {"type": "integer", "minimum": 0, "maximum": 150}, "email": {"type": "string", "format": "email"}, "interests": {"type": "array", "items": {"type": "string"}} }, "required": ["name", "age"]}3.2 支持的数据类型
class JSONSchemaTypes: """JSON Schema 支持的主要类型"""
PRIMITIVE = { 'string': '文本字符串', 'integer': '整数', 'number': '数字（整数或浮点数）', 'boolean': '布尔值', 'null': '空值' }
COMPOSITE = { 'array': '数组', 'object': '对象' }
FORMATS = { 'date-time': 'RFC3339 日期时间', 'date': 'ISO 日期', 'email': '邮箱地址', 'uri': 'URI 地址', 'uuid': 'UUID' }3.3 复杂的工具 Schema 示例
# 示例 1: 创建订单工具create_order_schema = { "type": "function", "function": { "name": "create_order", "description": "创建一个新订单", "parameters": { "type": "object", "properties": { "customer_id": { "type": "string", "description": "客户 ID" }, "items": { "type": "array", "description": "订单商品列表", "items": { "type": "object", "properties": { "product_id": {"type": "string"}, "quantity": { "type": "integer", "minimum": 1, "description": "购买数量" }, "price": { "type": "number", "minimum": 0, "description": "单价" } }, "required": ["product_id", "quantity"] } }, "shipping_address": { "type": "object", "properties": { "street": {"type": "string"}, "city": {"type": "string"}, "postal_code": {"type": "string"}, "country": {"type": "string"} }, "required": ["street", "city", "country"] }, "payment_method": { "type": "string", "enum": ["credit_card", "alipay", "wechat_pay"], "description": "支付方式" }, "notes": { "type": "string", "description": "订单备注（可选）" } }, "required": ["customer_id", "items", "shipping_address", "payment_method"] } }}
# 示例 2: 多重安全约束search_flights_schema = { "type": "function", "function": { "name": "search_flights", "description": "搜索航班", "parameters": { "type": "object", "properties": { "origin": { "type": "string", "description": "出发地 IATA 代码", "pattern": "^[A-Z]{3}$" }, "destination": { "type": "string", "description": "目的地 IATA 代码", "pattern": "^[A-Z]{3}$" }, "departure_date": { "type": "string", "format": "date", "description": "出发日期 YYYY-MM-DD" }, "passengers": { "type": "integer", "minimum": 1, "maximum": 9, "default": 1 }, "class": { "type": "string", "enum": ["economy", "premium_economy", "business", "first"], "default": "economy" }, "max_price": { "type": "integer", "description": "最高价格（可选）", "minimum": 0 } }, "required": ["origin", "destination", "departure_date"] } }}3.4 Schema 设计技巧
class SchemaDesignTips: """JSON Schema 设计的最佳实践"""
TIPS = { 'description': '为每个字段提供清晰的描述', 'enum': '枚举值限制选项', 'required': '明确必需参数', 'default': '提供默认值减少误解', 'min/max': '数值范围限制', 'pattern': '字符串格式约束', 'example': '提供示例值', 'nullable': '明确允许为空', }
EXAMPLES = { 'before': { "name": "get_user", "parameters": { "type": "object", "properties": { "id": {"type": "string"} } } }, 'after': { "name": "get_user", "description": "根据用户 ID 获取用户信息", "parameters": { "type": "object", "properties": { "id": { "type": "string", "description": "用户的唯一标识符，UUID 格式", "pattern": "^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", "example": "550e8400-e29b-41d4-a716-446655440000" } }, "required": ["id"] } } }4. 主流平台的 Function Calling
4.1 OpenAI
from openai import OpenAI
client = OpenAI()
# 基本调用response = client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": "上海天气如何？"}], tools=[ { "type": "function", "function": { "name": "get_weather", "description": "获取天气", "parameters": { "type": "object", "properties": { "city": {"type": "string"} }, "required": ["city"] } } } ], tool_choice="auto", # auto / none / specific function parallel_tool_calls=True, # 允许并行调用 temperature=0.0)
# 处理响应message = response.choices[0].messageif message.tool_calls: for tool_call in message.tool_calls: print(f"调用: {tool_call.function.name}") print(f"参数: {tool_call.function.arguments}") print(f"ID: {tool_call.id}")OpenAI 特有的功能：
# 1. 强制调用特定函数tool_choice = {"type": "function", "function": {"name": "get_weather"}}
# 2. 并行工具调用response = client.chat.completions.create( model="gpt-4-turbo", messages=messages, tools=tools, parallel_tool_calls=True # GPT-4 支持一次返回多个)
# 3. 严格模式 (Structured Outputs)response = client.chat.completions.create( model="gpt-4o-2024-08-06", messages=messages, response_format={ "type": "json_schema", "json_schema": { "name": "weather_response", "schema": { "type": "object", "properties": { "city": {"type": "string"}, "temperature": {"type": "number"}, "conditions": {"type": "array", "items": {"type": "string"}} }, "required": ["city", "temperature"], "additionalProperties": False }, "strict": True } })4.2 Anthropic Claude
import anthropic
client = anthropic.Anthropic()
response = client.messages.create( model="claude-3-5-sonnet-20240620", max_tokens=1024, tools=[ { "name": "get_weather", "description": "获取天气信息", "input_schema": { "type": "object", "properties": { "city": {"type": "string"}, "unit": { "type": "string", "enum": ["celsius", "fahrenheit"] } }, "required": ["city"] } } ], messages=[{"role": "user", "content": "北京今天天气怎么样？"}])
# Claude 使用 content blocksfor block in response.content: if block.type == "tool_use": print(f"调用: {block.name}") print(f"输入: {block.input}") print(f"ID: {block.id}") elif block.type == "text": print(f"文本: {block.text}")Claude 特色功能：
# 1. 精细的工具控制tools = [{ "name": "read_file", "description": "读取文件内容", "input_schema": { "type": "object", "properties": { "path": {"type": "string"}, "start_line": {"type": "integer"}, "end_line": {"type": "integer"} }, "required": ["path"] }}]
# 2. 添加工具结果messages = [ {"role": "user", "content": "读取 test.py"}, {"role": "assistant", "content": [ {"type": "tool_use", "id": "toolu_xxx", "name": "read_file", "input": {"path": "test.py"}} ]}, {"role": "user", "content": [ {"type": "tool_result", "tool_use_id": "toolu_xxx", "content": "print('hello world')"} ]}]
# 3. 指定工具选择response = client.messages.create( model="claude-3-5-sonnet-20240620", tools=tools, tool_choice={"type": "any"}, # 必须调用工具 messages=messages)4.3 Google Gemini
import google.generativeai as genai
genai.configure(api_key="your-key")
model = genai.GenerativeModel( model_name="gemini-1.5-pro", tools=[ { "function_declarations": [ { "name": "get_weather", "description": "获取天气", "parameters": { "type": "object", "properties": { "city": {"type": "string"} }, "required": ["city"] } } ] } ])
response = model.generate_content("北京天气如何？")
# 处理 function callfor part in response.parts: if part.function_call: print(f"调用: {part.function_call.name}") print(f"参数: {dict(part.function_call.args)}")4.4 对比各平台
| 特性 | OpenAI | Anthropic Claude | Google Gemini |
|---|---|---|---|
| 模型 | GPT-4/4o/4-turbo | Claude 3.5 | Gemini 1.5 |
| 并行调用 | ✓ | ✓ | ✓ |
| 强制调用 | tool_choice | tool_choice | tool_config |
| Schema 严格性 | 严格模式 | 严格模式 | 严格模式 |
| 多模态工具 | Vision | Vision | Vision + Audio |
| 流式响应 | ✓ | ✓ | ✓ |
| 工具缓存 | - | ✓ (Prompt caching) | - |
5. 高级特性
5.1 并行工具调用
LLM 可以在单次响应中调用多个工具：
def handle_parallel_calls(response): """处理并行工具调用""" message = response.choices[0].message
# OpenAI: message.tool_calls 是列表 if message.tool_calls: # 并发执行所有工具调用 import asyncio results = await asyncio.gather(*[ execute_tool_async(tc) for tc in message.tool_calls ])
# 将结果加入消息 for tool_call, result in zip(message.tool_calls, results): messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": str(result) })并行调用场景：
用户: "对比北京、上海、广州三地今天的天气和空气质量"
LLM 决定调用: get_weather(city="北京") get_weather(city="上海") get_weather(city="广州") get_air_quality(city="北京") get_air_quality(city="上海") get_air_quality(city="广州")
并行执行 → 一次性返回所有结果 → LLM 综合回答5.2 强制工具调用
# OpenAItool_choice = {"type": "function", "function": {"name": "calculator"}}
# Claudetool_choice = {"type": "tool", "name": "calculator"}
# 应用场景:# - 必须使用结构化输出# - 强制工具调用保证确定性# - 抽取信息（如 NER）5.3 嵌套工具调用
# 工具返回的结果可以作为下一个工具的输入
# 第一轮: 调用 search 获取用户search_result = search_users(query="张")
# 第二轮: 基于搜索结果创建订单create_order( customer_id=search_result[0]["id"], items=[{"product_id": "P001", "quantity": 2}])5.4 流式响应中的工具调用
def stream_with_tools(messages, tools): """流式处理工具调用""" stream = client.chat.completions.create( model="gpt-4", messages=messages, tools=tools, stream=True )
# 收集工具调用块 tool_calls = []
for chunk in stream: delta = chunk.choices[0].delta
# 流式文本 if delta.content: print(delta.content, end="")
# 工具调用 (流式累积) if delta.tool_calls: for tc in delta.tool_calls: # 累积参数 tool_calls.append({ 'id': tc.id, 'function': { 'name': tc.function.name, 'arguments': tc.function.arguments } })
return tool_calls5.5 Structured Outputs（结构化输出）
OpenAI 的 Structured Outputs 让 LLM 直接返回符合 Schema 的 JSON：
# 不再使用 Function Calling，而是直接要求结构化输出
response = client.chat.completions.create( model="gpt-4o-2024-08-06", messages=[ {"role": "system", "content": "你是一个数据提取助手"}, {"role": "user", "content": "张三，35 岁，住在上海，是一个软件工程师"} ], response_format={ "type": "json_schema", "json_schema": { "name": "person", "strict": True, "schema": { "type": "object", "properties": { "name": {"type": "string"}, "age": {"type": "integer"}, "city": {"type": "string"}, "occupation": {"type": "string"} }, "required": ["name", "age", "city", "occupation"], "additionalProperties": False } } })
# 返回的是严格的 JSON, 100% 符合 schemaperson = json.loads(response.choices[0].message.content)6. 工具系统的工程设计
6.1 工具注册表
from typing import Callable, Dict, Any, List, Optionalfrom dataclasses import dataclassfrom functools import wrapsimport inspect
@dataclassclass Tool: """工具元数据""" name: str description: str func: Callable parameters_schema: Dict[str, Any] requires_auth: bool = False is_long_running: bool = False
def to_openai_format(self) -> Dict[str, Any]: """转换为 OpenAI 格式""" return { "type": "function", "function": { "name": self.name, "description": self.description, "parameters": self.parameters_schema } }
def to_anthropic_format(self) -> Dict[str, Any]: """转换为 Anthropic 格式""" return { "name": self.name, "description": self.description, "input_schema": self.parameters_schema }
def to_gemini_format(self) -> Dict[str, Any]: """转换为 Gemini 格式""" return { "name": self.name, "description": self.description, "parameters": self.parameters_schema }
class ToolRegistry: """工具注册表"""
def __init__(self): self.tools: Dict[str, Tool] = {}
def register(self, name: str = None, description: str = None, requires_auth: bool = False, is_long_running: bool = False): """装饰器注册工具""" def decorator(func: Callable): tool_name = name or func.__name__ tool_desc = description or func.__doc__ or ""
# 自动从函数签名生成 schema schema = self._generate_schema_from_func(func)
tool = Tool( name=tool_name, description=tool_desc, func=func, parameters_schema=schema, requires_auth=requires_auth, is_long_running=is_long_running )
self.tools[tool_name] = tool return func return decorator
def _generate_schema_from_func(self, func: Callable) -> Dict: """从函数自动生成 JSON Schema""" sig = inspect.signature(func) properties = {} required = []
type_mapping = { int: "integer", float: "number", bool: "boolean", str: "string", list: "array", dict: "object" }
for param_name, param in sig.parameters.items(): if param_name == 'self': continue
param_type = type_mapping.get(param.annotation, "string") properties[param_name] = { "type": param_type, "description": f"参数 {param_name}" }
if param.default is param.empty: required.append(param_name)
return { "type": "object", "properties": properties, "required": required }
def get(self, name: str) -> Optional[Tool]: return self.tools.get(name)
def to_openai_tools(self) -> List[Dict]: """导出为 OpenAI 格式""" return [tool.to_openai_format() for tool in self.tools.values()]
# 使用示例registry = ToolRegistry()
@registry.register(description="获取指定城市的天气", requires_auth=True)def get_weather(city: str, unit: str = "celsius") -> str: """获取天气信息""" return f"{city} 的天气是..."
@registry.register(description="搜索餐厅", is_long_running=True)def search_restaurants(location: str, cuisine: str = "") -> str: """搜索餐厅""" return f"在 {location} 找到餐厅..."6.2 工具调用器
class ToolCaller: """统一的工具调用器"""
def __init__(self, registry: ToolRegistry, auth_provider=None): self.registry = registry self.auth_provider = auth_provider
def execute(self, name: str, arguments: Dict[str, Any], context: Optional[Dict] = None) -> str: """执行工具调用""" tool = self.registry.get(name) if not tool: return f"错误: 未知工具 {name}"
# 1. 认证 if tool.requires_auth and self.auth_provider: if not self.auth_provider.authenticate(context): return "错误: 未认证"
# 2. 参数验证 validated_args = self._validate_arguments(tool, arguments)
# 3. 执行 try: result = tool.func(**validated_args) return str(result) except Exception as e: return f"执行错误: {e}"
async def execute_async(self, name: str, arguments: Dict[str, Any], context: Optional[Dict] = None) -> str: """异步执行""" # 异步执行逻辑 ...
def _validate_arguments(self, tool: Tool, arguments: Dict) -> Dict: """验证参数""" # 1. JSON Schema 验证 try: import jsonschema jsonschema.validate(arguments, tool.parameters_schema) except jsonschema.ValidationError as e: raise ValueError(f"参数验证失败: {e}")
return arguments6.3 调用追踪
class ToolCallTracer: """工具调用追踪器"""
def __init__(self): self.calls = []
def record(self, tool_name: str, arguments: Dict, result: str, duration_ms: float, success: bool): """记录调用""" self.calls.append({ 'timestamp': time.time(), 'tool': tool_name, 'arguments': arguments, 'result_preview': result[:200], 'duration_ms': duration_ms, 'success': success })
def get_stats(self) -> Dict: """获取统计信息""" if not self.calls: return {}
total = len(self.calls) success = sum(1 for c in self.calls if c['success']) avg_duration = sum(c['duration_ms'] for c in self.calls) / total
return { 'total_calls': total, 'success_rate': success / total, 'avg_duration_ms': avg_duration, 'tool_distribution': self._get_tool_distribution() }
def _get_tool_distribution(self) -> Dict[str, int]: """工具使用分布""" dist = {} for call in self.calls: tool = call['tool'] dist[tool] = dist.get(tool, 0) + 1 return dist7. 错误处理与重试
7.1 错误类型
class ToolCallError(Exception): """工具调用错误基类""" pass
class ArgumentValidationError(ToolCallError): """参数验证失败""" pass
class ToolNotFoundError(ToolCallError): """工具不存在""" pass
class ToolExecutionError(ToolCallError): """工具执行失败""" pass
class RateLimitError(ToolCallError): """速率限制""" pass
class AuthenticationError(ToolCallError): """认证失败""" pass
class TimeoutError(ToolCallError): """执行超时""" pass7.2 重试策略
import asyncioimport randomfrom typing import TypeVar, Callable, Awaitable
T = TypeVar('T')
class RetryStrategy: """重试策略"""
@staticmethod async def exponential_backoff( func: Callable[..., Awaitable[T]], max_retries: int = 3, base_delay: float = 1.0, max_delay: float = 60.0, exceptions: tuple = (Exception,) ) -> T: """指数退避重试""" for attempt in range(max_retries): try: return await func() except exceptions as e: if attempt == max_retries - 1: raise
# 计算延迟（指数退避 + jitter） delay = min(base_delay * (2 ** attempt), max_delay) delay = delay * (0.5 + random.random())
print(f"重试 {attempt + 1}/{max_retries}, 等待 {delay:.2f}s") await asyncio.sleep(delay)
raise Exception("无法处理")
@staticmethod def should_retry(error: Exception) -> bool: """判断是否应该重试""" # 网络错误、超时: 重试 if isinstance(error, (ConnectionError, TimeoutError)): return True
# 速率限制: 重试 if isinstance(error, RateLimitError): return True
# 参数错误: 不重试 if isinstance(error, ArgumentValidationError): return False
# 认证错误: 不重试 if isinstance(error, AuthenticationError): return False
return True7.3 容错的对话循环
class RobustFunctionCallingLoop: """容错的 Function Calling 循环"""
def __init__(self, client, tools, tool_caller, max_iterations=10): self.client = client self.tools = tools self.tool_caller = tool_caller self.max_iterations = max_iterations
def run(self, user_message: str) -> str: """主循环""" messages = [{"role": "user", "content": user_message}]
for iteration in range(self.max_iterations): try: # 1. 调用 LLM response = self.client.chat.completions.create( model="gpt-4", messages=messages, tools=self.tools, )
message = response.choices[0].message messages.append(message)
# 2. 检查工具调用 if not message.tool_calls: return message.content
# 3. 处理所有工具调用 for tool_call in message.tool_calls: try: # 执行工具 args = json.loads(tool_call.function.arguments) result = self.tool_caller.execute( name=tool_call.function.name, arguments=args ) except Exception as e: # 错误处理 result = json.dumps({ "error": str(e), "tool": tool_call.function.name, "suggestion": "请检查参数或重试" }, ensure_ascii=False)
# 加入消息 messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": result })
except Exception as e: # LLM 调用失败处理 print(f"LLM 调用失败: {e}") return f"抱歉，系统遇到错误: {e}"
return "达到最大迭代次数，未能完成任务"7.4 参数自愈
让 LLM 自动修正错误的参数：
class SelfHealingToolCaller: """自愈的工具调用器"""
def __init__(self, llm, tools): self.llm = llm self.tools = tools
def execute_with_healing(self, name, args, original_request): """带自愈的执行""" max_attempts = 3
for attempt in range(max_attempts): try: # 尝试执行 return self._execute(name, args) except ArgumentValidationError as e: if attempt == max_attempts - 1: raise
# 让 LLM 修正参数 args = self._ask_llm_to_fix(name, args, str(e), original_request)
raise Exception("无法自愈")
def _ask_llm_to_fix(self, name, args, error, original_request): """让 LLM 修正参数""" prompt = f""" 原始请求：{original_request} 工具：{name} 错误参数：{args} 错误信息：{error}
请修正参数以消除错误。返回 JSON 格式： """ response = self.llm.generate(prompt) return json.loads(response)8. 工具的权限和安全
8.1 权限分级
class ToolPermissionLevel: """工具权限级别"""
READ_ONLY = "read_only" # 只读操作 SAFE_WRITE = "safe_write" # 安全写入（如个人笔记） SENSITIVE = "sensitive" # 敏感操作（如支付） DESTRUCTIVE = "destructive" # 破坏性操作（如删除） ADMIN = "admin" # 管理权限
class PermissionManager: """权限管理器"""
PERMISSION_REQUIREMENTS = { 'get_weather': ToolPermissionLevel.READ_ONLY, 'send_email': ToolPermissionLevel.SAFE_WRITE, 'make_payment': ToolPermissionLevel.SENSITIVE, 'delete_account': ToolPermissionLevel.DESTRUCTIVE, 'modify_system': ToolPermissionLevel.ADMIN, }
def check_permission(self, tool_name: str, user_context: Dict) -> bool: """检查权限""" required_level = self.PERMISSION_REQUIREMENTS.get( tool_name, ToolPermissionLevel.SAFE_WRITE )
user_level = user_context.get('permission_level', ToolPermissionLevel.READ_ONLY)
# 权限等级对比 level_order = [ ToolPermissionLevel.READ_ONLY, ToolPermissionLevel.SAFE_WRITE, ToolPermissionLevel.SENSITIVE, ToolPermissionLevel.DESTRUCTIVE, ToolPermissionLevel.ADMIN, ]
return level_order.index(user_level) >= level_order.index(required_level)8.2 用户确认
class ConfirmationRequiredTool: """需要用户确认的工具"""
def __init__(self, tool, confirmation_message=None): self.tool = tool self.confirmation_message = confirmation_message or tool.description
def execute(self, **kwargs) -> str: """执行前请求确认""" confirmation = self._request_confirmation(kwargs) if not confirmation: return "用户取消了操作"
return self.tool.func(**kwargs)
def _request_confirmation(self, args): """请求用户确认""" print(f"\n⚠️ 确认执行以下操作？") print(f"工具: {self.tool.name}") print(f"参数: {args}")
response = input("输入 yes 确认: ") return response.lower() == 'yes'8.3 输入消毒
class SanitizedTool: """输入消毒的工具"""
DANGEROUS_PATTERNS = { 'sql_injection': [ r"';.*--", r"\b(DROP|DELETE|UPDATE|INSERT)\b.*\b(FROM|INTO|SET)\b", r"\bOR\b.*=", ], 'command_injection': [ r";\s*", r"\|\s*", r"&&\s*", r"`", r"\$\(", ], 'path_traversal': [ r"\.\./", r"\.\.\\", r"/etc/passwd", r"~", ] }
def execute_safe(self, name, args): """安全执行""" # 检查每个参数 for key, value in args.items(): if isinstance(value, str): self._check_injection(value, name, key)
return self.tool.func(**args)
def _check_injection(self, value, tool_name, param_name): """检查注入""" for category, patterns in self.DANGEROUS_PATTERNS.items(): for pattern in patterns: if re.search(pattern, value, re.IGNORECASE): raise SecurityError( f"检测到可能的 {category} 攻击: " f"工具={tool_name}, 参数={param_name}" )9. Function Calling 与 Agent
9.1 作为 Agent 引擎
class FunctionCallingAgent: """基于 Function Calling 的 Agent"""
def __init__(self, llm_client, tool_registry, system_prompt=None): self.client = llm_client self.tools = tool_registry.to_openai_tools() self.registry = tool_registry self.system_prompt = system_prompt or "你是一个有用的 AI 助手。"
def run(self, user_message: str, max_iterations=10) -> str: """Agent 主循环""" messages = [ {"role": "system", "content": self.system_prompt}, {"role": "user", "content": user_message} ]
for i in range(max_iterations): # 调用 LLM response = self.client.chat.completions.create( model="gpt-4", messages=messages, tools=self.tools, tool_choice="auto" )
message = response.choices[0].message messages.append(message)
# 检查是否完成 if message.content and not message.tool_calls: return message.content
# 执行工具调用 if message.tool_calls: for tc in message.tool_calls: args = json.loads(tc.function.arguments) result = self._safe_execute(tc.function.name, args)
messages.append({ "role": "tool", "tool_call_id": tc.id, "content": str(result) })
return "达到最大迭代次数"
def _safe_execute(self, name, args): """安全执行工具""" try: tool = self.registry.get(name) if not tool: return f"错误: 未知工具 {name}" return tool.func(**args) except Exception as e: return f"执行错误: {str(e)}"9.2 Function Calling vs ReAct
| 维度 | Function Calling | ReAct |
|---|---|---|
| 解析难度 | 原生支持 | 需要字符串解析 |
| 标准化 | 高度标准化 | 实现各异 |
| 多平台一致 | 较好（OpenAI/Anthropic） | 因实现而异 |
| Thought 可见 | ✗（隐藏） | ✓（可见） |
| 灵活度 | 中 | 高 |
| 性能 | 较快 | 稍慢 |
| 调试 | 较难 | 容易 |
10. Function Calling 的成本与性能
10.1 Token 成本分析
def estimate_cost(messages, tools, response): """估算 token 成本""" # 1. Token 计数 input_tokens = count_tokens(messages) tool_tokens = sum(count_tokens(json.dumps(tool)) for tool in tools) output_tokens = count_tokens(response.content or '')
# 2. 计算成本（GPT-4 假设） input_cost = (input_tokens + tool_tokens) * 0.00003 output_cost = output_tokens * 0.00006 total_cost = input_cost + output_cost
return { 'input_tokens': input_tokens, 'tool_tokens': tool_tokens, # 每次请求都要发送工具定义 'output_tokens': output_tokens, 'total_cost_usd': total_cost }工具定义的 token 开销：
工具数量 平均 Schema Token 每轮成本增加─────────────────────────────────────5 500 +$0.01510 1000 +$0.03020 2000 +$0.06050 5000 +$0.15010.2 优化策略
class OptimizedFunctionCalling: """性能优化的 Function Calling"""
def __init__(self, client, all_tools): self.client = client self.all_tools = all_tools
def smart_tool_selection(self, user_intent): """智能工具选择""" # 根据用户意图选择相关工具 relevant_tools = self._select_relevant_tools(user_intent) return relevant_tools
def _select_relevant_tools(self, intent): """只选择相关工具""" # 方法 1: 关键词匹配 keywords = intent.lower().split() relevant = [] for tool in self.all_tools: tool_text = json.dumps(tool).lower() if any(kw in tool_text for kw in keywords): relevant.append(tool)
# 方法 2: 嵌入相似度 # ... 使用 embedding 找出最相关的工具
return relevant[:10] # 限制最多 10 个
def tool_description_compression(self, tools): """工具描述压缩""" compressed = [] for tool in tools: # 压缩描述，减少 token tool_copy = tool.copy() tool_copy['function']['description'] = self._compress( tool['function']['description'] ) compressed.append(tool_copy) return compressed10.3 Anthropic Prompt Caching
Claude 支持工具的 Prompt Caching，大幅降低重复调用成本：
response = client.messages.create( model="claude-3-5-sonnet-20240620", max_tokens=1024, tools=[{ "name": "get_weather", "description": "获取天气", "input_schema": { "type": "object", "properties": { "city": {"type": "string"} }, "required": ["city"] } }], extra_headers={ # 标记工具可缓存 "anthropic-beta": "prompt-caching-2024-07-31" }, system=[ { "type": "text", "text": "你是天气预报员", "cache_control": {"type": "ephemeral"} } ], messages=[{"role": "user", "content": "查询天气"}])11. 测试 Function Calling
11.1 单元测试
import pytestfrom unittest.mock import Mock
class TestWeatherTool: """天气工具的测试"""
def test_get_weather_valid_city(self): result = get_weather(city="北京") assert "北京" in result assert "温度" in result or "weather" in result
def test_get_weather_invalid_city(self): # 处理未知城市 result = get_weather(city="火星") assert "未知" in result or "default" in result
def test_get_weather_unit_conversion(self): # 测试单位转换 result_c = get_weather(city="北京", unit="celsius") result_f = get_weather(city="北京", unit="fahrenheit") # 验证两种单位的差异 # ...
class TestFunctionCallingFlow: """Function Calling 流程测试"""
def test_llm_calls_correct_tool(self): """测试 LLM 调用正确的工具""" # Mock LLM 响应 mock_response = Mock() mock_response.choices[0].message.tool_calls = [ Mock( function=Mock( name="get_weather", arguments='{"city": "北京"}' ), id="call_123" ) ]
# 执行流程 # ...
def test_handles_no_tool_call(self): """测试不需要工具调用的场景""" # ...
def test_handles_tool_error(self): """测试工具执行错误""" # ...11.2 集成测试
class IntegrationTestFunctionCalling: """集成测试"""
def test_complete_conversation(self): """测试完整对话流程""" agent = FunctionCallingAgent(client, registry)
# 测试需要工具的查询 result = agent.run("北京今天天气怎么样？") assert "天气" in result
# 测试不需要工具的查询 result = agent.run("你好") assert len(result) > 0
# 测试多步骤任务 result = agent.run("查北京天气并推荐景点") assert "天气" in result assert "景点" in result11.3 模拟 LLM 响应
class MockLLMClient: """模拟 LLM 客户端用于测试"""
def __init__(self, scripted_responses): self.scripted_responses = scripted_responses self.call_count = 0
def chat(self, messages, tools, **kwargs): """返回预设的响应""" if self.call_count < len(self.scripted_responses): response = self.scripted_responses[self.call_count] self.call_count += 1 return response return Mock(content="测试完成")12. Function Calling vs 其他模式
12.1 与其他模式的对比
| 特性 | Function Calling | ReAct | RAG | Code Interpreter |
|---|---|---|---|---|
| 工具集成 | 原生 | 手动 | 通过检索 | 通过执行 |
| 结构化 | ✓ | ✗ | ✗ | ✗ |
| 可读性 | 低 | 高 | 中 | 中 |
| 灵活性 | 中 | 高 | 中 | 中 |
| 延迟 | 低 | 中 | 中 | 高 |
| 成本 | 中 | 高 | 低 | 高 |
12.2 何时使用 Function Calling
使用 Function Calling: ✓ 标准化工具集成 ✓ 高可靠性要求 ✓ 多平台支持 ✓ 流式响应 ✓ 与 ChatGPT 生态集成
考虑其他方案: ✗ 简单的 prompt 应用 ✗ 需要极度灵活的推理（如 Tree of Thoughts） ✗ 主要关注思考过程（用 ReAct）13. 实战最佳实践
13.1 工具设计原则
TOOL_DESIGN_PRINCIPLES = """1. 单一职责: 一个工具做一件事2. 清晰描述: description 明确说明何时使用3. 明确参数: 必填/可选清晰4. 错误处理: 返回结构化错误信息5. 输出格式: 一致的 JSON 格式6. 文档完善: 包含示例和边界条件"""
# 良好的工具设计示例good_tool = { "name": "get_weather", "description": ( "获取指定城市的当前天气。" "支持的城市包括：北京、上海、广州等。" "不要用于查询历史天气或天气预报。" ), "parameters": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称，例如'北京'" } }, "required": ["city"] }}13.2 Prompt 工程
SYSTEM_PROMPT_FOR_FUNCTION_CALLING = """你是一个 AI 助手，可以使用多种工具来帮助用户。
工具使用指南:1. 仔细分析用户意图2. 选择合适的工具3. 提供准确的参数4. 基于工具结果回答
如果不需要工具，可以直接回答。如果多个工具都相关，可以并行调用。"""13.3 调试技巧
DEBUGGING_TIPS = """1. 启用 verbose 日志: - 打印每个工具调用 - 记录参数和结果 - 跟踪 token 使用
2. 使用 LangSmith / OpenAI Traces: - 可视化调用链 - 分析性能瓶颈
3. 单元测试工具: - 先测试工具本身 - 再测试 Function Calling 流程
4. 模拟测试: - 使用 Mock 测试工具逻辑 - 避免实际 API 调用
5. 错误注入: - 模拟工具失败 - 测试错误处理"""13.4 生产环境 Checklist
PRODUCTION_CHECKLIST = { '工具设计': [ '✓ 工具有清晰的描述', '✓ 参数符合 JSON Schema', '✓ 错误处理完善', '✓ 有单元测试' ], '权限安全': [ '✓ 敏感操作有确认', '✓ 输入消毒', '✓ 权限分级', '✓ 审计日志' ], '性能优化': [ '✓ 工具数量控制在 10-20 个', '✓ 实施 Prompt Caching', '✓ 并行调用独立工具', '✓ 监控 Token 使用' ], '可靠性': [ '✓ 重试机制', '✓ 降级方案', '✓ 超时处理', '✓ 限流保护' ], '可观测性': [ '✓ 调用追踪', '✓ 性能监控', '✓ 错误告警', '✓ 使用统计' ]}14. 未来趋势
14.1 多模态工具调用
# 未来的工具可能包含图像、音频、视频multimodal_tool = { "name": "analyze_image", "description": "分析图片内容", "parameters": { "type": "object", "properties": { "image_url": {"type": "string"}, "questions": { "type": "array", "items": {"type": "string"} } } }}
# GPT-4V/Gemini 已经支持response = client.chat.completions.create( model="gpt-4o", messages=[{ "role": "user", "content": [ {"type": "text", "text": "这张图里有几个人？"}, {"type": "image_url", "image_url": {"url": "..."}} ] }], tools=[multimodal_tool])14.2 工具发现
# 自动从 API 文档生成工具定义def auto_discover_tools(api_docs): """从 OpenAPI 规范自动生成工具""" tools = [] for endpoint in api_docs['paths']: for method in endpoint.values(): tool = { "name": method['operationId'], "description": method.get('description', ''), "parameters": method.get('requestBody', {}) } tools.append(tool) return tools14.3 工具组合
# 工具可以自动组合解决复杂问题def compose_tools(tools, problem): """自动组合工具""" # LLM 决定工具调用顺序 plan = llm_plan(problem, tools)
# 执行计划 result = None for step in plan: result = execute_tool(step, result)
return result14.4 自适应工具
# 工具可以根据上下文自适应class AdaptiveTool: """自适应工具"""
def execute(self, context, **kwargs): """根据上下文调整行为""" if context['user_type'] == 'premium': return self.premium_handler(**kwargs) else: return self.basic_handler(**kwargs)15. 核心概念总结
15.1 Function Calling 的核心组件
FUNCTION_CALLING_COMPONENTS = """1. Tool Schema (JSON Schema): - 定义工具的功能和参数 - 标准化的元数据
2. LLM Decision: - 决定是否调用工具 - 选择哪个工具 - 生成参数
3. Tool Execution: - 程序执行工具 - 返回结构化结果
4. Result Integration: - LLM 基于结果生成回答 - 完成对话
关键 API:- chat.completions.create(model, messages, tools)- response.tool_calls- message.append(tool_call_id, content)"""15.2 Function Calling 的核心数学表达
- 决策模型：
其中 是可用工具集。
- 参数生成：
- 多轮决策：
其中 是用户消息， 是助手响应。
16. 实战建议
16.1 学习路径
LEARNING_PATH = """1. 基础: - 理解 JSON Schema - 学会 OpenAI Function Calling - 实现一个简单工具
2. 进阶: - 复杂工具设计 - 错误处理 - 多工具协作
3. 高级: - 自定义工具执行 - 权限和安全 - 性能优化
4. 专家: - 多模态工具 - 工具自动发现 - 工具组合"""16.2 常见问题
FAQ = { 'Q: LLM 没有调用工具怎么办?': 'A: 检查 tool_choice 设置, 或改进 system prompt', 'Q: 工具参数错误怎么办?': 'A: 改进 description, 添加 example, 验证 schema', 'Q: 成本太高怎么办?': 'A: 减少工具数量, 启用 Prompt Caching', 'Q: 工具调用慢怎么办?': 'A: 并行调用, 优化工具实现', 'Q: 如何调试?': 'A: 启用日志, 使用 LangSmith 等追踪工具'}17. 总结与展望
Function Calling 的核心价值
- 连接 LLM 与真实世界：打破信息孤岛
- 标准化工具调用：统一的 JSON Schema 标准
- 提高准确性：基于真实数据而非幻觉
- 构建 Agent 基础：现代 Agent 系统的核心
Function Calling 的局限与挑战
| 局限 | 描述 | 缓解 |
|---|---|---|
| Schema 设计 | 需要精心设计 | 自动生成工具 |
| Token 成本 | 工具定义占用 token | Prompt Caching |
| 错误处理 | 工具失败易传播 | 容错设计 |
| 学习曲线 | 需要理解多个概念 | 框架支持 |
未来方向
- 更智能的工具选择：LLM 自主选择最合适的工具
- 工具自学习：根据使用历史自动优化
- 多模态工具：图像、音频、视频工具
- 工具组合语言：高层语言组合工具
- 完全自主 Agent：最小化人工干预
总结
Function Calling 是 LLM 走向真实世界的关键技术。从 OpenAI 2023 年 6 月推出至今，已经成为现代 LLM 应用的标准特性。
掌握 Function Calling 是构建 LLM 应用的必备技能——无论是简单的天气查询，还是复杂的 Agent 系统，Function Calling 都提供了统一、可靠、标准化的工具调用机制。
参考资料
- OpenAI. (2023). “Function Calling and Other API Updates.” OpenAI Blog.
- Anthropic. (2024). “Tool Use (Function Calling) with Claude.” Anthropic Docs.
- Google. (2024). “Function Calling with Gemini.” Google AI Docs.
- OpenAI. (2024). “Structured Outputs Guide.” OpenAI Cookbook.
- Anthropic. (2024). “Prompt Caching for Tools.” Anthropic Docs.
- JSON Schema Official Specification. json-schema.org.
- Schick, T., et al. (2023). “Toolformer: Language Models Can Teach Themselves to Use Tools.” NeurIPS.
- Yao, S., et al. (2022). “ReAct: Synergizing Reasoning and Acting in Language Models.” ICLR.
- OpenAPI Specification. spec.openapis.org.
- Mialon, G., et al. (2023). “Augmented Language Models: a Survey.” TMLR.
- Patil, S., et al. (2024). “Function Calling Best Practices.” OpenAI Cookbook.
- Anthropic. (2024). “Building Effective Agents with Claude.” Anthropic Engineering Blog.
文章分享
如果这篇文章对你有帮助，欢迎分享给更多人！
深入理解 Function Calling：让大模型连接真实世界
https://aiattnstudio.link/posts/function-calling/欢迎来到Federico的个人博客
57文章
7分类
404标签