Function Calling
--- title: "Function Calling" description: "掌握OpenAI Function Calling机制的原理与实现，理解结构化输出的最佳实践" tags: ["Function Calling", "OpenAI", "结构化输出", "工具调用"] category: "llm" icon: "🧠"
Function Calling
Function Calling简介
Function Calling是OpenAI推出的一项功能，允许GPT模型在响应中调用预定义的函数。这使得模型能够与外部系统交互，获取实时数据，执行复杂计算，从而大大扩展了LLM的应用范围。
与传统的文本生成不同，Function Calling让模型能够输出结构化的函数调用信息，包括函数名和参数，然后由应用程序执行这些函数并将结果返回给模型。
核心机制
1. 函数定义与调用
import json
from typing import List, Dict, Any, Optional
from openai import OpenAI
class FunctionCaller:
def __init__(self, model: str = "gpt-4"):
self.client = OpenAI()
self.model = model
self.functions: List[Dict] = []
def add_function(self, name: str, description: str,
parameters: Dict, required: List[str] = None):
"""添加函数定义"""
function_def = {
"name": name,
"description": description,
"parameters": {
"type": "object",
"properties": parameters,
"required": required or []
}
}
self.functions.append(function_def)
def call(self, messages: List[Dict],
function_call: str = "auto") -> Dict[str, Any]:
"""调用LLM并处理函数调用"""
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
functions=self.functions,
function_call=function_call
)
message = response.choices[0].message
# 检查是否有函数调用
if message.function_call:
return {
"type": "function_call",
"function_name": message.function_call.name,
"arguments": json.loads(message.function_call.arguments),
"message": message
}
else:
return {
"type": "text",
"content": message.content,
"message": message
}
def execute_function(self, function_name: str, arguments: Dict) -> str:
"""执行函数"""
# 这里实现实际的函数逻辑
function_map = {
"get_weather": self._get_weather,
"search_products": self._search_products,
"calculate": self._calculate
}
if function_name in function_map:
return function_map[function_name](**arguments)
else:
return f"Unknown function: {function_name}"
def _get_weather(self, city: str, unit: str = "celsius") -> str:
"""获取天气"""
return json.dumps({
"city": city,
"temperature": 25,
"unit": unit,
"condition": "sunny"
})
def _search_products(self, query: str, max_results: int = 5) -> str:
"""搜索产品"""
return json.dumps({
"query": query,
"results": [
{"name": f"Product {i}", "price": 100 + i * 10}
for i in range(max_results)
]
})
def _calculate(self, expression: str) -> str:
"""计算表达式"""
try:
result = eval(expression)
return json.dumps({"expression": expression, "result": result})
except Exception as e:
return json.dumps({"error": str(e)})
# 使用示例
caller = FunctionCaller()
# 添加函数
caller.add_function(
name="get_weather",
description="获取指定城市的当前天气",
parameters={
"city": {
"type": "string",
"description": "城市名称，如'北京'"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位"
}
},
required=["city"]
)
# 对话
messages = [
{"role": "user", "content": "北京今天天气怎么样？"}
]
result = caller.call(messages)
print(f"调用类型: {result['type']}")
if result['type'] == 'function_call':
print(f"函数名: {result['function_name']}")
print(f"参数: {result['arguments']}")
2. 多轮函数调用
class MultiTurnFunctionCaller:
def __init__(self, model: str = "gpt-4"):
self.client = OpenAI()
self.model = model
self.functions: List[Dict] = []
self.function_implementations: Dict[str, callable] = {}
def register_function(self, name: str, description: str,
parameters: Dict, implementation: callable,
required: List[str] = None):
"""注册函数"""
self.functions.append({
"name": name,
"description": description,
"parameters": {
"type": "object",
"properties": parameters,
"required": required or []
}
})
self.function_implementations[name] = implementation
def chat(self, user_message: str,
max_iterations: int = 5) -> str:
"""多轮对话，支持多次函数调用"""
messages = [{"role": "user", "content": user_message}]
for _ in range(max_iterations):
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
functions=self.functions,
function_call="auto"
)
message = response.choices[0].message
messages.append(message)
# 检查是否有函数调用
if message.function_call:
func_name = message.function_call.name
func_args = json.loads(message.function_call.arguments)
# 执行函数
result = self.function_implementations[func_name](**func_args)
# 添加函数结果到消息
messages.append({
"role": "function",
"name": func_name,
"content": result
})
else:
# 没有函数调用，返回最终回答
return message.content
return "达到最大迭代次数"
# 使用示例
caller = MultiTurnFunctionCaller()
# 注册函数
caller.register_function(
name="get_stock_price",
description="获取股票价格",
parameters={
"symbol": {"type": "string", "description": "股票代码"}
},
implementation=lambda symbol: json.dumps({"symbol": symbol, "price": 150.25})
)
caller.register_function(
name="calculate_portfolio_value",
description="计算投资组合价值",
parameters={
"holdings": {"type": "array", "items": {"type": "object"}},
"prices": {"type": "object"}
},
implementation=lambda holdings, prices: json.dumps({"total_value": 50000})
)
# 多轮对话
response = caller.chat("帮我查一下苹果公司的股票价格，然后计算我的投资组合价值")
print(response)
结构化输出
from pydantic import BaseModel, Field
from typing import List, Optional
class WeatherResponse(BaseModel):
city: str = Field(description="城市名称")
temperature: float = Field(description="温度")
condition: str = Field(description="天气状况")
humidity: int = Field(description="湿度百分比")
class ProductSearchResult(BaseModel):
query: str = Field(description="搜索查询")
total_results: int = Field(description="总结果数")
products: List[Dict] = Field(description="产品列表")
class StructuredOutputCaller:
def __init__(self, model: str = "gpt-4"):
self.client = OpenAI()
self.model = model
def get_structured_output(self, messages: List[Dict],
response_model: BaseModel) -> BaseModel:
"""获取结构化输出"""
# 构建JSON Schema
schema = response_model.model_json_schema()
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
response_format={"type": "json_object"}
)
content = json.loads(response.choices[0].message.content)
# 验证并返回Pydantic模型
return response_model(**content)
def parse_weather(self, query: str) -> WeatherResponse:
"""解析天气查询"""
messages = [
{"role": "user", "content": query}
]
return self.get_structured_output(messages, WeatherResponse)
def parse_product_search(self, query: str) -> ProductSearchResult:
"""解析产品搜索"""
messages = [
{"role": "user", "content": query}
]
return self.get_structured_output(messages, ProductSearchResult)
# 使用示例
caller = StructuredOutputCaller()
weather = caller.parse_weather("北京今天天气怎么样？")
print(f"城市: {weather.city}, 温度: {weather.temperature}°C")
错误处理
class FunctionCallError(Exception):
def __init__(self, message: str, error_type: str = "unknown"):
self.message = message
self.error_type = error_type
super().__init__(self.message)
class RobustFunctionCaller:
def __init__(self, model: str = "gpt-4"):
self.client = OpenAI()
self.model = model
self.functions: List[Dict] = []
self.retry_count = 3
def safe_call(self, messages: List[Dict]) -> Dict:
"""安全调用，包含错误处理"""
for attempt in range(self.retry_count):
try:
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
functions=self.functions if self.functions else None,
function_call="auto"
)
message = response.choices[0].message
if message.function_call:
# 验证函数名和参数
func_name = message.function_call.name
if not self._validate_function_name(func_name):
raise FunctionCallError(f"Invalid function: {func_name}")
try:
args = json.loads(message.function_call.arguments)
self._validate_arguments(func_name, args)
except json.JSONDecodeError as e:
raise FunctionCallError(f"Invalid JSON arguments: {e}")
except ValueError as e:
raise FunctionCallError(f"Invalid arguments: {e}")
return {
"success": True,
"function_name": func_name,
"arguments": args
}
else:
return {
"success": True,
"content": message.content
}
except FunctionCallError as e:
if attempt < self.retry_count - 1:
messages.append({
"role": "system",
"content": f"Error: {e.message}. Please try again."
})
continue
else:
return {
"success": False,
"error": str(e)
}
except Exception as e:
if attempt < self.retry_count - 1:
continue
return {
"success": False,
"error": str(e)
}
def _validate_function_name(self, name: str) -> bool:
"""验证函数名"""
valid_names = [f["name"] for f in self.functions]
return name in valid_names
def _validate_arguments(self, function_name: str, args: Dict):
"""验证参数"""
# 查找函数定义
func_def = next(
(f for f in self.functions if f["name"] == function_name),
None
)
if func_def is None:
raise ValueError(f"Function not found: {function_name}")
# 检查必需参数
required = func_def["parameters"].get("required", [])
for param in required:
if param not in args:
raise ValueError(f"Missing required parameter: {param}")
总结
Function Calling使得LLM能够与外部系统交互，是构建AI应用的核心技术。通过正确设计函数定义、处理多轮调用和错误情况，可以创建强大的智能应用。