LLM 能力集成：Function Calling 与工具调用的工程实践

一、LLM 的行动力缺失：为什么模型只会"说"不会"做"

大语言模型在文本生成方面表现出色，但面对"查询用户订单状态""发送邮件通知""调用支付接口"等需要与外部系统交互的任务时，模型只能输出文本建议，无法直接执行操作。Function Calling 机制正是为弥补这一鸿沟而生——它让模型能够识别何时需要调用工具、选择哪个工具、提取哪些参数，并将执行结果回传给模型继续推理。然而，从 Demo 到生产级实现之间，存在参数校验、错误恢复、并发安全和工具编排等多重工程挑战。

二、Function Calling 的执行流程与协议设计

Function Calling 的核心流程分为四步：工具定义注册、模型决策调用、参数提取校验、结果回传续推。模型并不直接执行函数，而是输出一个结构化的调用意图（包含函数名和参数），由应用层负责实际执行并将结果注入上下文。

sequenceDiagram
    participant U as 用户
    participant A as 应用层
    participant L as LLM
    participant T as 工具服务

    U->>A: 发送消息
    A->>L: 消息 + 工具定义
    L->>A: 返回 tool_calls 意图

    alt 参数校验通过
        A->>T: 执行函数调用
        T->>A: 返回执行结果
        A->>L: 工具结果 + 历史上下文
        L->>A: 最终回复
        A->>U: 展示结果
    else 参数校验失败
        A->>L: 校验错误信息 + 历史上下文
        L->>A: 修正后的调用意图
    end

关键设计点在于工具定义的 JSON Schema 规范。Schema 的描述质量直接影响模型的参数提取准确率——模糊的描述会导致模型在参数提取时产生幻觉，遗漏必填参数或填入类型不匹配的值。

三、生产级 Function Calling 的工程实现

3.1 工具注册与 Schema 定义

from dataclasses import dataclass
from typing import Callable, Dict, Any, List, Optional
import jsonschema

@dataclass
class ToolDefinition:
    """工具定义：将函数签名、Schema 与执行逻辑绑定"""
    name: str
    description: str
    parameters: Dict[str, Any]  # JSON Schema
    handler: Callable[..., Any]
    timeout_seconds: int = 30
    max_retries: int = 2

class ToolRegistry:
    """工具注册中心：统一管理工具定义与执行"""

    def __init__(self):
        self._tools: Dict[str, ToolDefinition] = {}

    def register(self, tool: ToolDefinition):
        """注册工具：校验 Schema 合法性后存入注册表"""
        # 校验 Schema 是否为合法的 JSON Schema
        try:
            jsonschema.Draft7Validator.check_schema(tool.parameters)
        except jsonschema.SchemaError as e:
            raise ValueError(f"工具 {tool.name} 的参数 Schema 不合法: {e}")

        self._tools[tool.name] = tool

    def get_openai_tools_format(self) -> List[Dict[str, Any]]:
        """生成 OpenAI API 所需的 tools 参数格式"""
        return [
            {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.parameters,
                },
            }
            for tool in self._tools.values()
        ]

    async def execute(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
        """执行工具调用：校验参数 → 执行 → 返回结果"""
        tool = self._tools.get(tool_name)
        if tool is None:
            raise ToolNotFoundError(f"未注册的工具: {tool_name}")

        # 参数校验：在执行前拦截非法参数，避免下游服务异常
        try:
            jsonschema.validate(arguments, tool.parameters)
        except jsonschema.ValidationError as e:
            raise ToolArgumentError(
                f"工具 {tool_name} 参数校验失败: {e.message}"
            )

        # 执行工具调用，带超时与重试
        import asyncio
        for attempt in range(tool.max_retries + 1):
            try:
                result = await asyncio.wait_for(
                    tool.handler(**arguments),
                    timeout=tool.timeout_seconds,
                )
                return result
            except asyncio.TimeoutError:
                if attempt == tool.max_retries:
                    raise ToolTimeoutError(
                        f"工具 {tool_name} 执行超时 ({tool.timeout_seconds}s)"
                    )
            except Exception as e:
                if attempt == tool.max_retries:
                    raise ToolExecutionError(f"工具 {tool_name} 执行失败: {e}")

class ToolNotFoundError(Exception): pass
class ToolArgumentError(Exception): pass
class ToolTimeoutError(Exception): pass
class ToolExecutionError(Exception): pass

3.2 多轮工具调用的编排器

import openai
from typing import List, Dict, Any

class FunctionCallingOrchestrator:
    """多轮工具调用编排器：处理模型的连续调用意图"""

    def __init__(
        self,
        client: openai.AsyncOpenAI,
        registry: ToolRegistry,
        model: str = "gpt-4o",
        max_rounds: int = 5,
    ):
        self.client = client
        self.registry = registry
        self.model = model
        self.max_rounds = max_rounds

    async def run(self, messages: List[Dict[str, Any]]) -> str:
        """执行多轮工具调用，直到模型输出最终文本回复"""
        tools = self.registry.get_openai_tools_format()

        for round_idx in range(self.max_rounds):
            response = await self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=tools if tools else None,
                tool_choice="auto",
            )

            choice = response.choices[0]

            # 模型未调用工具，返回文本回复
            if not choice.message.tool_calls:
                return choice.message.content or ""

            # 将模型的工具调用意图加入消息历史
            messages.append(choice.message)

            # 依次执行每个工具调用
            for tool_call in choice.message.tool_calls:
                try:
                    import json
                    arguments = json.loads(tool_call.function.arguments)
                    result = await self.registry.execute(
                        tool_call.function.name, arguments
                    )
                    # 将工具结果加入消息历史
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": json.dumps(
                            {"result": result}, ensure_ascii=False
                        ),
                    })
                except (ToolArgumentError, ToolNotFoundError) as e:
                    # 参数错误或工具不存在：将错误信息回传模型，让其修正
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": json.dumps(
                            {"error": str(e)}, ensure_ascii=False
                        ),
                    })
                except (ToolTimeoutError, ToolExecutionError) as e:
                    # 执行失败：记录日志，回传错误信息，模型可选择降级方案
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": json.dumps(
                            {"error": f"执行失败: {str(e)}"}, ensure_ascii=False
                        ),
                    })

        return "工具调用轮次已达上限，请简化请求后重试。"

四、Function Calling 的边界与权衡

Function Calling 的首要风险是参数幻觉。模型可能为必填参数编造不存在的值，或为枚举类型填入 Schema 之外的选项。生产环境中必须在执行前进行严格的 JSON Schema 校验，拒绝非法参数并将错误信息回传模型修正。但校验本身也有成本——过于严格的 Schema 会限制模型的灵活性，过于宽松则失去防护意义。

多轮调用的状态管理是另一个难点。当模型连续调用多个工具时，前一个工具的输出可能影响后一个工具的参数。编排器需要维护完整的消息历史，但上下文窗口有限，长链调用可能触及 Token 上限。截断历史会丢失关键上下文，保留全部则增加延迟和成本。

在安全性方面，工具调用等同于让模型拥有了执行代码的能力。必须建立严格的权限控制——不同用户角色可调用的工具集不同，敏感操作（如删除数据、发送邮件）需要二次确认。缺少权限控制的 Function Calling 等于将系统接口暴露给不可控的输入源。

五、总结

Function Calling 让 LLM 从"只会说"进化为"能做事"，但生产级实现需要解决参数校验、错误恢复、多轮编排和权限控制等工程问题。核心实践包括：用 JSON Schema 约束工具参数并在执行前校验，用编排器管理多轮调用的消息历史，用错误回传机制让模型自我修正，用权限控制防止越权操作。在工具设计上，优先选择幂等操作，避免模型重试导致重复执行。