大模型函数调用实战:Function Calling在AI Agent工具链中的工程化实现

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/

(0)
小编小编
上一篇 2小时前
下一篇 1小时前

相关推荐