OpenAI 函数调用 #
什么是函数调用？ #
函数调用（Function Calling）是 OpenAI 提供的一项强大功能，它允许 GPT 模型在对话过程中"调用"预定义的函数，从而获取实时数据、执行操作或与外部系统交互。
text
┌─────────────────────────────────────────────────────────────┐
│ 函数调用工作流程 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 用户："北京今天天气怎么样？" │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 1. GPT 分析意图，决定调用 get_weather 函数 │ │
│ │ 参数: {"city": "北京"} │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 2. 应用执行函数，获取天气数据 │ │
│ │ 返回: {"temp": 25, "condition": "晴"} │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 3. GPT 根据数据生成自然语言回复 │ │
│ │ "北京今天天气晴朗，气温 25 度。" │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
函数调用的优势 #
text
┌─────────────────────────────────────────────────────────────┐
│ 函数调用优势 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ✅ 获取实时数据 │
│ - 天气、股价、新闻等实时信息 │
│ - 数据库查询 │
│ │
│ ✅ 执行操作 │
│ - 发送邮件 │
│ - 创建订单 │
│ - 调用外部 API │
│ │
│ ✅ 结构化输出 │
│ - 提取特定格式的数据 │
│ - 参数验证 │
│ │
│ ✅ 扩展能力 │
│ - 突破模型知识截止日期限制 │
│ - 访问私有数据 │
│ │
└─────────────────────────────────────────────────────────────┘
基本用法 #
定义函数 #
python
from openai import OpenAI
import json
client = OpenAI()
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的当前天气",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，如：北京、上海"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位"
}
},
"required": ["city"]
}
}
}
]
发送请求 #
python
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "user", "content": "北京今天天气怎么样？"}
],
tools=tools
)
print(response.choices[0].message)
处理函数调用 #
python
message = response.choices[0].message
if message.tool_calls:
for tool_call in message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
print(f"函数名: {function_name}")
print(f"参数: {function_args}")
if function_name == "get_weather":
result = get_weather(
city=function_args["city"],
unit=function_args.get("unit", "celsius")
)
print(f"结果: {result}")
完整示例 #
python
from openai import OpenAI
import json
client = OpenAI()
def get_weather(city: str, unit: str = "celsius") -> dict:
"""模拟天气 API"""
weather_data = {
"北京": {"temp": 25, "condition": "晴"},
"上海": {"temp": 28, "condition": "多云"},
"广州": {"temp": 32, "condition": "雨"}
}
data = weather_data.get(city, {"temp": 20, "condition": "未知"})
if unit == "fahrenheit":
data["temp"] = data["temp"] * 9/5 + 32
return data
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的当前天气",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"]
}
},
"required": ["city"]
}
}
}
]
def run_conversation(user_input: str) -> str:
messages = [{"role": "user", "content": user_input}]
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools
)
message = response.choices[0].message
messages.append(message)
if message.tool_calls:
for tool_call in message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
if function_name == "get_weather":
result = get_weather(**function_args)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False)
})
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools
)
return response.choices[0].message.content
return message.content
print(run_conversation("北京今天天气怎么样？"))
函数定义详解 #
函数结构 #
text
┌─────────────────────────────────────────────────────────────┐
│ 函数定义结构 │
├─────────────────────────────────────────────────────────────┤
│ │
│ { │
│ "type": "function", │
│ "function": { │
│ "name": "函数名称", // 必填 │
│ "description": "函数描述", // 推荐 │
│ "parameters": { // JSON Schema │
│ "type": "object", │
│ "properties": { │
│ "参数名": { │
│ "type": "类型", │
│ "description": "参数描述" │
│ } │
│ }, │
│ "required": ["必需参数列表"] │
│ } │
│ } │
│ } │
│ │
└─────────────────────────────────────────────────────────────┘
参数类型 #
python
tools = [
{
"type": "function",
"function": {
"name": "search_products",
"description": "搜索商品",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
},
"category": {
"type": "string",
"enum": ["electronics", "clothing", "books", "food"],
"description": "商品类别"
},
"price_range": {
"type": "object",
"properties": {
"min": {"type": "number"},
"max": {"type": "number"}
}
},
"in_stock": {
"type": "boolean",
"description": "是否只显示有库存"
},
"limit": {
"type": "integer",
"description": "返回结果数量限制",
"default": 10
},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "标签列表"
}
},
"required": ["query"]
}
}
}
]
多函数定义 #
python
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string"}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "get_stock_price",
"description": "获取股票价格",
"parameters": {
"type": "object",
"properties": {
"symbol": {"type": "string", "description": "股票代码"}
},
"required": ["symbol"]
}
}
},
{
"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"]
}
}
}
]
高级用法 #
并行函数调用 #
python
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "user", "content": "北京和上海的天气怎么样？"}
],
tools=tools
)
message = response.choices[0].message
if message.tool_calls:
print(f"需要调用 {len(message.tool_calls)} 个函数")
for tool_call in message.tool_calls:
print(f"函数: {tool_call.function.name}")
print(f"参数: {tool_call.function.arguments}")
函数调用流程封装 #
python
from openai import OpenAI
import json
from typing import Callable, Dict, Any, List
client = OpenAI()
class FunctionCaller:
def __init__(self):
self.functions: Dict[str, Callable] = {}
self.tools: List[Dict] = []
def register(self, func: Callable, description: str, parameters: dict):
"""注册函数"""
self.functions[func.__name__] = func
self.tools.append({
"type": "function",
"function": {
"name": func.__name__,
"description": description,
"parameters": parameters
}
})
def execute(self, name: str, arguments: dict) -> Any:
"""执行函数"""
if name not in self.functions:
raise ValueError(f"Unknown function: {name}")
return self.functions[name](**arguments)
def chat(self, user_input: str, max_iterations: int = 5) -> str:
"""执行对话"""
messages = [{"role": "user", "content": user_input}]
for _ in range(max_iterations):
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=self.tools
)
message = response.choices[0].message
messages.append(message)
if not message.tool_calls:
return message.content
for tool_call in message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
try:
result = self.execute(function_name, function_args)
result_str = json.dumps(result, ensure_ascii=False)
except Exception as e:
result_str = json.dumps({"error": str(e)})
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result_str
})
return "达到最大迭代次数"
caller = FunctionCaller()
def get_weather(city: str) -> dict:
return {"city": city, "temp": 25, "condition": "晴"}
caller.register(
get_weather,
"获取城市天气",
{
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"}
},
"required": ["city"]
}
)
print(caller.chat("北京天气怎么样？"))
流式函数调用 #
python
from openai import OpenAI
import json
client = OpenAI()
def stream_with_tools(user_input: str):
messages = [{"role": "user", "content": user_input}]
stream = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools,
stream=True
)
tool_calls = []
current_content = ""
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
print(delta.content, end="", flush=True)
current_content += delta.content
if delta.tool_calls:
for tool_call_delta in delta.tool_calls:
idx = tool_call_delta.index
while len(tool_calls) <= idx:
tool_calls.append({"id": "", "name": "", "arguments": ""})
if tool_call_delta.id:
tool_calls[idx]["id"] = tool_call_delta.id
if tool_call_delta.function.name:
tool_calls[idx]["name"] = tool_call_delta.function.name
if tool_call_delta.function.arguments:
tool_calls[idx]["arguments"] += tool_call_delta.function.arguments
return tool_calls
tool_calls = stream_with_tools("北京天气怎么样？")
print("\n\n函数调用:")
for tc in tool_calls:
print(f" {tc['name']}: {tc['arguments']}")
实际应用示例 #
数据库查询 #
python
import sqlite3
from openai import OpenAI
import json
client = OpenAI()
def query_database(query: str) -> list:
"""执行 SQL 查询"""
conn = sqlite3.connect("example.db")
cursor = conn.cursor()
cursor.execute(query)
results = cursor.fetchall()
conn.close()
return results
tools = [
{
"type": "function",
"function": {
"name": "query_database",
"description": "执行 SQL 查询获取数据",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "SQL 查询语句"
}
},
"required": ["query"]
}
}
}
]
def chat_with_db(user_input: str) -> str:
messages = [
{
"role": "system",
"content": "你是一个数据分析助手，可以查询数据库。"
},
{"role": "user", "content": user_input}
]
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools
)
message = response.choices[0].message
messages.append(message)
if message.tool_calls:
for tool_call in message.tool_calls:
args = json.loads(tool_call.function.arguments)
result = query_database(args["query"])
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result)
})
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools
)
return response.choices[0].message.content
return message.content
API 调用 #
python
import requests
from openai import OpenAI
import json
client = OpenAI()
def call_external_api(endpoint: str, method: str = "GET", data: dict = None) -> dict:
"""调用外部 API"""
base_url = "https://api.example.com"
if method == "GET":
response = requests.get(f"{base_url}/{endpoint}")
else:
response = requests.post(f"{base_url}/{endpoint}", json=data)
return response.json()
tools = [
{
"type": "function",
"function": {
"name": "call_external_api",
"description": "调用外部 API",
"parameters": {
"type": "object",
"properties": {
"endpoint": {
"type": "string",
"description": "API 端点"
},
"method": {
"type": "string",
"enum": ["GET", "POST"]
},
"data": {
"type": "object",
"description": "POST 数据"
}
},
"required": ["endpoint"]
}
}
}
]
结构化数据提取 #
python
from openai import OpenAI
import json
client = OpenAI()
tools = [
{
"type": "function",
"function": {
"name": "extract_person_info",
"description": "提取人物信息",
"parameters": {
"type": "object",
"properties": {
"name": {"type": "string"},
"age": {"type": "integer"},
"email": {"type": "string"},
"phone": {"type": "string"},
"address": {"type": "string"}
},
"required": ["name"]
}
}
}
]
def extract_info(text: str) -> dict:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": "从文本中提取人物信息，使用函数返回结构化数据。"
},
{"role": "user", "content": text}
],
tools=tools,
tool_choice={"type": "function", "function": {"name": "extract_person_info"}}
)
tool_call = response.choices[0].message.tool_calls[0]
return json.loads(tool_call.function.arguments)
text = "张三今年28岁，邮箱是zhangsan@example.com，住在北京朝阳区"
info = extract_info(text)
print(json.dumps(info, indent=2, ensure_ascii=False))
工具选择控制 #
tool_choice 参数 #
text
┌─────────────────────────────────────────────────────────────┐
│ tool_choice 选项 │
├─────────────────────────────────────────────────────────────┤
│ │
│ "auto"（默认） │
│ ───────────────────────────────────────────────────────── │
│ 模型自动决定是否调用函数 │
│ │
│ "none" │
│ ───────────────────────────────────────────────────────── │
│ 不调用任何函数，只生成文本 │
│ │
│ "required" │
│ ───────────────────────────────────────────────────────── │
│ 必须调用至少一个函数 │
│ │
│ {"type": "function", "function": {"name": "函数名"}} │
│ ───────────────────────────────────────────────────────── │
│ 强制调用指定函数 │
│ │
└─────────────────────────────────────────────────────────────┘
强制调用特定函数 #
python
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "user", "content": "用户输入"}
],
tools=tools,
tool_choice={
"type": "function",
"function": {"name": "get_weather"}
}
)
禁止调用函数 #
python
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "user", "content": "用户输入"}
],
tools=tools,
tool_choice="none"
)
最佳实践 #
1. 函数描述要清晰 #
python
tools = [
{
"type": "function",
"function": {
"name": "search_products",
"description": "在商品数据库中搜索商品。支持按名称、类别和价格范围筛选。",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词，如商品名称或描述"
},
"category": {
"type": "string",
"enum": ["electronics", "clothing", "books"],
"description": "商品类别，可选"
}
},
"required": ["query"]
}
}
}
]
2. 错误处理 #
python
def safe_execute_function(name: str, arguments: dict) -> dict:
try:
result = execute_function(name, arguments)
return {"success": True, "data": result}
except Exception as e:
return {"success": False, "error": str(e)}
3. 验证参数 #
python
import jsonschema
def validate_arguments(schema: dict, arguments: dict) -> bool:
try:
jsonschema.validate(arguments, schema)
return True
except jsonschema.ValidationError:
return False
下一步 #
现在你已经掌握了函数调用的使用方法，接下来学习 图像生成，了解如何使用 DALL·E 生成图像！
最后更新：2026-03-29