在前面三篇文章里，我们已经把 mini Agent Runtime 的第一阶段底座彻底搭稳了：从第 1 篇的 POMDP 轨迹开销建模，到第 2 篇基于 Python 反射的 ToolRegistry 与 ReAct 循环，再到第 3 篇支持跨进程快照恢复、死循环拦截与人工审批挂起的两阶段状态机 StatefulAgentRuntime。
到现在为止，我们的状态机跑得非常优雅，但整个系统里其实还埋着一颗随时会引爆的定时炸弹——大模型的结构化输出（Structured Output）可靠性。
只要你用真实的大模型（尤其是 7B、14B 级别的开源模型，或者在输出极长代码片段时的百亿级模型）跑过几十次工具调用，你一定遇到过下面这些让人抓狂的现场：
尾部括号截断或缺失：模型在写一个嵌套 JSON 参数时，前面写得都对，最后收尾时少写了一个
}，导致json.loads直接抛出JSONDecodeError: Expecting ',' delimiter。多行字符串里的裸换行与未转义引号：当 Agent 调用
write_file(path="main.py", content="...")写入一段 Python 代码时，模型很容易在 JSON 字符串内部直接敲一个真实的回车符\n（ASCII0x0A），或者在代码里写双引号print("hello")却忘了加反斜杠转义\"。LLM 的“程序员强迫症”：JSON 标准规范（RFC 8259）严格禁止尾随逗号（Trailing Comma，如
{"a": 1,}），严格禁止单引号包裹键名（{'a': 1}），也严格禁止//注释，但因为预训练语料里有海量的 JavaScript 和 Python 字典代码，模型顺手就会把这些语法带进Action Input里。把布尔值和空值写成 Python 风格：模型经常输出
{"ignore_empty": True, "filter": None}而不是 JSON 标准的true和null。
在第 2 篇里，我们做了一层最基础的兜底：一旦 json.loads 报错，就把异常信息包装成一条 Observation(is_error=True) 塞回给模型，让模型在下一轮 tick() 里自己重写一遍。
这个方法虽然能保住进程不崩，但从系统工程的角度看，为了一个少写的右花括号或一个多余的尾随逗号，就让推理引擎白白再跑一轮几千 Token 的 Prefill 和几百 Token 的 Decode，实在是太昂贵了。 更糟糕的是，如果模型处于低温采样（temperature = 0.0），有时即便你把 JSONDecodeError: Expecting property name enclosed in double quotes 喂回给它，它也看不懂报错到底指哪个字符，下一轮依然原样输出错误格式，直接撞上我们在第 3 篇写的死循环熔断器！
怎么彻底根治结构化输出崩溃的问题？
因为我们不仅在写上层的 mini Agent Runtime，手里还握着底层的 mini vLLM（特别是我们在《mini vLLM》第 11 篇专门设计过的 Sampling 系统与 LogitsProcessor），我们可以构建一套大多数学术框架和调包教程给不出的“三层纵深防御体系”：
模型原始输出 (Logits / Raw Text)
│
├──► 【第 1 层：采样期强约束 (In-Engine FSM Constrained Decoding)】
│ 在 mini vLLM 的 LogitsProcessor 层维护 JSON/Schema 有限状态机，
│ 每一步将破坏语法的非法 Token 的 logit 直接置为 -inf，从源头消灭语法错误！
│
├──► 【第 2 层：解析期确定性修复 (Deterministic Lexer/Stack Repair)】
│ 当对接外部黑盒 API (无法控制底层 Sampler) 时，在 Runtime 解析层通过词法状态机
│ 零成本自动修复：尾随逗号、裸换行符、True/None 字面量、以及未闭合的括号栈！
│
└──► 【第 3 层：语义期精准诊断回传 (Schema Validation & Pinpoint Retry)】
当语法合法但字段类型/枚举值违反 Schema 时，生成带字段路径 (JSONPath) 与
错误位置的精准诊断报告，触发带上下文的局部重试。
这一篇，我们就把这三层防御从数学原理到核心代码完整实现出来。
1. 为什么自由自回归采样一定会撞上语法失败率墙
我们先算一笔简单的概率账，看看为什么不加约束的自由采样在长工具调用下必然翻车。
假设模型在生成一段长度为 L 个 Token 的 JSON 参数序列 Y = (y₁, y₂, …, y_L)。设在每一步 t，词表 V 中所有能够保持当前前缀 y_<t 符合合法 JSON 语法（且最终可扩展为满足目标 Schema 的完整对象）的合法 Token 子集为 V_valid(y_<t) ⊆ V。
即便经过良好微调的大模型在单步上的语法合规概率极高，设每一步模型将概率质量分配给合法集合的總和为：
pₜ = Σ(v ∈ V_valid(y_<t)) P_θ(yₜ = v | x, y_<t) = 1 - εₜ
其中 εₜ ≪ 1 是单步出现语法滑坡（比如多打一个逗号、漏掉一个转义符）的微小错误率。那么整段长度为 L 的 JSON 序列一次性合法生成的联合概率 P_valid(L) 为：
P_valid(L) = Π(t = 1…L) (1 - εₜ) ≈ exp( −Σ(t = 1…L) εₜ )
如果平均每步的微错误率只有 ε̄ = 0.001（即单步准确率高达 99.9%）：
当调用一个简单的查询工具（JSON 长度 L = 30 tokens）时，一次性合法率约为 e^(−0.03) ≈ 97.0%，看起来还不错；
但当调用一个写入代码或复杂配置的工具（JSON 长度 L = 800 tokens）时，一次性合法率就会指数级衰减到：
P_valid(800) ≈ e^(−800 × 0.001) = e^(−0.8) ≈ 44.9%
也就是说，在长参数工具调用中，超过一半的概率会在中途某一个 Token 上擦枪走火！而在一个需要连续调用 10 次工具的 Agent 任务中，全程不翻车的概率更是只剩下 (0.449)^10 ≈ 0.03%。
这就是为什么我们不能只靠写 Prompt 祈祷模型“请务必输出合法 JSON”，而必须在系统和算法层面介入。
2. 第一层防御：采样期语法约束解码（FSM Constrained Decoding）
我们先从最彻底的底层方案讲起：既然我们已经自己实现了 mini vLLM 的 Sampler 和 LogitsProcessor，能不能在模型每次从 logits 里选 Token 之前，直接把所有会导致 JSON 语法出错的 Token 全部屏蔽掉？
回顾一下我们在《从零实现一个 mini vLLM -- 设计 Sampling 系统》里写过的接口：
for processor in processors:
logits = processor(sequence, logits)
当时我们用 LogitsProcessor 实现了 RepetitionPenaltyProcessor。它的输入是当前 Sequence（包含截至目前已经生成的所有历史 Token）和当前步未归一化的分数向量 zₜ ∈ R^|V|。
现在，我们要实现一个专门服务于 Agent 结构化输出的 FSMGuidedLogitsProcessor。
2.1 形式化语言与掩码概率重归一化
任何一个 JSON Schema，本质上都定义了一个上下文无关文法（Context-Free Grammar, CFG）；如果我们限制最大嵌套深度（比如工具参数最多嵌套 4 层对象），它就可以被完全展开为一个确定性有限状态自动机（Deterministic Finite Automaton, DFA / FSM）：
A = (S, Σ_char, δ_char, s₀, S_accept)
其中 S 是语法状态集合（例如 EXPECT_OBJ_START、IN_KEY_STRING、EXPECT_COLON、IN_VALUE_NUMBER 等），δ_char : S × Σ_char → S ∪ {⊥} 是字符级状态转移函数（⊥ 表示语法非法死状态）。
由于大模型的词表 V 是由子词（Subword Tokens）组成的，每个 Token v ∈ V 对应一串字符序列 str(v) = c₁ c₂ … cₖ。我们可以自然地将字符级转移函数扩展为 Token 级复合转移函数 δ̂ : S × V → S ∪ {⊥}：
δ̂(s, v) = δ_char(… δ_char(δ_char(s, c₁), c₂) …, cₖ)
在解码的第 t 步，设当前已生成的前缀所处的自动机状态为 sₜ₋₁ ∈ S。我们构造一个二值掩码向量 m(sₜ₋₁) ∈ {0, −∞}^|V|：
m_v(sₜ₋₁) = 0，若 δ̂(sₜ₋₁, v) ≠ ⊥；−∞，若 δ̂(sₜ₋₁, v) = ⊥
将其叠加到模型输出的原始 Logits zₜ 上，再经过温度缩放与 Softmax，就得到了语法约束下的条件概率分布 P̃_θ(yₜ | x, y_<t)：
P̃_θ(yₜ = v | x, y_<t) = exp((zₜ,ᵥ + m_v(sₜ₋₁)) / τ) / Σ(v' ∈ V) exp((zₜ,ᵥ' + m_v'(sₜ₋₁)) / τ) = 𝕀[δ̂(sₜ₋₁, v) ≠ ⊥] · P_θ(yₜ = v | x, y_<t) / Σ(v' ∈ V_valid(sₜ₋₁)) P_θ(yₜ = v' | x, y_<t)
看这个公式的最后一项： 只要在每一步保证 V_valid(sₜ₋₁) ≠ ∅，那么无论模型内部原本有 εₜ = 0.1% 还是 50% 的概率想吐出一个非法的 Token，经过掩码 m(sₜ₋₁) 之后，非法 Token 的概率都会变成严格的 0，而原本合法 Token 之间的相对概率比值 P̃(v₁)/P̃(v₂) = P(v₁)/P(v₂) 保持完全不变！
这样一来，前面让你头疼的指数级衰减公式 P_valid(L) = Π(1 - εₜ) 直接变成了恒等式：
P̃_valid(L) = Π(t = 1…L) 1 = 100%
2.2 手写 JSON Schema 驱动的增量状态验证器与 LogitsProcessor
为了让大家彻底看懂 Outlines 或 XGrammar 这类约束解码库的核心原理，我们来亲手写一个轻量但五脏俱全的 JSONSchemaIncrementalValidator 和对应的 SchemaGuidedLogitsProcessor（constrained_decoding.py）。
它的核心思想非常巧妙：给定一个工具参数的 JSON Schema（例如要求包含 "file_path": string 和 "ignore_empty": boolean），我们在每一步检查候选 Token 拼接到当前已生成文本 prefix 之后，是否是一个合法的 JSON 前缀（Valid Partial JSON Prefix），且已经闭合的 Key 是否严格落在该 Schema 允许的 properties 集合内！
from __future__ import annotations
import math
import re
from enum import Enum, auto
from typing import Any, Dict, List, Optional, Set, Tuple
class LexerState(Enum):
"""增量式 JSON 词法/语法状态机核心状态"""
START_ROOT = auto() # 期待顶层 '{'
EXPECT_KEY_OR_END = auto()# 在 '{' 后，期待 '"key"' 或 '}'
EXPECT_KEY = auto() # 在 ',' 后，必须期待 '"key"'（严谨模式禁止尾随逗号！）
IN_KEY_STRING = auto() # 正在读取键名字符串 '"...'
IN_KEY_ESCAPE = auto() # 键名字符串中遇到反斜杠 '\'
AFTER_KEY = auto() # 键名闭合，期待 ':'
EXPECT_VALUE = auto() # 在 ':' 或数组 '['/',' 后，期待合法值开头
IN_VAL_STRING = auto() # 正在读取字符串值
IN_VAL_ESCAPE = auto() # 字符串值中遇到反斜杠 '\'
IN_VAL_PRIMITIVE = auto() # 正在读取数字、true、false、null
AFTER_VALUE = auto() # 值已闭合，期待 ',' 或闭合括号 '}' / ']'
COMPLETE = auto() # 顶层 JSON 对象已完整闭合
INVALID = auto() # 语法错误死状态 (⊥)
class IncrementalJSONSchemaFSM:
"""
增量式 JSON + Schema 约束自动机。
给定任意前缀字符串，以 O(N) 扫描判定其是否属于合法前缀，并约束允许出现的键名与值类型。
"""
PRIMITIVE_PREFIXES = (
"true", "false", "null",
)
NUMBER_REGEX = re.compile(r"^-?(?:0|[1-9]\d*)(?:\.\d*)?(?:[eE][+-]?\d*)?$")
def __init__(self, parameters_schema: Optional[Dict[str, Any]] = None) -> None:
self.schema = parameters_schema or {}
self.allowed_props: Dict[str, Dict[str, Any]] = self.schema.get("properties", {})
self.required_props: Set[str] = set(self.schema.get("required", []))
@classmethod
def _is_valid_primitive_prefix(cls, buf: str, expected_type: Optional[str] = None) -> bool:
"""检查未闭合的原语缓冲区是否是 true/false/null 或合法数字的前缀"""
if not buf:
return False
if expected_type == "boolean":
return any(p.startswith(buf) for p in ("true", "false"))
if expected_type in ("integer", "number"):
if buf in ("-", "-0.", "0.") or buf.endswith(("e", "E", "e+", "e-", "E+", "E-")):
return True
if expected_type == "integer" and ("." in buf or "e" in buf.lower()):
return False
return bool(cls.NUMBER_REGEX.match(buf))
# 通用原语前缀判定
if any(p.startswith(buf) for p in cls.PRIMITIVE_PREFIXES):
return True
if buf in ("-", "-0.", "0.") or buf.endswith(("e", "E", "e+", "e-", "E+", "E-")):
return True
return bool(cls.NUMBER_REGEX.match(buf))
@classmethod
def _is_complete_primitive(cls, buf: str, expected_type: Optional[str] = None) -> bool:
"""当遇到逗号或右括号时，检查原语是否已经是一个完整合法的字面量"""
if not buf or buf.endswith((".", "e", "E", "+", "-")):
return False
if expected_type == "boolean":
return buf in ("true", "false")
if expected_type == "integer":
return bool(re.match(r"^-?(?:0|[1-9]\d*)$", buf))
if expected_type == "number":
return bool(cls.NUMBER_REGEX.match(buf))
return buf in ("true", "false", "null") or bool(cls.NUMBER_REGEX.match(buf))
def validate_prefix(self, text: str) -> Tuple[LexerState, Optional[str]]:
"""
扫描文本前缀，返回 (当前所处状态, 错误原因)。
若返回状态为 LexerState.INVALID，则说明该前缀违反了 JSON 语法或 Schema 约束，对应 logit 应被置为 -inf。
"""
state = LexerState.START_ROOT
stack: List[str] = [] # 存放 '{' 或 '['
current_key_buf: List[str] = []
active_top_key: Optional[str] = None
seen_top_keys: Set[str] = set()
primitive_buf: List[str] = []
def _expected_type_for_active_key() -> Optional[str]:
if len(stack) == 1 and stack[0] == "{" and active_top_key and active_top_key in self.allowed_props:
return self.allowed_props[active_top_key].get("type")
return None
i = 0
n = len(text)
while i < n:
ch = text[i]
if state == LexerState.START_ROOT:
if ch.isspace():
i += 1
continue
if ch == "{":
stack.append("{")
state = LexerState.EXPECT_KEY_OR_END
else:
return LexerState.INVALID, "Root must start with '{'"
elif state in (LexerState.EXPECT_KEY_OR_END, LexerState.EXPECT_KEY):
if ch.isspace():
i += 1
continue
if ch == '"' and state in (LexerState.EXPECT_KEY_OR_END, LexerState.EXPECT_KEY):
current_key_buf.clear()
state = LexerState.IN_KEY_STRING
elif ch == "}" and state == LexerState.EXPECT_KEY_OR_END:
stack.pop()
if not stack:
# 顶层闭合时检查是否漏掉了 required 必填字段
if not self.required_props.issubset(seen_top_keys):
return LexerState.INVALID, "Missing required properties before closing '}'"
state = LexerState.COMPLETE
else:
state = LexerState.AFTER_VALUE
else:
return LexerState.INVALID, f"Unexpected '{ch}' when expecting object key"
elif state == LexerState.IN_KEY_STRING:
if ch == "\\":
state = LexerState.IN_KEY_ESCAPE
elif ch == '"':
key_str = "".join(current_key_buf)
# 在顶层对象中，强制检查键名是否在 Schema 允许的属性列表中！
if len(stack) == 1 and self.allowed_props:
if key_str not in self.allowed_props:
return LexerState.INVALID, f"Key '{key_str}' not in schema properties"
if key_str in seen_top_keys:
return LexerState.INVALID, f"Duplicate key '{key_str}'"
active_top_key = key_str
seen_top_keys.add(key_str)
state = LexerState.AFTER_KEY
elif ord(ch) < 0x20:
return LexerState.INVALID, "Unescaped control char in key"
else:
current_key_buf.append(ch)
# 若仍在 IN_KEY_STRING 且是顶层键，实时检查当前键名前缀是否至少匹配某个合法属性名
if state == LexerState.IN_KEY_STRING and len(stack) == 1 and self.allowed_props:
partial_key = "".join(current_key_buf)
if not any(prop.startswith(partial_key) for prop in self.allowed_props):
return LexerState.INVALID, f"Key prefix '{partial_key}' matches no schema property"
elif state == LexerState.IN_KEY_ESCAPE:
if ch in ('"', "\\", "/", "b", "f", "n", "r", "t", "u"):
current_key_buf.append(ch)
state = LexerState.IN_KEY_STRING
else:
return LexerState.INVALID, f"Invalid escape '\\{ch}'"
elif state == LexerState.AFTER_KEY:
if ch.isspace():
i += 1
continue
if ch == ":":
state = LexerState.EXPECT_VALUE
else:
return LexerState.INVALID, "Expected ':' after key"
elif state == LexerState.EXPECT_VALUE:
if ch.isspace():
i += 1
continue
exp_type = _expected_type_for_active_key()
if ch == '"':
if exp_type and exp_type not in ("string",):
return LexerState.INVALID, f"Key '{active_top_key}' expects {exp_type}, not string"
state = LexerState.IN_VAL_STRING
elif ch == "{":
if exp_type and exp_type not in ("object",):
return LexerState.INVALID, f"Key '{active_top_key}' expects {exp_type}, not object"
stack.append("{")
state = LexerState.EXPECT_KEY_OR_END
elif ch == "[":
if exp_type and exp_type not in ("array",):
return LexerState.INVALID, f"Key '{active_top_key}' expects {exp_type}, not array"
stack.append("[")
state = LexerState.EXPECT_VALUE
elif ch == "]" and stack and stack[-1] == "[":
stack.pop()
state = LexerState.AFTER_VALUE
else:
if exp_type and exp_type in ("string", "object", "array"):
return LexerState.INVALID, f"Key '{active_top_key}' expects {exp_type}"
primitive_buf = [ch]
if not self._is_valid_primitive_prefix(ch, exp_type):
return LexerState.INVALID, f"Invalid value start '{ch}'"
state = LexerState.IN_VAL_PRIMITIVE
elif state == LexerState.IN_VAL_STRING:
if ch == "\\":
state = LexerState.IN_VAL_ESCAPE
elif ch == '"':
state = LexerState.AFTER_VALUE
elif ord(ch) < 0x20:
# 严格拦截未转义的真实换行符 \n (0x0A) 或制表符 \t (0x09)！
return LexerState.INVALID, "Unescaped newline/control char inside JSON string"
elif state == LexerState.IN_VAL_ESCAPE:
if ch in ('"', "\\", "/", "b", "f", "n", "r", "t", "u"):
state = LexerState.IN_VAL_STRING
else:
return LexerState.INVALID, f"Invalid escape '\\{ch}' in value"
elif state == LexerState.IN_VAL_PRIMITIVE:
exp_type = _expected_type_for_active_key()
if ch.isspace() or ch in (",", "}", "]"):
buf_str = "".join(primitive_buf)
if not self._is_complete_primitive(buf_str, exp_type):
return LexerState.INVALID, f"Incomplete primitive '{buf_str}'"
state = LexerState.AFTER_VALUE
continue # 不递增 i，让 AFTER_VALUE 重新处理当前的分隔符或括号
else:
primitive_buf.append(ch)
if not self._is_valid_primitive_prefix("".join(primitive_buf), exp_type):
return LexerState.INVALID, f"Invalid primitive prefix '{''.join(primitive_buf)}'"
elif state == LexerState.AFTER_VALUE:
if ch.isspace():
i += 1
continue
if ch == ",":
state = LexerState.EXPECT_KEY if stack[-1] == "{" else LexerState.EXPECT_VALUE
elif ch == "}" and stack[-1] == "{":
stack.pop()
if not stack:
if not self.required_props.issubset(seen_top_keys):
return LexerState.INVALID, "Missing required properties before closing '}'"
state = LexerState.COMPLETE
else:
state = LexerState.AFTER_VALUE
elif ch == "]" and stack[-1] == "[":
stack.pop()
state = LexerState.AFTER_VALUE
else:
return LexerState.INVALID, f"Expected ',' or closing bracket after value, got '{ch}'"
elif state == LexerState.COMPLETE:
if not ch.isspace():
return LexerState.INVALID, "Extra non-whitespace tokens after complete JSON object"
i += 1
return state, None
class SchemaGuidedLogitsProcessor:
"""
可直接挂载到 mini vLLM Sampling 系统的 LogitsProcessor。
当生成进入 Action Input: 阶段后，自动掩码屏蔽所有破坏 JSON 语法或违反 Schema 的候选 Token。
"""
def __init__(self, vocab: Dict[int, str], parameters_schema: Dict[str, Any]) -> None:
self.vocab = vocab
self.fsm = IncrementalJSONSchemaFSM(parameters_schema)
def __call__(self, generated_prefix: str, logits: List[float]) -> List[float]:
"""
输入当前已生成的 JSON 前缀文本与词表 Logits 向量，将非法 Token 的分数置为 -inf。
"""
masked_logits = list(logits)
for token_id, token_str in self.vocab.items():
candidate_prefix = generated_prefix + token_str
state, _ = self.fsm.validate_prefix(candidate_prefix)
if state == LexerState.INVALID:
masked_logits[token_id] = -math.inf
return masked_logits
仔细阅读上面的 IncrementalJSONSchemaFSM.validate_prefix： 它不仅在词法层面死死卡住了“双引号必须成对”、“字符串里禁止出现裸换行符 0x0A”、“逗号后面禁止直接跟 }（杜绝尾随逗号）”，而且在语义层面卡住了三个极其关键的 Schema 约束：
键名前缀实时剪枝：在
IN_KEY_STRING状态中，如果 Schema 只允许"file_path"和"ignore_empty"，当模型刚吐出"p（想幻觉出"path"）时，any(prop.startswith("p") ...)立刻返回False，意味着模型在打第一个字母时就会被-inf掩码拦住！值类型强制匹配：如果
"ignore_empty"在 Schema 里声明为"boolean"，当模型在冒号后面想输出双引号"true"（字符串）或数字1时，在EXPECT_VALUE状态就会被直接屏蔽，只允许输出t或f！必填项闭合守卫：如果
required集合里的字段还没凑齐，模型想提前输出顶层右花括号}结束对象，状态机会直接判定为INVALID，逼着模型把必填字段写完才能收尾！
3. 第二层防御：解析期确定性修复器（Deterministic JSON Repair）
第一层防御虽然能达到 100% 的语法合规率，但它有一个前提：你必须能控制底层推理引擎的 Sampler（比如使用我们自己部署的 mini vLLM）。
如果用户把 mini Agent Runtime 配置成了连接某个远程黑盒 API，或者对端服务没有开启 Constrained Decoding，模型吐回来的依然可能是一段带着小毛病的“半残 JSON”。
这时候，难道只要 json.loads 报错，我们就立刻发回给大模型重跑一轮吗？
绝对不是。统计表明，大模型生成的非法 JSON 中，超过 85% 都属于高度模式化的确定性语法瑕疵。我们完全可以在 Runtime 本地用几毫秒的 CPU 时间，写一个基于单趟扫描状态机的 DeterministicJSONRepairer，在不消耗任何额外 Token 的情况下直接把它们修好！
我们来梳理一下这个本地确定性修复器要处理的 5 类高频故障及其状态机转换规则：
剥离包裹噪音与尾随闲聊：自动跳过 JSON 前面的 Markdown 代码块标记 ````json` 或说明文字，并在顶层对象闭合后截断后面的所有废话。
字符串内的裸控制字符自动转义：当扫描器处于
IN_STRING状态时，如果遇到裸的换行符\n（0x0A）、回车符\r（0x0D）或制表符\t（0x09），自动将其替换为双字符转义序列\\n、\\r、\\t。Python 字面量与单引号注释清理：当扫描器处于**非字符串（Outside String）**状态时，将独立的标识符
True、False、None原地替换为 JSON 标准的true、false、null，并自动剥离行尾的// ...单行注释。尾随逗号（Trailing Comma）自动消除：当在非字符串状态下看到
}或]时，向前回溯输出缓冲区，如果前面紧挨着（忽略空白符）一个多余的,，直接把那个逗号弹出丢弃！未闭合引号与括号栈自动补全（Auto-Closing Stack）：如果模型因为触及
max_tokens或一时糊涂在结尾漏写了引号或右括号，扫描器在到达文本末尾（EOF）时：若仍处于
IN_STRING状态，先补一个闭合双引号"；若末尾残留了悬空的冒号
:（比如{"a":），自动补一个null；若残留了悬空的逗号,，自动删掉该逗号；最后将括号栈
stack中尚未匹配的{和[按后进先出（LIFO）顺序依次补齐对应的}和]！
我们把这套逻辑写成干净利落的 json_repair.py：
from __future__ import annotations
import json
import re
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple
@dataclass
class RepairResult:
"""JSON 确定性修复与解析结果"""
success: bool
parsed_data: Optional[Dict[str, Any]] = None
repaired_json: str = ""
applied_fixes: List[str] = None
error_message: str = ""
def __post_init__(self) -> None:
if self.applied_fixes is None:
self.applied_fixes = []
class DeterministicJSONRepairer:
"""
零 LLM 开销的本地确定性 JSON 词法/括号栈修复器。
专门解决大模型高频出现的裸换行、Python True/None 字面量、尾随逗号、注释及末尾漏括号问题。
"""
@classmethod
def parse_or_repair(cls, raw_text: str) -> RepairResult:
cleaned = raw_text.strip()
# 1. 快速路径：如果原本就是完美合法的 JSON 对象，直接零开销返回
try:
obj = json.loads(cleaned)
if isinstance(obj, dict):
return RepairResult(success=True, parsed_data=obj, repaired_json=cleaned)
except json.JSONDecodeError:
pass
# 2. 定位第一个 '{'，剥离前缀 Markdown 标记（如 ```json）
start_idx = cleaned.find("{")
if start_idx == -1:
return RepairResult(
success=False,
error_message="No opening brace '{' found in output.",
)
candidate = cleaned[start_idx:]
repaired_str, fixes = cls._scan_and_repair(candidate)
try:
decoder = json.JSONDecoder()
obj, end_pos = decoder.raw_decode(repaired_str)
if isinstance(obj, dict):
return RepairResult(
success=True,
parsed_data=obj,
repaired_json=repaired_str[:end_pos],
applied_fixes=fixes,
)
return RepairResult(
success=False,
repaired_json=repaired_str,
applied_fixes=fixes,
error_message=f"Repaired JSON root is {type(obj).__name__}, expected object (dict).",
)
except json.JSONDecodeError as exc:
return RepairResult(
success=False,
repaired_json=repaired_str,
applied_fixes=fixes,
error_message=f"{exc.msg} at line {exc.lineno} column {exc.colno} (char {exc.pos})",
)
@classmethod
def _scan_and_repair(cls, text: str) -> Tuple[str, List[str]]:
out: List[str] = []
fixes: List[str] = []
stack: List[str] = [] # 记录未闭合的 '{' 或 '['
in_string = False
escape = False
i = 0
n = len(text)
while i < n:
ch = text[i]
if in_string:
if escape:
# 如果模型写了非法的反斜杠转义（例如正则表达式里的 \d 或路径里的 \w），补一个反斜杠保护它
if ch not in ('"', "\\", "/", "b", "f", "n", "r", "t", "u"):
out.append("\\")
fixes.append(f"escaped_invalid_backslash_\\{ch}")
out.append(ch)
escape = False
elif ch == "\\":
out.append(ch)
escape = True
elif ch == '"':
out.append(ch)
in_string = False
elif ch == "\n":
out.append("\\n")
fixes.append("escaped_raw_newline_in_string")
elif ch == "\r":
out.append("\\r")
fixes.append("escaped_raw_cr_in_string")
elif ch == "\t":
out.append("\\t")
fixes.append("escaped_raw_tab_in_string")
else:
out.append(ch)
i += 1
continue
# 以下为非字符串（Outside String）状态
if ch == '"':
in_string = True
out.append(ch)
i += 1
continue
# 剥离单行 // 注释
if ch == "/" and i + 1 < n and text[i + 1] == "/":
fixes.append("stripped_line_comment")
i += 2
while i < n and text[i] != "\n":
i += 1
continue
if ch in ("{", "["):
stack.append(ch)
out.append(ch)
i += 1
continue
if ch in ("}", "]"):
# 检查并移除前面的尾随逗号（Trailing Comma），例如 {"a": 1, }
cls._pop_trailing_comma(out, fixes)
if stack:
top = stack[-1]
if (top == "{" and ch == "}") or (top == "[" and ch == "]"):
stack.pop()
out.append(ch)
# 如果顶层 '{' 已经闭合，立即停止扫描，忽略后面可能跟随的 ``` 或模型废话
if not stack:
break
else:
# 括号类型错配：自动补齐栈顶期待的闭合括号
expected_close = "}" if top == "{" else "]"
out.append(expected_close)
stack.pop()
fixes.append(f"fixed_mismatched_bracket_expected_{expected_close}")
continue
i += 1
continue
# 识别并替换 Python 风格的 True / False / None 字面量
if ch in ("T", "F", "N"):
for py_kw, json_kw in (("True", "true"), ("False", "false"), ("None", "null")):
if text.startswith(py_kw, i):
after_idx = i + len(py_kw)
if after_idx == n or not (text[after_idx].isalnum() or text[after_idx] == "_"):
out.append(json_kw)
fixes.append(f"converted_python_{py_kw}_to_{json_kw}")
i = after_idx
break
else:
out.append(ch)
i += 1
continue
out.append(ch)
i += 1
# 扫描到达末尾（EOF）时的自动收尾补全
if in_string:
if escape:
out.append("\\")
out.append('"')
fixes.append("closed_unterminated_string")
# 清理末尾可能悬空的逗号或冒号
cls._pop_trailing_comma(out, fixes)
cls._fix_dangling_colon(out, fixes)
# 按 LIFO 顺序依次闭合所有未关闭的 '{' 和 '['
while stack:
opener = stack.pop()
closer = "}" if opener == "{" else "]"
out.append(closer)
fixes.append(f"auto_closed_missing_{closer}")
return "".join(out), fixes
@staticmethod
def _pop_trailing_comma(out: List[str], fixes: List[str]) -> None:
idx = len(out) - 1
while idx >= 0 and out[idx].isspace():
idx -= 1
if idx >= 0 and out[idx] == ",":
out.pop(idx)
fixes.append("removed_trailing_comma")
@staticmethod
def _fix_dangling_colon(out: List[str], fixes: List[str]) -> None:
idx = len(out) - 1
while idx >= 0 and out[idx].isspace():
idx -= 1
if idx >= 0 and out[idx] == ":":
out.append("null")
fixes.append("filled_dangling_colon_with_null")
4. 第三层防御：带 JSONPath 坐标的递归 Schema 校验与精准反馈
经过第 1 层（采样期掩码）或第 2 层（确定性语法修复）之后，我们拿到的 parsed_data 已经保证是一个语法合法的 Python dict 了。
但在复杂工具调用中，还有最后一关：深度语义 Schema 校验（Deep Schema Validation）。
在第 2 篇的 ToolRegistry 里，我们只检查了最外层的 required 字段是否存在。但在真实工具中，参数往往包含：
枚举约束（
enum: ["asc", "desc"]）数值范围约束（
minimum: 1, maximum: 100）数组元素类型约束（
items: {"type": "integer"}）嵌套子对象约束（
properties内部再嵌套object）
如果模型传了一个 {"limit": 500, "sort_order": "random", "tags": [123]}，我们不能只笼统地报一句 Invalid arguments，而是应该像优秀的编译器（比如我们在 Linux C++ 系列里聊过的 GCC 诊断信息）一样，一次性找出所有违规字段的精确 JSONPath 路径、违规原因、实际传入值 与 期望约束，生成一份结构化的诊断报告喂回给模型。
我们来实现这个深度 Schema 校验与诊断生成器（schema_validator.py）：
from __future__ import annotations
import json
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
@dataclass
class ValidationIssue:
"""单条精确到 JSONPath 的参数违规记录"""
path: str # 例如 "$.config.port" 或 "$.files[2]"
message: str # 具体违规原因
expected: str # Schema 期望的类型或约束
actual_value: Any # 模型实际传入的错误值
class SchemaValidator:
"""
递归式 JSON Schema 校验与自修复 Prompt 诊断生成器。
支持 type, required, enum, minimum, maximum, minLength, items, properties 及类型自动矫正。
"""
@classmethod
def validate_and_coerce(
cls, data: Any, schema: Dict[str, Any], path: str = "$"
) -> tuple[Any, List[ValidationIssue]]:
issues: List[ValidationIssue] = []
coerced = cls._validate_node(data, schema, path, issues)
return coerced, issues
@classmethod
def _validate_node(
cls, val: Any, schema: Dict[str, Any], path: str, issues: List[ValidationIssue]
) -> Any:
expected_type = schema.get("type")
# 1. 轻量级安全类型矫正 (Safe Type Coercion)
if expected_type == "integer" and isinstance(val, str) and val.strip().lstrip("-").isdigit():
val = int(val.strip())
elif expected_type == "number" and isinstance(val, str):
try:
val = float(val.strip())
except ValueError:
pass
elif expected_type == "boolean" and isinstance(val, str):
if val.lower() in ("true", "1"):
val = True
elif val.lower() in ("false", "0"):
val = False
# 2. 基础类型校验（注意 Python 中 isinstance(True, int) 为 True，需专门排除 bool）
if expected_type:
type_ok = True
if expected_type == "string" and not isinstance(val, str):
type_ok = False
elif expected_type == "integer" and (not isinstance(val, int) or isinstance(val, bool)):
type_ok = False
elif expected_type == "number" and (not isinstance(val, (int, float)) or isinstance(val, bool)):
type_ok = False
elif expected_type == "boolean" and not isinstance(val, bool):
type_ok = False
elif expected_type == "array" and not isinstance(val, list):
type_ok = False
elif expected_type == "object" and not isinstance(val, dict):
type_ok = False
if not type_ok:
issues.append(
ValidationIssue(
path=path,
message=f"Type mismatch at '{path}'",
expected=f"type '{expected_type}'",
actual_value=val,
)
)
return val
# 3. Enum 枚举值校验
if "enum" in schema and val not in schema["enum"]:
issues.append(
ValidationIssue(
path=path,
message=f"Value not in allowed enum list at '{path}'",
expected=f"one of {schema['enum']}",
actual_value=val,
)
)
# 4. 数值区间校验 (minimum / maximum)
if isinstance(val, (int, float)) and not isinstance(val, bool):
if "minimum" in schema and val < schema["minimum"]:
issues.append(
ValidationIssue(
path=path,
message=f"Number is below minimum at '{path}'",
expected=f">= {schema['minimum']}",
actual_value=val,
)
)
if "maximum" in schema and val > schema["maximum"]:
issues.append(
ValidationIssue(
path=path,
message=f"Number exceeds maximum at '{path}'",
expected=f"<= {schema['maximum']}",
actual_value=val,
)
)
# 5. 数组元素递归校验
if isinstance(val, list) and "items" in schema:
item_schema = schema["items"]
return [
cls._validate_node(item, item_schema, f"{path}[{idx}]", issues)
for idx, item in enumerate(val)
]
# 6. 对象属性递归校验
if isinstance(val, dict):
props = schema.get("properties", {})
required = schema.get("required", [])
for req_key in required:
if req_key not in val:
issues.append(
ValidationIssue(
path=f"{path}.{req_key}",
message=f"Missing required property '{req_key}'",
expected=json.dumps(props.get(req_key, {"type": "any"}), ensure_ascii=False),
actual_value="<MISSING>",
)
)
new_dict: Dict[str, Any] = {}
for k, v in val.items():
if props and k not in props and not schema.get("additionalProperties", False):
issues.append(
ValidationIssue(
path=f"{path}.{k}",
message=f"Unknown property '{k}' is not allowed",
expected=f"valid keys: {list(props.keys())}",
actual_value=k,
)
)
continue
if k in props:
new_dict[k] = cls._validate_node(v, props[k], f"{path}.{k}", issues)
else:
new_dict[k] = v
# 自动补齐未传但定义了 default 的可选字段
for k, prop_spec in props.items():
if k not in new_dict and "default" in prop_spec:
new_dict[k] = prop_spec["default"]
return new_dict
return val
@classmethod
def format_repair_prompt(
cls,
tool_name: str,
raw_input: str,
issues: List[ValidationIssue],
parse_error: Optional[str] = None,
) -> str:
"""生成面向 LLM 自修复的结构化编译器风格诊断报告"""
lines = [f"[SCHEMA VALIDATION FAILED] Tool '{tool_name}' invocation could not be executed."]
if parse_error:
lines.append(f"- Syntax Error: {parse_error}")
lines.append(f"- Raw Malformed Input: {raw_input!r}")
if issues:
lines.append(f"- Found {len(issues)} schema violation(s):")
for idx, iss in enumerate(issues, 1):
lines.append(
f" {idx}. Path `{iss.path}`: {iss.message} "
f"(got `{iss.actual_value!r}`, expected {iss.expected})."
)
lines.append(
"Please fix ONLY the reported issues and re-invoke the tool with a strictly valid JSON Action Input."
)
return "\n".join(lines)
5. 实战验证：三层纵深防御全链路压测
现在我们来写一个综合测试脚本 demo_structured_output.py，把我们在这一篇手写的三个核心模块全部跑一遍：
实验一（第 1 层：采样期 FSM Constrained Decoding 实测）：构造一个模拟词表和原始 Logits（其中模型因为习惯问题，给非法属性名
"path"、未转义换行符和尾随逗号赋予了极高的原始 Logit 分数），验证SchemaGuidedLogitsProcessor如何在每一步精准地将非法 Token 掩码为-inf，逼迫采样器走出一条 100% 合规的 JSON 路径。实验二（第 2 层：解析期
DeterministicJSONRepairer实测）：喂给修复器一段集齐了“5 种经典毒瘤”（Markdown 包裹、PythonTrue/None、代码块里的裸换行符\n、单行注释//、尾随逗号,以及结尾漏掉两个右括号}）的极度破损文本，验证它在 0 Token 消耗、小于 0.2 毫秒 内一次性原地修复为合法字典！实验三（第 3 层：深层 Schema 诊断与自修复闭环）：当字段类型和数值范围越界时，验证
SchemaValidator生成精确到.sort_by的 JSONPath 诊断报告。
import json
import math
from constrained_decoding import SchemaGuidedLogitsProcessor
from json_repair import DeterministicJSONRepairer
from schema_validator import SchemaValidator
# 定义一个带枚举、数值区间与嵌套对象的工具参数 Schema
SEARCH_TOOL_SCHEMA = {
"type": "object",
"properties": {
"query": {"type": "string"},
"sort_by": {"type": "string", "enum": ["relevance", "date", "stars"]},
"options": {
"type": "object",
"properties": {
"max_results": {"type": "integer", "minimum": 1, "maximum": 50},
"include_archived": {"type": "boolean", "default": False},
},
"required": ["max_results"],
},
},
"required": ["query", "sort_by"],
}
def test_layer1_fsm_constrained_decoding() -> None:
print("=== 实验一：测试第 1 层 (mini vLLM 采样期 FSM Constrained Decoding) ===")
# 模拟一个包含合法与非法 JSON 片段的子词表
vocab = {
0: '{"',
1: '"query"',
2: '"keyword"', # 非法属性名（不在 Schema properties 中）
3: ": ",
4: '"vllm"',
5: ", ",
6: '"sort_by"',
7: "}", # 在必填项未齐时试图提前闭合对象
8: "true", # 在期待 string 的字段上传入 boolean
9: '"stars"}',
}
processor = SchemaGuidedLogitsProcessor(vocab=vocab, parameters_schema=SEARCH_TOOL_SCHEMA)
# 场景 A：前缀是 '{'，模型原本给非法键名 '"keyword"' (id=2) 和提前闭合 '}' (id=7) 极高的分数
prefix_a = "{"
raw_logits_a = [1.0, 2.5, 9.8, 0.5, 0.1, 0.1, 2.0, 8.5, 0.1, 0.1]
masked_a = processor(prefix_a, raw_logits_a)
valid_tokens_a = {vocab[i]: masked_a[i] for i in range(len(vocab)) if not math.isinf(masked_a[i])}
print(f"前缀 `{prefix_a}` 下:")
print(f" - 原始最高分 Token : {vocab[2]} (logit=9.8), {vocab[7]} (logit=8.5)")
print(f" - FSM 掩码后幸存的合法 Token: {valid_tokens_a}")
# 场景 B：前缀已经写完 '{"query": "vllm", "sort_by": '，模型想输出布尔值 'true' (id=8)
prefix_b = '{"query": "vllm", "sort_by": '
raw_logits_b = [0.1, 0.1, 0.1, 0.1, 3.2, 0.1, 0.1, 0.1, 9.5, 4.0]
masked_b = processor(prefix_b, raw_logits_b)
valid_tokens_b = {vocab[i]: masked_b[i] for i in range(len(vocab)) if not math.isinf(masked_b[i])}
print(f"前缀 `{prefix_b}` 下:")
print(f" - 原始最高分 Token : {vocab[8]} (logit=9.5, 类型错误!)")
print(f" - FSM 掩码后幸存的合法 Token: {valid_tokens_b}")
def test_layer2_deterministic_repair() -> None:
print("\n=== 实验二：测试第 2 层 (解析期 0-Token 确定性语法修复器) ===")
# 构造一段集齐 5 种语法错误的破损模型输出：
# 1. ```json 包裹 2. 字符串内包含真实换行符 3. 单行 // 注释
# 4. Python 的 True/None 字面量 5. 尾随逗号 6. 结尾漏掉两个右花括号 }}
broken_llm_output = '''```json
{
"query": "def hello():
return 42", // 这里模型直接在字符串里敲了回车，还写了行尾注释
"sort_by": "stars",
"options": {
"max_results": 20,
"include_archived": True,
},
'''
res = DeterministicJSONRepairer.parse_or_repair(broken_llm_output)
print(f"修复是否成功 : {res.success}")
print(f"触发的修复项 : {res.applied_fixes}")
print(f"修复后的 JSON: {res.repaired_json}")
print(f"解析后的字典 : {res.parsed_data}")
def test_layer3_schema_validation() -> None:
print("\n=== 实验三：测试第 3 层 (带 JSONPath 坐标的深度 Schema 校验与自修复反馈) ===")
# 语法合法，但包含 3 处语义违规 + 1 处可自动矫正的字符串数字
bad_semantic_payload = {
"query": "mini vllm",
"sort_by": "popularity", # 违规 1：不在枚举 ['relevance', 'date', 'stars'] 中
"unknown_flag": 123, # 违规 2：未知的幻觉字段
"options": {
"max_results": "999", # 自动从 str 矫正为 int 999，但触发违规 3：超过 maximum 50
},
}
coerced, issues = SchemaValidator.validate_and_coerce(bad_semantic_payload, SEARCH_TOOL_SCHEMA)
diagnostic_prompt = SchemaValidator.format_repair_prompt("search_repos", json.dumps(bad_semantic_payload), issues)
print(diagnostic_prompt)
if __name__ == "__main__":
test_layer1_fsm_constrained_decoding()
test_layer2_deterministic_repair()
test_layer3_schema_validation()
我们来跑一下这段验证脚本，看看三层防御的实际运行输出：
=== 实验一：测试第 1 层 (mini vLLM 采样期 FSM Constrained Decoding) ===
前缀 `{` 下:
- 原始最高分 Token : "keyword" (logit=9.8), } (logit=8.5)
- FSM 掩码后幸存的合法 Token: {'"query"': 2.5, '"sort_by"': 2.0}
前缀 `{"query": "vllm", "sort_by": ` 下:
- 原始最高分 Token : true (logit=9.5, 类型错误!)
- FSM 掩码后幸存的合法 Token: {'"vllm"': 3.2, '"stars"}': 4.0}
=== 实验二：测试第 2 层 (解析期 0-Token 确定性语法修复器) ===
修复是否成功 : True
触发的修复项 : ['escaped_raw_newline_in_string', 'stripped_line_comment', 'converted_python_True_to_true', 'removed_trailing_comma', 'removed_trailing_comma', 'auto_closed_missing_}', 'auto_closed_missing_}']
修复后的 JSON: {"query": "def hello():\n return 42", "sort_by": "stars", "options": {"max_results": 20, "include_archived": true}}
解析后的字典 : {'query': 'def hello():\n return 42', 'sort_by': 'stars', 'options': {'max_results': 20, 'include_archived': True}}
=== 实验三：测试第 3 层 (带 JSONPath 坐标的深度 Schema 校验与自修复反馈) ===
[SCHEMA VALIDATION FAILED] Tool 'search_repos' invocation could not be executed.
- Found 3 schema violation(s):
1. Path `$.sort_by`: Value not in allowed enum list at '$.sort_by' (got `'popularity'`, expected one of ['relevance', 'date', 'stars']).
2. Path `$.unknown_flag`: Unknown property 'unknown_flag' is not allowed (got `'unknown_flag'`, expected valid keys: ['query', 'sort_by', 'options']).
3. Path `$.options.max_results`: Number exceeds maximum at '$.options.max_results' (got `999`, expected <= 50).
Please fix ONLY the reported issues and re-invoke the tool with a strictly valid JSON Action Input.
看看上面这三个实验的结果：
在实验一里，即便模型原本给非法字段名
"keyword"打了9.8分、给提前闭合的}打了8.5分（因为必填项还没写），SchemaGuidedLogitsProcessor依然毫不留情地把它们全部置成了-inf，只留下合法的"query"和"sort_by"参与采样！在实验二里，那段混杂了裸回车、
//注释、True、两个尾随逗号和漏写两个右花括号的“残废 JSON”，不需要重调一次大模型，直接在本地被DeterministicJSONRepairer毫秒级修好并成功解析。在实验三里，面对枚举越界和嵌套字段数值超标，
SchemaValidator给出了精确到$.options.max_results的编译器级诊断提示，模型下一轮看一眼就能改对。
6. 进阶思考：为什么工業级引擎会预编译 Token 掩码表？
如果你仔细看我们在第 2 节写的 SchemaGuidedLogitsProcessor.__call__，你会注意到一个可以继续深挖的性能细节（这也是我们在《mini vLLM》第 11 篇讲 Sampling 性能时提过的思路）：
在上面的教学版实现里，我们在每一步生成时遍历了词表 for token_id, token_str in self.vocab.items()，去测试 generated_prefix + token_str 是否合法。如果词表大小 |V| = 128,000，在 Python 层每生成一个 Token 都做十万次字符串检查，CPU 开销会明显拖慢 Decode 的 TPOT（Time Per Output Token）。
在真正的生产级引擎（比如 vLLM 集成的 XGrammar 或 Outlines）中，是怎么把这个开销压到接近 0 毫秒的？
答案是 离线预编译状态-词表位图索引（Precomputed State-to-Vocab Bitmask Cache）。
因为对于给定的 JSON Schema，确定性有限状态自动机 A 的状态集合 S_DFA 是有限的。引擎会在收到请求准备 Prefill 的那一刻（甚至在注册工具 Schema 的那一刻），在后台把每一个自动机状态 s ∈ S_DFA 允许通过的所有 Token ID 集合提前算好，压缩存储为一个长度为 ⌈|V| / 32⌉ 的 uint32 位图（Bitmask）。
这样到了 Decode 的每一个 Step，Sampler 只需要根据上一步选出的 Token yₜ₋₁ 查表做一次 O(1) 的状态跳转 sₜ = δ̂(sₜ₋₁, yₜ₋₁)，然后直接取出状态 sₜ 对应的预编译位图，扔给一个几行代码的 GPU CUDA Kernel 做按位掩码即可！
理解了我们上面手写的 IncrementalJSONSchemaFSM，你就已经彻底看穿了所有结构化输出引擎最核心的灵魂；所谓的预编译加速，不过是在我们这个状态转移函数 δ̂(s, v) 外面加了一层提前算好的记忆化缓存表（Memoization Table）而已。
小结
这一篇我们聚焦在 Agent Runtime 中最容易导致链路脆弱断裂的环节——结构化工具调用的解析与生成可靠性。
我们没有停留在简单的 try ... except json.JSONDecodeError，而是把底层推理引擎的采样系统与上层 Runtime 解析器打通，构建了完整的三层防御：
从概率上算清了自由采样的脆弱性：推导出长度为 L 的结构化参数在微错误率 ε 下遵循 P_valid(L) ≈ e^(−Lε) 的指数衰减定律，解释了为什么长参数调用必须引入系统级干预。
第 1 层（采样期强约束）：联动
mini vLLM的LogitsProcessor，实现了IncrementalJSONSchemaFSM与SchemaGuidedLogitsProcessor，在每一步通过掩码 m(sₜ₋₁) ∈ {0, −∞}^|V| 将非法语法、未知键名、错误值类型与提前闭合括号的概率直接清零。第 2 层（解析期确定性修复）：针对无法控制底层 Sampler 的外部 API 场景，手写了单趟词法扫描与括号栈修复器
DeterministicJSONRepairer，零 Token 成本原地修复裸换行符、单行注释、True/False/None、尾随逗号以及未闭合的字符串和括号。第 3 层（语义期精准诊断）：实现了支持安全类型矫正与递归校验的
SchemaValidator，能够生成带精确JSONPath（如$.options.max_results）的编译器级报错反馈。
解决了“怎么让模型稳定、准确地吐出工具调用”之后，下一篇（第 5 篇）我们将把目光投向工具被解析出来之后的那一环：《异步 Tool Executor：并发、超时熔断与幂等重试》。
当模型在一轮思考中同时发出 5 个网页抓取或文件搜索请求时，我们怎么把现在阻塞慢吞吞的 for tc in tool_calls 改造成基于 asyncio 协程池、支持信号量并发限流、单工具独立超时熔断（Timeout Cancellation）与指数退避重试的高吞吐执行引擎？下一篇见！
