Function Calling机制原理与API接口规范
大模型函数调用(Function Calling)是指大语言模型在推理过程中识别用户意图,生成结构化函数调用参数,由外部系统执行实际操作并将结果回传模型的过程。这一机制是AI Agent工具链的核心能力,使大模型从纯文本生成扩展到可操作外部工具和API。
OpenAI兼容API中,Function Calling通过tools参数定义可用函数列表,模型在响应中返回tool_calls字段,包含函数名和参数JSON。以OpenAI API为例:
import openai
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"]
}
}
}],
tool_choice="auto"
)
tool_calls = response.choices[0].message.tool_calls
# model返回: get_weather(city="北京", date="2026-09-04")
tool_choice参数控制调用行为:auto由模型自主决定是否调用函数,none禁止调用,required强制调用,也可以指定具体函数名。模型不直接执行函数,而是返回调用意图,由应用层负责实际执行。
工具函数的JSON Schema定义与参数校验
函数定义遵循JSON Schema规范,parameters字段描述参数类型、嵌套结构和约束条件。复杂参数通过properties嵌套定义:
tools = [{
"type": "function",
"function": {
"name": "search_products",
"description": "搜索商品信息",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"},
"filters": {
"type": "object",
"properties": {
"price_min": {"type": "number"},
"price_max": {"type": "number"},
"category": {"type": "string", "enum": ["electronics", "books", "clothing"]}
}
},
"sort": {"type": "string", "enum": ["price_asc", "price_desc", "relevance"]}
},
"required": ["query"]
}
}
}]
参数校验应在应用层完成。模型生成的参数可能存在类型错误或枚举值越界,使用Pydantic校验是常见做法:
from pydantic import BaseModel, validator
class SearchFilters(BaseModel):
price_min: float | None = None
price_max: float | None = None
category: str | None = None
class SearchRequest(BaseModel):
query: str
filters: SearchFilters | None = None
sort: str | None = "relevance"
@validator("sort")
def validate_sort(cls, v):
allowed = ["price_asc", "price_desc", "relevance"]
if v not in allowed:
raise ValueError(f"sort must be one of {allowed}")
return v
多轮对话中的函数调用流程与结果回传
完整的Function Calling流程包含四个阶段:用户输入、模型决策、函数执行、结果回传。结果回传时需要将函数输出作为tool角色的消息追加到对话历史:
import json
messages = [{"role": "user", "content": "北京今天适合户外运动吗?"}]
# 第一轮:模型返回函数调用
response = openai.chat.completions.create(
model="gpt-4o", messages=messages, tools=tools, tool_choice="auto"
)
assistant_msg = response.choices[0].message
messages.append(assistant_msg)
# 执行函数
if assistant_msg.tool_calls:
for tc in assistant_msg.tool_calls:
args = json.loads(tc.function.arguments)
result = get_weather(**args)
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": json.dumps(result, ensure_ascii=False)
})
# 第二轮:模型基于函数结果生成最终回复
final_response = openai.chat.completions.create(
model="gpt-4o", messages=messages, tools=tools
)
print(final_response.choices[0].message.content)
# 输出: 北京今天晴,气温22-30度C,风力3级,适合户外运动。
tool_call_id字段是关键,模型通过该ID将函数结果与调用请求对应。多个函数调用时,每个tool消息必须包含对应的tool_call_id。
并发函数调用与错误处理策略
GPT-4等模型支持单次响应中返回多个函数调用。并行执行多个独立函数可以显著降低延迟:
import asyncio
async def execute_tool_calls(tool_calls):
tasks = []
for tc in tool_calls:
func_name = tc.function.name
args = json.loads(tc.function.arguments)
tasks.append(execute_single(func_name, args, tc.id))
return await asyncio.gather(*tasks, return_exceptions=True)
async def execute_single(func_name, args, call_id):
try:
func = TOOL_REGISTRY.get(func_name)
if not func:
return {"role": "tool", "tool_call_id": call_id,
"content": json.dumps({"error": f"未知函数: {func_name}"})}
result = await func(**args)
return {"role": "tool", "tool_call_id": call_id,
"content": json.dumps(result, ensure_ascii=False)}
except Exception as e:
return {"role": "tool", "tool_call_id": call_id,
"content": json.dumps({"error": str(e)}, ensure_ascii=False)}
错误处理策略包括三种模式:函数执行失败时返回错误信息让模型自行重试或调整参数;设置最大重试次数防止无限循环;对超时函数返回降级结果。实际项目中建议设置max_tool_rounds限制函数调用轮次,通常3-5轮即可覆盖大部分场景。
Function Calling在Agent工作流中的编排实践
在AI Agent工作流中,Function Calling通常与任务规划、记忆管理和多步推理结合。一个典型的Agent执行循环如下:
class AgentRunner:
def __init__(self, tools, system_prompt, max_rounds=5):
self.tools = tools
self.max_rounds = max_rounds
self.messages = [{"role": "system", "content": system_prompt}]
async def run(self, user_input):
self.messages.append({"role": "user", "content": user_input})
for round_num in range(self.max_rounds):
resp = openai.chat.completions.create(
model="gpt-4o",
messages=self.messages,
tools=self.tools,
tool_choice="auto"
)
msg = resp.choices[0].message
self.messages.append(msg)
if not msg.tool_calls:
return msg.content
results = await execute_tool_calls(msg.tool_calls)
self.messages.extend(results)
return "达到最大执行轮次,任务未完成"
实际部署中还需要考虑函数注册表的动态加载、调用日志审计、权限控制和成本统计。函数注册表建议采用装饰器模式自动收集函数元数据:
TOOL_REGISTRY = {}
def tool(description, parameters):
def decorator(func):
TOOL_REGISTRY[func.__name__] = func
func.tool_spec = {
"type": "function",
"function": {
"name": func.__name__,
"description": description,
"parameters": parameters
}
}
return func
return decorator
@tool("查询订单状态", {"type": "object", "properties": {"order_id": {"type": "string"}}, "required": ["order_id"]})
async def check_order(order_id):
return {"order_id": order_id, "status": "shipped", "eta": "2026-09-06"}
这种模式下,新增工具只需添加装饰器,Agent框架自动注册并暴露给模型。大型项目中可将工具按业务域拆分到不同模块,通过import自动注册。函数调用的返回值应保持简洁结构化,避免返回大段文本导致上下文膨胀。对于需要返回大量数据的场景(如数据库查询结果),在函数内部做分页或摘要处理,只将关键信息回传模型。
原创文章,作者:小编,如若转载,请注明出处:https://www.yunthe.com/da-mo-xing-han-shu-diao-yong-shi-zhan-functioncalling-zai/