函数调用(Function Calling)是大语言模型与外部世界交互的核心能力。通过函数调用,AI不再局限于文本生成,而是能够查询数据库、调用API、执行代码,真正成为连接数字世界的智能助手。本文将深入解析函数调用的原理、实现方式和最佳实践。
📋 目录
核心要点
- 函数调用:让LLM能够识别用户意图并生成结构化的函数调用请求,而非直接执行函数
- JSON Schema:使用标准化的JSON Schema描述函数参数,确保类型安全和参数验证
- 工具选择:LLM根据用户输入智能选择合适的工具,支持单次或并行调用多个函数
- 执行分离:LLM只负责决策"调用什么",实际执行由应用程序完成,保证安全性
- MCP协议:函数调用的标准化演进,实现跨平台工具复用
想要快速生成函数调用所需的JSON Schema?试试我们的在线工具:
什么是函数调用
函数调用(Function Calling),也称为工具调用(Tool Calling),是一种让大语言模型能够与外部系统交互的技术。它允许LLM在对话过程中识别用户意图,并生成结构化的函数调用请求。
函数调用解决的问题
| 问题 | 传统方式 | 函数调用方式 |
|---|---|---|
| 获取实时信息 | LLM无法获取训练数据之后的信息 | 调用搜索API获取最新数据 |
| 执行计算 | LLM数学计算不可靠 | 调用计算器函数精确计算 |
| 操作外部系统 | 无法实现 | 调用API执行数据库操作、发送邮件等 |
| 结构化输出 | 需要复杂提示工程 | 函数参数天然结构化 |
函数调用 vs 普通对话
函数调用工作原理
函数调用的核心流程包含四个关键步骤:定义、选择、执行、整合。
完整工作流程
关键概念解析
1. 函数定义(Function Definition)
开发者预先定义LLM可以调用的函数,包括函数名、描述和参数Schema:
functions = [
{
"name": "get_weather",
"description": "获取指定城市的当前天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称,如:北京、上海"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位"
}
},
"required": ["city"]
}
}
]
2. 函数选择(Function Selection)
LLM根据用户输入和函数描述,智能判断是否需要调用函数,以及调用哪个函数。
3. 参数提取(Parameter Extraction)
LLM从用户输入中提取函数所需的参数,并按照Schema格式化输出。
4. 结果整合(Result Integration)
应用程序执行函数后,将结果返回给LLM,LLM将结果整合到最终回答中。
JSON Schema定义函数
JSON Schema是定义函数参数的标准方式,它确保了参数的类型安全和结构化验证。
基础Schema结构
{
"name": "search_products",
"description": "在商品数据库中搜索产品",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
},
"category": {
"type": "string",
"enum": ["electronics", "clothing", "books", "home"],
"description": "商品类别"
},
"price_range": {
"type": "object",
"properties": {
"min": {"type": "number", "minimum": 0},
"max": {"type": "number", "minimum": 0}
},
"description": "价格范围"
},
"in_stock": {
"type": "boolean",
"description": "是否只显示有库存商品"
}
},
"required": ["query"]
}
}
Schema关键字段说明
| 字段 | 作用 | 示例 |
|---|---|---|
type |
定义参数类型 | string, number, boolean, object, array |
description |
参数描述,帮助LLM理解用途 | "用户的邮箱地址" |
enum |
限定可选值 | ["low", "medium", "high"] |
required |
必填参数列表 | ["query", "user_id"] |
minimum/maximum |
数值范围限制 | "minimum": 0, "maximum": 100 |
minLength/maxLength |
字符串长度限制 | "minLength": 1, "maxLength": 100 |
复杂参数示例
{
"name": "create_order",
"description": "创建新订单",
"parameters": {
"type": "object",
"properties": {
"customer_id": {
"type": "string",
"description": "客户ID"
},
"items": {
"type": "array",
"items": {
"type": "object",
"properties": {
"product_id": {"type": "string"},
"quantity": {"type": "integer", "minimum": 1},
"price": {"type": "number", "minimum": 0}
},
"required": ["product_id", "quantity"]
},
"minItems": 1,
"description": "订单商品列表"
},
"shipping_address": {
"type": "object",
"properties": {
"street": {"type": "string"},
"city": {"type": "string"},
"postal_code": {"type": "string"}
},
"required": ["street", "city"]
}
},
"required": ["customer_id", "items"]
}
}
💡 快速生成Schema:手动编写JSON Schema容易出错,使用 JSON Schema生成器 可以从示例JSON自动生成规范的Schema。
OpenAI函数调用实战
以下是使用OpenAI API实现函数调用的完整示例。
基础实现
import openai
import json
client = openai.OpenAI()
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的当前天气",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称"
}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "search_web",
"description": "搜索网络获取信息",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
}
},
"required": ["query"]
}
}
}
]
def get_weather(city: str) -> dict:
return {
"city": city,
"temperature": 22,
"condition": "晴朗",
"humidity": 45
}
def search_web(query: str) -> dict:
return {
"query": query,
"results": [
{"title": "搜索结果1", "snippet": "相关内容..."},
{"title": "搜索结果2", "snippet": "更多内容..."}
]
}
available_functions = {
"get_weather": get_weather,
"search_web": search_web
}
def chat_with_tools(user_message: str) -> str:
messages = [{"role": "user", "content": user_message}]
response = client.chat.completions.create(
model="gpt-4-turbo",
messages=messages,
tools=tools,
tool_choice="auto"
)
assistant_message = response.choices[0].message
if assistant_message.tool_calls:
messages.append(assistant_message)
for tool_call in assistant_message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
function_response = available_functions[function_name](**function_args)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(function_response, ensure_ascii=False)
})
final_response = client.chat.completions.create(
model="gpt-4-turbo",
messages=messages
)
return final_response.choices[0].message.content
return assistant_message.content
result = chat_with_tools("北京今天天气怎么样?")
print(result)
tool_choice参数详解
| 值 | 行为 |
|---|---|
"auto" |
LLM自动决定是否调用函数(默认) |
"none" |
禁止函数调用,只生成文本 |
"required" |
强制调用至少一个函数 |
{"type": "function", "function": {"name": "xxx"}} |
强制调用指定函数 |
并行函数调用
当用户请求涉及多个独立操作时,LLM可以同时生成多个函数调用请求,提高效率。
并行调用示例
def handle_parallel_calls(user_message: str) -> str:
messages = [{"role": "user", "content": user_message}]
response = client.chat.completions.create(
model="gpt-4-turbo",
messages=messages,
tools=tools,
tool_choice="auto"
)
assistant_message = response.choices[0].message
if assistant_message.tool_calls:
messages.append(assistant_message)
results = []
for tool_call in assistant_message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
result = available_functions[function_name](**function_args)
results.append((tool_call.id, result))
for tool_call_id, result in results:
messages.append({
"role": "tool",
"tool_call_id": tool_call_id,
"content": json.dumps(result, ensure_ascii=False)
})
final_response = client.chat.completions.create(
model="gpt-4-turbo",
messages=messages
)
return final_response.choices[0].message.content
return assistant_message.content
result = handle_parallel_calls("北京和上海今天天气怎么样?")
print(result)
并行调用流程图
与MCP协议的关系
MCP(Model Context Protocol)是函数调用的标准化演进,由Anthropic推出,旨在解决工具复用和跨平台兼容问题。
函数调用 vs MCP
| 对比维度 | 函数调用 | MCP协议 |
|---|---|---|
| 标准化程度 | 各厂商自定义 | 统一开放标准 |
| 工具复用 | 需要为每个平台适配 | 一次开发,多处使用 |
| 通信方式 | HTTP API调用 | JSON-RPC 2.0 |
| 生态系统 | 碎片化 | 统一Server生态 |
| 适用场景 | 简单工具集成 | 复杂AI应用开发 |
从函数调用到MCP
何时选择MCP
- 需要在多个AI客户端(Claude、Cursor等)复用工具
- 构建复杂的AI Agent系统
- 需要标准化的工具发现和调用机制
- 希望利用现有MCP Server生态
💡 探索MCP生态:访问 MCP Server导航 发现数百个可直接使用的MCP工具。
实战案例:构建AI助手
下面我们构建一个完整的AI助手,集成多种工具能力。
完整代码实现
import openai
import json
from datetime import datetime
from typing import Callable
client = openai.OpenAI()
def get_current_time(timezone: str = "Asia/Shanghai") -> dict:
return {
"timezone": timezone,
"datetime": datetime.now().isoformat(),
"formatted": datetime.now().strftime("%Y年%m月%d日 %H:%M:%S")
}
def calculate(expression: str) -> dict:
try:
result = eval(expression, {"__builtins__": {}}, {})
return {"expression": expression, "result": result}
except Exception as e:
return {"expression": expression, "error": str(e)}
def search_database(query: str, table: str, limit: int = 10) -> dict:
mock_data = {
"users": [
{"id": 1, "name": "张三", "email": "zhangsan@example.com"},
{"id": 2, "name": "李四", "email": "lisi@example.com"}
],
"products": [
{"id": 1, "name": "笔记本电脑", "price": 5999},
{"id": 2, "name": "无线鼠标", "price": 99}
]
}
return {
"query": query,
"table": table,
"results": mock_data.get(table, [])[:limit]
}
tools = [
{
"type": "function",
"function": {
"name": "get_current_time",
"description": "获取当前时间",
"parameters": {
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "时区,如Asia/Shanghai"
}
}
}
}
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "执行数学计算",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式,如 2+2*3"
}
},
"required": ["expression"]
}
}
},
{
"type": "function",
"function": {
"name": "search_database",
"description": "搜索数据库",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
},
"table": {
"type": "string",
"enum": ["users", "products"],
"description": "要搜索的表"
},
"limit": {
"type": "integer",
"minimum": 1,
"maximum": 100,
"description": "返回结果数量限制"
}
},
"required": ["query", "table"]
}
}
}
]
available_functions: dict[str, Callable] = {
"get_current_time": get_current_time,
"calculate": calculate,
"search_database": search_database
}
class AIAssistant:
def __init__(self):
self.conversation_history = []
self.system_prompt = """你是一个智能助手,可以帮助用户:
1. 查询当前时间
2. 执行数学计算
3. 搜索数据库信息
请根据用户需求选择合适的工具,并给出友好的回答。"""
def chat(self, user_message: str) -> str:
self.conversation_history.append({
"role": "user",
"content": user_message
})
messages = [
{"role": "system", "content": self.system_prompt}
] + self.conversation_history
response = client.chat.completions.create(
model="gpt-4-turbo",
messages=messages,
tools=tools,
tool_choice="auto"
)
assistant_message = response.choices[0].message
if assistant_message.tool_calls:
self.conversation_history.append(assistant_message)
for tool_call in assistant_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}")
try:
result = available_functions[function_name](**function_args)
except Exception as e:
result = {"error": str(e)}
self.conversation_history.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False)
})
messages = [
{"role": "system", "content": self.system_prompt}
] + self.conversation_history
final_response = client.chat.completions.create(
model="gpt-4-turbo",
messages=messages
)
final_content = final_response.choices[0].message.content
self.conversation_history.append({
"role": "assistant",
"content": final_content
})
return final_content
self.conversation_history.append({
"role": "assistant",
"content": assistant_message.content
})
return assistant_message.content
assistant = AIAssistant()
print(assistant.chat("现在几点了?"))
print(assistant.chat("帮我计算 (15 + 27) * 3"))
print(assistant.chat("查询一下用户表里有哪些用户"))
错误处理与最佳实践
错误处理策略
def safe_function_call(function_name: str, function_args: dict) -> dict:
if function_name not in available_functions:
return {"error": f"未知函数: {function_name}"}
try:
func = available_functions[function_name]
result = func(**function_args)
return {"success": True, "data": result}
except TypeError as e:
return {"error": f"参数错误: {str(e)}"}
except Exception as e:
return {"error": f"执行错误: {str(e)}"}
最佳实践清单
| 实践 | 说明 |
|---|---|
| 清晰的函数描述 | 详细描述函数用途,帮助LLM正确选择 |
| 参数验证 | 使用JSON Schema严格验证参数类型和范围 |
| 错误处理 | 捕获并返回有意义的错误信息 |
| 超时控制 | 为外部API调用设置合理的超时时间 |
| 日志记录 | 记录函数调用和结果,便于调试 |
| 安全考虑 | 避免执行危险操作,验证用户权限 |
安全注意事项
DANGEROUS_FUNCTIONS = ["exec", "eval", "system", "delete"]
def validate_function_call(function_name: str, function_args: dict) -> bool:
if function_name in DANGEROUS_FUNCTIONS:
return False
if "path" in function_args:
path = function_args["path"]
if ".." in path or path.startswith("/"):
return False
return True
常见问题
函数调用和Prompt Engineering有什么区别?
Prompt Engineering通过精心设计提示词来引导LLM输出特定格式的文本,但输出仍是非结构化的。函数调用则让LLM直接生成结构化的JSON格式调用请求,更可靠、更易于程序处理。
如何提高函数选择的准确性?
- 编写清晰、具体的函数描述
- 使用有意义的函数名
- 在description中包含使用场景示例
- 避免功能重叠的函数定义
- 使用enum限定参数可选值
函数调用会增加API成本吗?
是的,函数定义会计入Token消耗。优化建议:
- 只传递当前场景需要的函数
- 精简函数描述,去除冗余信息
- 使用简洁的参数名
如何处理函数执行失败?
- 返回结构化的错误信息给LLM
- 让LLM根据错误信息调整策略或向用户解释
- 实现重试机制处理临时性错误
- 设置最大重试次数避免无限循环
函数调用支持流式输出吗?
支持。在流式模式下,函数调用信息会在delta中逐步返回,需要累积完整的function_call后再执行。
总结
函数调用是让大语言模型从"能说会道"进化到"能说会做"的关键技术。通过标准化的JSON Schema定义和结构化的调用流程,AI能够安全、可靠地与外部系统交互。
关键要点回顾
✅ 函数调用让LLM能够执行实际操作,而非仅生成文本
✅ JSON Schema是定义函数参数的标准方式
✅ 并行调用可以同时处理多个独立请求
✅ MCP协议是函数调用的标准化演进方向
✅ 安全性和错误处理是生产环境的关键考虑
相关资源
- JSON Schema生成器 - 快速生成函数参数Schema
- MCP Server导航 - 发现可复用的MCP工具
- JSON格式化工具 - 调试函数调用数据
延伸阅读
- MCP协议深度解析 - 了解函数调用的标准化演进
- AI Agent开发实战 - 构建更复杂的AI系统
- JSON Schema验证详解 - 深入理解Schema验证
💡 开始实践:使用 JSON Schema生成器 快速创建函数定义,让你的AI应用连接真实世界!