Function Calling 工程实践：从工具定义到错误恢复的完整链路

一、LLM 工具调用的工程痛点：幻觉与不可靠性

大模型的 Function Calling 能力让 Agent 能够与外部系统交互，但生产环境中这套机制远比 Demo 复杂。最常见的问题是参数幻觉：LLM 生成了符合 Schema 结构但语义错误的参数。例如，调用天气 API 时传入了不存在的城市名，或者调用数据库查询时生成了语法错误的 SQL。更棘手的是，LLM 有时会"发明"不存在的工具名称，或者在应该调用工具时选择直接回答。

这些问题的根源在于 LLM 对工具语义的理解是统计性的而非确定性的。当工具数量增多、参数结构复杂时，指令遵循率显著下降。本文从工具定义规范、调用链路设计和错误恢复三个层面，给出生产级的工程方案。

二、Function Calling 的执行模型与可靠性机制

Function Calling 的完整执行链路包含四个阶段：意图识别→工具选择→参数填充→结果处理。每个阶段都有独立的失败模式，需要针对性的可靠性机制。

sequenceDiagram
    participant U as 用户
    participant A as Agent
    participant L as LLM
    participant T as 工具执行器

    U->>A: 用户请求
    A->>L: 系统 Prompt + 工具定义 + 用户消息
    L-->>A: 工具调用决策（tool_call）
    A->>A: 参数校验（Schema + 语义）
    alt 参数校验通过
        A->>T: 执行工具
        T-->>A: 工具结果
        A->>L: 注入工具结果，继续推理
        L-->>A: 最终回复
    else 参数校验失败
        A->>L: 反馈错误信息，请求修正
        L-->>A: 修正后的工具调用
    end
    A-->>U: 响应

关键可靠性机制是参数校验层：在 LLM 输出和工具执行之间插入校验逻辑，拦截语义错误和结构错误。这比单纯依赖 LLM 的指令遵循要可靠得多，因为校验逻辑是确定性的。

三、生产级 Function Calling 框架实现

import json
import re
from typing import Any, Callable
from pydantic import BaseModel, ValidationError

class ToolDefinition(BaseModel):
    """工具定义：包含 Schema 和执行函数"""
    name: str
    description: str
    parameters: dict[str, Any]  # JSON Schema
    executor: Callable[..., Any]
    # 语义校验规则：字段名 → 校验函数
    validators: dict[str, Callable[[Any], bool]] = {}

class ToolCallResult(BaseModel):
    success: bool
    data: Any = None
    error: str | None = None
    retryable: bool = False

class FunctionCallEngine:
    """Function Calling 执行引擎"""
    def __init__(self, max_retries: int = 2):
        self.tools: dict[str, ToolDefinition] = {}
        self.max_retries = max_retries

    def register(self, tool: ToolDefinition) -> None:
        self.tools[tool.name] = tool

    def get_tool_schemas(self) -> list[dict]:
        """生成 OpenAI Function Calling 格式的工具定义"""
        return [
            {
                "type": "function",
                "function": {
                    "name": t.name,
                    "description": t.description,
                    "parameters": t.parameters,
                }
            }
            for t in self.tools.values()
        ]

    def validate_args(self, tool_name: str, args: dict) -> tuple[bool, str]:
        """双层校验：结构校验 + 语义校验"""
        tool = self.tools.get(tool_name)
        if not tool:
            return False, f"工具 {tool_name} 不存在，可用工具：{list(self.tools.keys())}"

        # 第一层：结构校验（检查必填字段和类型）
        required = tool.parameters.get("required", [])
        for field_name in required:
            if field_name not in args:
                return False, f"缺少必填参数：{field_name}"

        # 第二层：语义校验（业务规则）
        for field_name, validator in tool.validators.items():
            if field_name in args and not validator(args[field_name]):
                return False, f"参数 {field_name} 的值 {args[field_name]} 未通过语义校验"

        return True, ""

    async def execute_tool_call(self, tool_call: dict) -> ToolCallResult:
        """执行单个工具调用，含重试逻辑"""
        tool_name = tool_call["function"]["name"]
        try:
            args = json.loads(tool_call["function"]["arguments"])
        except json.JSONDecodeError as e:
            return ToolCallResult(
                success=False,
                error=f"参数 JSON 解析失败：{e}",
                retryable=True
            )

        # 校验
        is_valid, err_msg = self.validate_args(tool_name, args)
        if not is_valid:
            return ToolCallResult(success=False, error=err_msg, retryable=True)

        tool = self.tools[tool_name]

        # 带重试的执行
        for attempt in range(self.max_retries + 1):
            try:
                result = tool.executor(**args)
                return ToolCallResult(success=True, data=result)
            except Exception as e:
                if attempt == self.max_retries:
                    return ToolCallResult(
                        success=False,
                        error=f"工具执行失败（重试 {self.max_retries} 次后）：{e}",
                        retryable=False
                    )

        return ToolCallResult(success=False, error="未预期的执行路径")

    async def run_with_tools(self, client, messages: list[dict]) -> dict:
        """完整的工具调用循环：LLM 推理 → 工具执行 → 结果注入 → 继续推理"""
        while True:
            response = await client.chat.completions.create(
                model="gpt-4o",
                messages=messages,
                tools=self.get_tool_schemas(),
                tool_choice="auto",
            )
            msg = response.choices[0].message

            # 无工具调用，返回最终回复
            if not msg.tool_calls:
                return {"content": msg.content, "tool_calls": []}

            # 处理所有工具调用
            messages.append({"role": "assistant", "content": msg.content, "tool_calls": msg.tool_calls})
            for tc in msg.tool_calls:
                result = await self.execute_tool_call(tc.model_dump())
                messages.append({
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": json.dumps({
                        "success": result.success,
                        "data": result.data,
                        "error": result.error
                    }, ensure_ascii=False)
                })

核心设计：validate_args 实现双层校验，结构校验拦截缺失字段，语义校验拦截业务错误；execute_tool_call 内置重试机制，区分可重试错误和不可重试错误；run_with_tools 实现完整的工具调用循环，自动处理多轮调用。

四、Function Calling 的 Trade-offs 分析

工具数量与选择准确率的负相关：当注册工具超过 15 个时，LLM 的工具选择准确率明显下降。解决方案是按业务域分组，Orchestrator 先做意图路由，再加载对应域的工具子集。这增加了架构复杂度，但显著提升选择准确率。

参数校验的成本：语义校验函数本身需要开发和维护，且可能引入误判。例如城市名校验需要维护城市列表，列表不全就会误拒合法输入。建议对高频工具做严格语义校验，低频工具只做结构校验。

重试循环的风险：LLM 修正参数后仍可能生成错误参数，导致无限重试。必须设置 max_retries 上限，并在重试耗尽后降级为直接回复用户，而非继续循环。

并行工具调用的顺序依赖：OpenAI 支持一次返回多个 tool_calls，但如果工具间有依赖关系（如工具 B 需要工具 A 的输出），并行执行会导致失败。需要在工具定义中声明依赖关系，由执行引擎做拓扑排序。

五、总结

Function Calling 的生产级落地关键在于三层防御：结构校验拦截格式错误，语义校验拦截业务错误，重试机制处理临时故障。工具数量增多时需要引入分组路由策略，避免选择准确率下降。参数校验和重试逻辑虽然增加了开发成本，但这是 LLM 统计性输出特性所必需的工程补偿。落地建议：先从 3-5 个核心工具起步，验证调用链路稳定性后再逐步扩展工具集。