LLM Function Calling 底层机制：从对话模板、有限状态机约束解码到工具循环

Function Calling 的本质：协议化文本与约束解码

大语言模型本身只定义从前缀 Token 序列 $x_{1:t-1} \in V^*$ 到词表 $V$ 上概率分布的映射 $P(x_t \mid x_{1:t-1})$，并不具备直接调用外部进程的原语。工程上所谓的 Function Calling（工具调用），是由以下三层机制协同构成的闭环：

1. 提示词与特殊 Token 协议（Chat Template Protocol）：在输入侧将工具的 JSON Schema 序列化进 System / Developer 消息，并在词表中引入专用控制 Token（如 <|tool_call|>、<|plugin|>）标记模式切换；
2. 受约束自回归解码（Grammar-Constrained Decoding）：在采样层将 JSON Schema 编译为形式文法（FSM / PDA），在每个解码步对不满足语法转移的词表分量施加 $-\infty$ 掩码，从数学上保证输出 100% 符合目标 Schema；
3. 宿主运行时拦截与回填（Host Orchestration Loop）：推理解码器识别停止 Token 后挂起生成，解析结构化参数并调用宿主函数，将返回结果包装为 role="tool" 消息追加回上下文，开启下一轮 Prefill。

---

对话模板与词表控制字符

工具定义注入与角色边界

以主流对话模板（如 Qwen2.5 / Llama-3 / ChatML 衍生协议）为例，宿主传入的 tools 列表在分词前被渲染为约定的文本块。模型在微调（SFT）与对齐阶段学习到两类关键分布转移：

• 何时触发工具：当用户问题需要外部状态或精确计算时，首 Token 在 <tool_call>（或 <|python_tag|>）上的 Logit 显著高于普通自然语言 Token；
• 何时终止并交出控制权：工具调用块结束时输出专用结束符（如 </tool_call> 或 <|eom_id|>），与普通回答结束符 <|im_end|> 区分，通知宿主引擎本轮应执行工具而非直接结束对话。

一个典型的多工具调用上下文在 Token 序列层的展开形态如下：

<|im_start|>system
You are a helpful assistant.
# Tools
You may call one or more functions to assist with the user query.
You are provided with function signatures within <tools></tools> XML tags:
<tools>
{"type":"function","function":{"name":"query_metric","parameters":{"type":"object","properties":{"service":{"type":"string"},"window_min":{"type":"integer"}},"required":["service","window_min"],"additionalProperties":false}}}
</tools>
For each function call, return a json object within <tool_call></tool_call> tags.<|im_end|>
<|im_start|>user
查一下 payment-api 最近 15 分钟的错误率<|im_end|>
<|im_start|>assistant
<tool_call>
{"name": "query_metric", "arguments": {"service": "payment-api", "window_min": 15}}
</tool_call><|im_end|>
<|im_start|>user
<tool_response>
{"error_rate": 0.0042, "status": "healthy"}
</tool_response><|im_end|>
<|im_start|>assistant
`payment-api` 最近 15 分钟的错误率为 0.42%，状态正常。<|im_end|>

---

受约束解码：从 JSON Schema 到 DFA 词表掩码

即便经过充分的指令微调，模型在高温采样、长上下文注意力稀释或深层嵌套 Schema 下仍可能输出未闭合引号、遗漏必填字段或非法枚举值。生产级推理引擎（vLLM、SGLang、llama.cpp）通过 Outlines / XGrammar / Guidance 等状态机约束解码器消除语法错误。

Logit 掩码的数学形式

设词表为 $V$（$|V| \approx 128\,000$），形式语言 $\mathcal{L}(\mathcal{G}) \subseteq \Sigma^*$ 为符合目标 JSON Schema 的合法字符串集合。对于已生成的字节前缀 $s_{t-1} \in \Sigma^*$，定义其合法延续前缀集合：

$$\operatorname{Prefix}(\mathcal{L}) = \left\{ u \in \Sigma^* \;\middle|\; \exists w \in \Sigma^*,\; u \cdot w \in \mathcal{L} \right\}$$

在第 $t$ 步模型前向传播输出原始对数几率 $z_t \in \mathbb{R}^{|V|}$ 后，构造二值掩码向量 $m(s_{t-1}) \in \{0, -\infty\}^{|V|}$：

$$m_v(s_{t-1}) = \begin{cases} 0, & \text{若 } s_{t-1} \cdot \operatorname{detok}(v) \in \operatorname{Prefix}(\mathcal{L}) \\ -\infty, & \text{否则} \end{cases}$$

修正后的采样分布为：

$$P'(x_t = v \mid s_{t-1}) = \frac{\exp\bigl((z_{t,v} + m_v(s_{t-1})) / \tau\bigr)}{\sum_{u \in V} \exp\bigl((z_{t,u} + m_u(s_{t-1})) / \tau\bigr)}$$

只要初始空串 $\varepsilon \in \operatorname{Prefix}(\mathcal{L})$ 且至少存在一条通往接受状态的有限路径，归纳可知每一步采样出的轨迹必然收敛在 $\mathcal{L}(\mathcal{G})$ 内。

JSON Schema 编译与 Token 级状态索引预计算

如果在每个解码步实时遍历词表中全部 $|V| = 128\,000$ 个 Token 并逐字符运行正则匹配，单步 CPU 开销将高达数十毫秒，拖垮 GPU 推理吞吐。Outlines 等引擎的核心思想是离线将字节级 DFA 提升（Lift）为 Token 级转移图。

对于不含递归引用（$ref 自环）的 JSON Schema（规定了属性顺序、required、enum、数值正则与字符串边界），可以将其精确展开为一条正规表达式 $R$，并经由 Thompson 构造与子集构造转化为最小化确定性有限自动机：

$$\mathcal{A} = (Q, \Sigma, \delta, q_0, F)$$

其中 $Q$ 为有限状态集，$\delta: Q \times \Sigma \to Q \cup \{\bot\}$ 为字符转移函数。对词表中的每个 Token $v \in V$，设其对应的字符序列为 $\operatorname{detok}(v) = c_1 c_2 \dots c_k$，定义扩展转移函数：

$$\hat{\delta}(q, v) = \delta\bigl(\dots \delta(\delta(q, c_1), c_2) \dots, c_k\bigr)$$

在请求初始化阶段（或按 Schema 哈希缓存），对每个自动机状态 $q \in Q$ 预计算合法 Token 位图：

$$\mathcal{M}(q) = \left\{ v \in V \;\middle|\; \hat{\delta}(q, v) \ne \bot \right\}$$

这样一来，解码期的单步开销从 $O(|V| \cdot \bar{\ell})$ 降为 $O(1)$ 的哈希查表与一次位图张量拷贝。同时，当某个状态 $q$ 的出边在所有合法路径上唯一确定后续若干字符（例如固定键名 {"service":）时，推理引擎甚至可以直接跳过中间步的模型前向传播（Jump-Forward Decoding / Speculative Token Forcing），一次填入多个确定性 Token 再一次性执行前向计算。

DFA 预计算与 LogitsProcessor 实现示例

下面用 Python 演示一个针对固定工具参数结构的字符级 DFA 到词表级掩码表预计算与解码过程：

from dataclasses import dataclass
import math


@dataclass(frozen=True)
class DFA:
    transitions: dict[int, dict[str, int]]
    initial_state: int
    accept_states: frozenset[int]

    def step_char(self, state: int, ch: str) -> int | None:
        return self.transitions.get(state, {}).get(ch)

    def step_token(self, state: int, token_str: str) -> int | None:
        cur: int | None = state
        for ch in token_str:
            assert cur is not None
            cur = self.step_char(cur, ch)
            if cur is None:
                return None
        return cur


class VocabularyMaskIndex:
    """将字符级 DFA 预编译为状态 -> 允许 Token ID 列表的 O(1) 索引。"""

    def __init__(self, dfa: DFA, vocab: dict[int, str], eos_token_id: int) -> None:
        self.dfa = dfa
        self.vocab = vocab
        self.eos_token_id = eos_token_id
        self.allowed_tokens: dict[int, list[int]] = {}
        self.next_state: dict[tuple[int, int], int] = {}
        self._build_index()

    def _build_index(self) -> None:
        for state in self.dfa.transitions:
            valid_ids: list[int] = []
            for tid, piece in self.vocab.items():
                if tid == self.eos_token_id:
                    continue
                nxt = self.dfa.step_token(state, piece)
                if nxt is not None:
                    valid_ids.append(tid)
                    self.next_state[(state, tid)] = nxt
            if state in self.dfa.accept_states:
                valid_ids.append(self.eos_token_id)
            self.allowed_tokens[state] = valid_ids

    def apply_mask(self, state: int, logits: list[float]) -> list[float]:
        masked = [-math.inf] * len(logits)
        for tid in self.allowed_tokens.get(state, []):
            masked[tid] = logits[tid]
        return masked

对于含递归定义（如嵌套 AST 或任意深度 JSON 值）的上下文无关文法（CFG），状态机升级为下推自动机（Pushdown Automaton）。XGrammar 等系统通过将与栈顶无关的确定性片段静态预编译为自适应 Token 掩码，仅在涉及栈帧压入/弹出的少数分叉 Token 上动态评估栈状态，将 CFG 约束解码的 CPU 开销控制在百微秒量级。

---

并行工具调用、流式解析与容错闭环

在实际生产 Agent 中，Function Calling 还需处理以下三项关键工程细节：

1. 并行调用与调用 ID 绑定

当用户请求“对比北京和上海今天的天气”时，模型在单个 Assistant 回合内生成多个独立的调用项，每个调用由运行时分配唯一的 tool_call_id（如 call_01、call_02）。宿主通过 asyncio.gather 并发执行外部 I/O 后，必须将每条 role="tool" 消息携带对应的 tool_call_id 顺序拼回上下文，以便注意力机制将每个返回值准确绑定回对应的调用参数。

2. 流式增量 JSON 解析

当工具参数包含长段文本（如生成文件补丁 diff 或 SQL 查询）时，等待完整 <tool_call> 闭合再渲染界面会造成明显的交互卡顿。流式前端通常维护一个增量词法状态机（Incremental Partial JSON Parser），在未闭合的字符串和对象末尾虚拟补齐 "} 以实时提取正在生成的字段内容，但工具的实际副作用执行（Side-effect Execution）必须严格推迟到结束符校验通过并完成 Pydantic / JSON Schema 二次校验之后。

3. 结构错误与执行异常的反馈回注

若未开启严格约束解码，或者虽然语法合法但违背业务语义（如时间范围起始大于结束、数据库连接超时），切忌直接抛出未捕获异常中断会话。标准做法是将校验错误或异常摘要格式化为一条标准 role="tool" 响应（设置 is_error=True）：

{
  "role": "tool",
  "tool_call_id": "call_01",
  "content": "ValidationError: 'window_min' must be between 1 and 1440, got -5. Please fix the argument and retry."
}

模型在下一轮 Prefill 读取该反馈后，即可利用上下文注意力修正参数并重新发起调用。为防止模型在永久性故障上陷入死循环，宿主循环必须同时设置最大工具迭代深度（如 max_steps = 8）与相同调用签名去重熔断。