大模型 Function Calling 与工具使用：从对话到行动的工程实践

一、大模型的"知行鸿沟"：能说不能做的困境

大语言模型拥有丰富的知识储备，但本质上只是一个文本生成器——它无法执行操作、无法查询实时数据、无法调用外部 API。用户问"北京今天天气如何"，模型只能基于训练数据猜测，无法获取实时天气信息。用户问"帮我查一下订单 #12345 的状态"，模型无法连接数据库。

Function Calling（函数调用）是打通"知"与"行"的桥梁：模型不再直接回答问题，而是生成结构化的函数调用指令，由外部执行器调用真实 API 获取结果，再将结果返回模型生成最终回答。这一机制让大模型从"百科全书"进化为"智能助手"。

二、Function Calling 的执行流程

Function Calling 的核心流程是：用户提问 → 模型生成函数调用 → 外部执行 → 结果回注 → 模型生成回答。

sequenceDiagram
    participant User as 用户
    participant LLM as 大模型
    participant Executor as 函数执行器
    participant API as 外部 API

    User->>LLM: 北京今天天气如何？
    LLM->>LLM: 判断需要调用 get_weather
    LLM-->>Executor: get_weather(city="北京")
    Executor->>API: HTTP GET /weather?city=北京
    API-->>Executor: {"temp": 28, "condition": "晴"}
    Executor-->>LLM: 函数返回结果
    LLM->>LLM: 基于结果生成自然语言回答
    LLM-->>User: 北京今天晴天，气温 28°C

关键机制：模型不直接执行函数，只生成调用意图和参数。执行器负责实际调用和错误处理。这种分离确保了安全性——模型无法绕过权限限制直接访问外部系统。

三、工程化实现

3.1 工具定义与注册

# tool_registry.py
from dataclasses import dataclass
from typing import Callable, Any
import json

@dataclass
class ToolDefinition:
    name: str
    description: str
    parameters: dict  # JSON Schema
    function: Callable

class ToolRegistry:
    def __init__(self):
        self.tools: dict[str, ToolDefinition] = {}

    def register(self, name: str, description: str, parameters: dict):
        """装饰器：注册工具函数"""
        def decorator(func: Callable):
            self.tools[name] = ToolDefinition(
                name=name,
                description=description,
                parameters=parameters,
                function=func,
            )
            return func
        return decorator

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

    def execute(self, name: str, arguments: dict) -> Any:
        """执行工具函数"""
        tool = self.tools.get(name)
        if not tool:
            raise ValueError(f"未知工具：{name}")
        return tool.function(**arguments)


# 注册具体工具
registry = ToolRegistry()

@registry.register(
    name="get_weather",
    description="获取指定城市的当前天气信息",
    parameters={
        "type": "object",
        "properties": {
            "city": {
                "type": "string",
                "description": "城市名称，如'北京'、'上海'",
            },
        },
        "required": ["city"],
    },
)
def get_weather(city: str) -> dict:
    import requests
    # 调用天气 API
    response = requests.get(
        f"https://api.weather.com/current",
        params={"city": city},
        timeout=10,
    )
    if response.status_code != 200:
        return {"error": f"天气 API 请求失败：HTTP {response.status_code}"}
    return response.json()

@registry.register(
    name="search_database",
    description="查询数据库中的订单信息",
    parameters={
        "type": "object",
        "properties": {
            "order_id": {
                "type": "string",
                "description": "订单编号",
            },
            "fields": {
                "type": "array",
                "items": {"type": "string"},
                "description": "需要查询的字段列表",
            },
        },
        "required": ["order_id"],
    },
)
def search_database(order_id: str, fields: list[str] = None) -> dict:
    # 模拟数据库查询
    return {
        "order_id": order_id,
        "status": "已发货",
        "amount": 299.00,
        "tracking_number": "SF1234567890",
    }

3.2 Function Calling 执行器

# function_calling_executor.py
import json
from openai import OpenAI

class FunctionCallingExecutor:
    def __init__(self, registry: ToolRegistry, model: str = "gpt-4o"):
        self.client = OpenAI()
        self.registry = registry
        self.model = model

    def run(self, user_message: str, max_rounds: int = 5) -> str:
        """执行 Function Calling 循环"""
        messages = [
            {"role": "system", "content": "你是一个智能助手，可以调用工具帮助用户。"},
            {"role": "user", "content": user_message},
        ]

        tools = self.registry.get_tool_definitions()

        for round_num in range(max_rounds):
            # 调用 LLM
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=tools if tools else None,
                tool_choice="auto",
            )

            choice = response.choices[0]
            message = choice.message

            # 如果没有工具调用，直接返回回答
            if not message.tool_calls:
                return message.content

            # 将助手消息（含工具调用）加入历史
            messages.append(message)

            # 执行每个工具调用
            for tool_call in message.tool_calls:
                function_name = tool_call.function.name
                try:
                    arguments = json.loads(tool_call.function.arguments)
                except json.JSONDecodeError:
                    arguments = {}

                try:
                    # 执行工具函数
                    result = self.registry.execute(function_name, arguments)
                    result_str = json.dumps(result, ensure_ascii=False)
                except Exception as e:
                    result_str = json.dumps(
                        {"error": f"工具执行失败：{str(e)}"},
                        ensure_ascii=False,
                    )

                # 将工具结果加入消息历史
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result_str,
                })

        return "抱歉，工具调用轮次已达上限，请简化您的问题。"

3.3 多轮工具调用与错误恢复

# advanced_executor.py
class AdvancedExecutor(FunctionCallingExecutor):
    def run_with_retry(self, user_message: str, max_rounds: int = 10) -> str:
        """支持重试和错误恢复的执行器"""
        messages = [
            {"role": "system", "content": (
                "你是一个智能助手。调用工具时请注意：\n"
                "1. 参数必须符合 Schema 定义\n"
                "2. 如果工具返回错误，分析原因并重试\n"
                "3. 不要编造工具返回的数据"
            )},
            {"role": "user", "content": user_message},
        ]

        tools = self.registry.get_tool_definitions()
        failed_calls = set()  # 记录失败的调用，避免无限重试

        for _ in range(max_rounds):
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=tools,
                tool_choice="auto",
            )

            choice = response.choices[0]
            message = choice.message

            if not message.tool_calls:
                return message.content

            messages.append(message)

            for tool_call in message.tool_calls:
                call_key = f"{tool_call.function.name}:{tool_call.function.arguments}"

                # 避免重复调用同一个失败的工具
                if call_key in failed_calls:
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": json.dumps({
                            "error": "此调用之前已失败，请尝试不同的参数或工具"
                        }),
                    })
                    continue

                try:
                    arguments = json.loads(tool_call.function.arguments)
                    result = self.registry.execute(
                        tool_call.function.name, arguments
                    )
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": json.dumps(result, ensure_ascii=False),
                    })
                except Exception as e:
                    failed_calls.add(call_key)
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": json.dumps({
                            "error": str(e),
                            "hint": "请检查参数格式是否正确",
                        }),
                    })

        return "工具调用轮次已达上限。"

四、Function Calling 的 Trade-offs

参数生成的准确性：模型可能生成不符合 Schema 的参数（如字符串传给了数字字段、缺少必填参数）。建议在执行器中加入参数校验层，对不符合 Schema 的参数返回错误信息，让模型重新生成。

工具选择的准确性：模型可能选择错误的工具（如用 search_database 查天气）。优化工具描述（description）是提高选择准确性的关键——描述必须清晰区分工具的适用场景。

延迟的累积效应：每轮 Function Calling 包含一次 LLM 调用 + 一次工具执行，总延迟 = LLM 延迟 × 轮数 + 工具延迟。3 轮调用可能需要 5-10 秒。建议对简单查询（无需工具）跳过工具调用，对复杂查询设置最大轮数限制。

安全性风险：工具函数可能执行危险操作（如删除数据、发送邮件）。建议对写操作工具设置确认机制——执行前向用户展示操作意图，获得确认后再执行。

五、总结

Function Calling 是大模型从"对话"走向"行动"的关键机制，通过模型生成调用意图、外部执行器实际调用的分离架构，兼顾了能力和安全性。落地路线上，建议先定义核心工具集（查询类），再逐步扩展到操作类工具（写入类需加确认）。关键原则：工具描述决定选择准确性，参数校验防止格式错误，错误恢复提升鲁棒性，安全确认守住底线。