在泥潭中打捞：Raw Text 工具解析器与 Fallback 兜底方案

(第 50 篇：Agent 协议之韧性)

上一篇我们讲述了在拥有顶级受限解码 API 支持的情况下，Schema 校验是多么丝滑。但在实际部署 Agent 的残酷工况中（尤其是客户需要在内网离线跑 Llama 3、Qwen 2 等本地开源模型时），模型通常并不原生支持 OpenAI 风格的 tool_calls 物理拦截。

此时，我们就只能让大模型在普通的对话流中输出纯文本（Raw Text）。这就需要极客最硬核的手艺——“文本打捞术”：从一堆废话中，精准地切割出机器指令。

0. 先把解析器放回安全语境：外部文本默认不可信

在开始“怎么解析”之前，先把威胁模型写清楚。 只要你允许 raw text 里携带“工具指令”，你就必须承认：

1. 模型输出不是可信边界。
2. 观测（网页、RAG、日志）也不是可信边界。
3. 解析成功只是“提取”，不是“授权”。

间接提示词注入（IPI）不是理论问题， 它已经在真实检索与工具输出链路里出现， 并且会把“上下文内容”伪装成“高优先级指令”。 citeturn0academia12

只要你允许 raw text 里包含“工具指令”，你就必须承认一个事实：

• raw text 可能来自不可信数据（用户输入、网页、RAG 结果）。
• 不可信数据里可能夹带间接提示词注入（IPI）。

因此解析器必须遵守两条硬规则：

1. 解析成功不等于可执行：解析器输出仍然是不可信输入，执行前必须走 allowlist/权限/审计。
2. 数据区块必须隔离：对检索/网页片段要用隔离标签包裹，并在审计里记录来源与时间戳。

1. 原生约定的 Tag Pattern：XML vs JSON 的工程较量

当大模型失去原生 API 接口支持，业界目前有两种最稳固的流派协议：一种是 Markdown JSON Block 协议，另一种是 Anthropic 和大码量 Agent 群体最推崇的 XML 标签协议。

1.0 先说结论：协议不是美学，是失败模式管理

你选择协议时，真正要比较的是：

1. “半截输出”时你还能不能抢救出一条完整的指令。
2. “混入废话”时你能不能从噪音里可靠切出边界。
3. “参数里包含代码/引号/反引号”时你会不会自爆。

“更优雅”不重要。 重要的是：解析失败时系统会不会进入重试风暴。

1.1 为什么 XML 比 JSON 更适合“打捞”？

JSON 的解析是“全或无”的：丢一个逗号，整个 json.loads() 就会崩溃抛出异常。而 XML（或者是自定义标签）具有极强的局部可识别性。

XML 核心优势：

• 起始原子性：看到 <tool_call> 就知道动作开始了，无关前文废话。
• 容错边界：即使 args 内部包含了极其复杂的特殊字符（如 shell 脚本里的引号），只要我们通过非贪婪正则匹配到 </tool_call>，就能完成闭合。

1.2 协议选择表：你应该在什么场景用哪一种

2. 暴力美学：Fuzzy Tool Parser 的实现

当 Agent 引擎接收到一段包含啰唆废话、解释信息以及工具标签的“泥潭文本”后，我们要利用多级正则进行拆解。

2.1 【核心代码】具备自愈能力的正则表达式提取器

这是一个能从 Markdown 和 XML 中暴力打捞指令的 Python 实现：

import re
import json

class FuzzyToolParser:
    """
    一个无视废话、只看指令的文本解析器。
    它不仅通过正则找标签，还负责修补碎片化的 JSON 字符串。
    """
    def __init__(self):
        # 兼容 XML 标签与 Markdown 代码块
        self.xml_pattern = re.compile(r"<tool_call>\s*<name>(.*?)<\/name>\s*<args>(.*?)<\/args>\s*<\/tool_call>", re.DOTALL)
        self.md_json_pattern = re.compile(r"```json\s*(\{.*?\})\s*```", re.DOTALL)

    def extract(self, text: str) -> list:
        tool_calls = []
        
        # 1. 尝试从 XML 标签中打捞
        for match in self.xml_pattern.finditer(text):
            tool_calls.append({
                "name": match.group(1).strip(),
                "args": self._lenient_json_parse(match.group(2).strip())
            })
            
        # 2. 如果 XML 没捞到，尝试从 Markdown 代码块里找
        if not tool_calls:
            for match in self.md_json_pattern.finditer(text):
                tool_calls.append(self._lenient_json_parse(match.group(1)))
                
        return tool_calls

    def _lenient_json_parse(self, raw_str: str):
        """
        宽容解析器：
        1. 尝试直接解析。
        2. 若失败，尝试修补尾部大括号。
        3. 若依然失败，按纯文本包装。
        """
        try:
            return json.loads(raw_str)
        except json.JSONDecodeError:
            # 暴力修补：防止模型输出中断
            fixed_str = raw_str.strip()
            if not fixed_str.endswith("}"): fixed_str += "}"
            try:
                return json.loads(fixed_str)
            except:
                return {"raw_text_args": raw_str}

2.2 失败模式与治理点：解析失败会变成重试风暴

3. 流式截断中的“残肢补全”逻辑

如果你做的是极速流式（Streaming）截断，情况会变得更糟。当网络突然中断或 Token 达到上限，你可能只拿到了： <tool_call><name>run_shell</name><args>rm -rf /

真正的工业级实现不是“多写几个正则”， 而是一个状态机：

1. 进入捕获态：看到 <tool_call>。
2. 进入字段态：捕获 <name>、<args>。
3. 进入闭合态：看到 </tool_call> 才提交。
4. 流结束但仍在捕获态：触发“修补策略”，但必须降级为只读模式。

状态机修补的核心原则是： 宁可少执行，也不要误执行。 修补出来的指令必须在安全层再过一遍 allowlist 与权限判定。

实现上可以有一个守护协程： 流结束但 is_capturing 仍为 True 时， 尝试注入 </args></tool_call> 完成结构闭合， 但闭合后的产物只能进入 shadow mode 进行只读工具， 禁止直接进入写入型工具路径。

3.2 降级路径：解析失败时如何继续推进任务

解析连续失败时，系统不应空转，更不应继续写入型工具。 建议最小降级策略：

1. 进入 shadow mode：只允许只读工具（ls/cat/grep/status）。
2. 把 parse error 与期望格式 one-shot 示例回注给模型。
3. 超过阈值触发断路器，要求人类介入或 handoff 给专用 agent。

这三个动作都必须写入审计链（观测/审计），否则你无法复盘为什么“突然停止执行”。

3.3 输入验证与 guardrails：解析器后面必须有第二道门

解析器解决的是“从文本里抠出结构”。 安全层解决的是“这个结构能不能执行”。

建议至少做三层防线：

1. allowlist：只允许一小部分工具名（写入型工具默认禁用）。
2. 参数校验：对路径、URL、命令做长度/字符集/危险模式约束。
3. 审计与回放：把每次工具调用的来源与上下文 hash 记录下来，便于追责与回归测试。

这类“输入验证与防护栏”在 agentic 系统里被明确建议采用纵深防御。 citeturn0search1turn0search16

4. 纠错回路：如何教育“不听话”的模型？

如果解析器彻底失败（比如模型输出了一串无法辨认的乱码），你绝对不能让程序空转。

极客的 Fallback 策略：

1. 注入惩罚 Observation：[PARSE_ERROR]: 我完全无法理解你刚才输出的格式，请你重新检查 System Prompt 里的格式要求，严格按照 <tool_call> 标签输出！。
2. 强制降级：如果连续 2 次解析失败，系统应主动修改当前的 System Message，增加一个极其简单的 One-Shot 例子（示例），强行把模型的概率分布拉回正轨。

更进一步： 当系统进入连续失败状态时， Runner 应该把“写入型工具”硬性锁死， 只允许只读工具收集证据， 直到解析与验证稳定恢复。

本章精粹

1. 文本即通信：不要迷信 SDK。在 Agent 层面，网络传输的只是字节流，你要有从字节泥潭里抠出逻辑的能力。
2. 鲁棒性大于优雅：在离线小模型场景下，一个沾满正则、看似不那么“优雅”的解析器，往往能救你的 Agent 一命。
3. ** XML 是绝佳的中继格式**：尤其是在处理包含代码片段的参数时。

掌握了文本打捞术，你的 Agent 就不再依赖顶级的商业 API。它能在 5GB 显存的老笔记本上，用开源的 Llama 3 依然能稳健地执行代码、管理文件。

下一章，我们将彻底揭开操作系统的面纱：【终端劫持与 PTY：Agent 是如何伪装成人类接管你的 iTerm2 的？】。我们要开始写 C 代码，去控制伪终端了！

(本文完 - 深度解析系列 16 / 全文约 1600 字) (注：建议将本章的 FuzzyToolParser 配合 pre-commit 钩子，定期扫描测试集中的极端文本案例。)

参考资料（写作核验）

• Anthropic streaming messages: https://docs.anthropic.com/claude/reference/messages-streaming
• IPI in the wild: https://arxiv.org/abs/2601.07072
• Prompt injection best practices (AWS): https://docs.aws.amazon.com/pdfs/prescriptive-guidance/latest/llm-prompt-engineering-best-practices/llm-prompt-engineering-best-practices.pdf