markdown 流里嵌 JSON 如何校验？fluxmend 用字符级 FSM 把这事做明白了

处理markdown 流里嵌 JSON 如何校验？fluxmend 用字符级 FSM 把这事做明白了这类问题时，先确认目标场景，再按步骤核对配置或玩法细节。

markdown 流里嵌 JSON 怎么校验？fluxmend 用字符级 FSM 把这事做明白了

〇、先说清楚：fluxmend 解决的是哪类问题

这点必须放在最前面，因为它和很多人脑子里"LLM 结构化输出"不是一回事。

现在主流的"结构化输出"方案是 厂商原生 schema 约束，比如：

1. OpenAI 的 response_format: json_schema
2. Anthropic 的 tool_use 强制结构
3. 各家 Constrained Decoding / Grammar-based sampling

这类方案的特点是：让 LLM 整段 response 就是结构化数据本身（一个完整的 JSON、一个 tool_call）。它解决的是"我要拿到一个干净可 parse 的结构体"。

但实际业务里有一大类场景不是这样的 —— 你需要的是一段 markdown 流式文本，中间嵌着结构化数据块，用来给前端做组件渲染。比如一个 AI 助手在 markdown 回复里穿插图表、卡片、地图标记、商品位等富组件：

根据你的问题，我整理了几个关键指标：<metric>{"name": "latency_p99", "value": 42.5, "unit": "ms"}</metric>从趋势看，Q3 比 Q2 改善明显：<chart>{"type": "line", "series": [...], "xAxis": [...]}</chart>具体店铺表现如下：<shop>{"id": 101, "name": "老王面馆", "rating": 4.8}</shop><shop>{"id": 102, "name": "川味小炒", "rating": 4.6}</shop>

这种场景下：

1. markdown 自由文本要实时流式显示给用户（首字延迟敏感，打字机效果）
2. 中间嵌入的 <metric> / <chart> / <shop> JSON 必须结构化校验通过才能喂给前端组件渲染 —— 一个字段错就渲染崩、或者更糟，渲染出错误的数据
3. 一段流里多个结构化块、多种格式（JSON + XML + Regex）混合，每个对应不同组件
4. LLM 偶尔会写错（漏括号、字段名拼错、True 写成 true），但不能因为一块错了就整段崩 —— markdown 文本得照常显示，坏了的那块该修就修、修不好就降级

原生 schema 约束解决不了这个 —— 它要么把整段 response 变成 JSON（markdown 文本没了，前端没法做打字机渲染），要么用 tool_call 但打断文本流（一段 response 只能塞一个 tool_call，且 tool_call 字段独立、不混在 markdown 里，前端拿不到"文本+组件"的混合流）。

fluxmend 解决的就是这一类问题：markdown 流式输出中间带结构化数据，结构化块用于组件渲染等场景。它不替代原生 schema 约束，而是补上"混合流"这个空白场景。

一、痛点：为什么"markdown + 嵌套 JSON"流式这么难搞？

写过 Agent / RAG / 工具调用的同学应该都踩过这几个坑：

1. 流式输出不能等：用户在等首字延迟，你却要把整段 JSON 收完才能 json.loads，体验直接拉胯。
2. LLM 偷偷给你"加戏"：明明只要 JSON，它给你塞一段 markdown 包裹；明明要 true，它给你输出 True；明明字段叫 nearby_pois，它写成了 nearby_poi。
3. 修都修不动：用 json.loads 失败后，再丢给 LLM "请修复"，二次调用贵且慢，还可能修不对。
4. 多组件混合：一段流里同时有 <shop>、<map>、<metric> 多种结构化块，每个都要校验，怎么办？
5. 想在流到 </tag> 时立刻执行一个函数（比如把解析结果塞进队列 / 触发后续工具），但又不想打断 LLM 的文本流 —— 怎么做？

业界常见的解法无非三种，都有硬伤：

fluxmend 的思路完全不同：把校验做成字符级 FSM，每个 token 进来就推进状态机，错了立刻进入分层修复流程，markdown 文本部分全程透传、不阻塞。

二、fluxmend 是什么？

一句话定义：

翻译过来就是：面向"LLM 输出中嵌入结构化组件"场景的，流式 FSM 校验 + 分层修复库。

特性一览：

1. 字符级流式校验，无需等 close 标签再 parse
2. 三层修复：规则修复 → json-repair 库 → LLM-as-Repairer → 检查点回滚
3. 多格式：JSON / XML / Regex / CFG (lark) / 自定义 FSM 插件
4. Schema 感知：支持 Pydantic 类、JSON Schema dict、DSL
5. Per-tag handler：每个 tag 可挂一个函数，</tag> 闭合时自动执行，不打断 LLM 文本流
6. OpenAI / Anthropic SDK 都能接（LLMClient Protocol）
7. 异步修复模式，多组件流不互相阻塞
8. mypy strict + 310+ 测试，ruff 干净
9. Apache-2.0 协议

三、30 秒上手

from pydantic import BaseModelfrom fluxmend import FluxmendclassShop(BaseModel):id: intname: strwith Fluxmend(schemas=[("shop", Shop)]) as guard:for chunk in llm_stream("推荐一家店"):for event in guard.feed(chunk):if event.type == "text":print(event.content, end="")# 已校验的安全文本elif event.type == "repair_applied":r = event.contentprint(f"n[修复 layer={r.layer}] {r.original!r} ->{r.repaired!r}")result = guard.result# {"shop": [Shop(id=101, name="...")]}

任何能产文本的源都能接 —— agno / pydantic-ai / langgraph / 裸 OpenAI / Anthropic SDK / 甚至纯文本流。

四、核心架构：两层流水线

fluxmend 把工作拆成两层：

LLM 流 ──▶ Enhancement Layer (可选) ──▶ Core Layer ──▶ Events│││修开标签残缺│DetectionFSM (识别标签边界)│shop> → <shop> │GrammarValidator (字符级 FSM)│<shp> → <shop> │分层修复循环

Enhancement Layer（前置过滤器）：LLM 偶尔会把 <shop> 写成 shop>（漏 <），或者把标签名拼错。这层用 Trie 树识别候选标签名，在 close 标签确认后插入缺失的 <。

Core Layer（核心校验）：DetectionFSM 负责识别 <tag> / </tag> 边界，进入组件后切换到对应格式的 GrammarValidator，逐字符推进 FSM。状态机一掉链子（某个字符没法转移），立刻进入修复流程。

五、分层修复：把便宜方案放前面

这是 fluxmend 最有讲究的地方 —— 修复不等于"再问一次 LLM"。它按成本从低到高分两层：

流式阶段（标签还没闭合，不能插括号）

1. Schema-aware Local Repair：布尔/null 大小写（True → true）、类型推断（"42" → 42）
2. 字段名纠错：编辑距离 1 匹配 schema 字段（nearby_poi → nearby_pois）

这两步纯规则、纳秒级，绝大部分低级错误在这一层就消化了。

</tag> 闭合阶段（安全插括号）

3. json-repair 库：修语法层（少括号 / 尾逗号 / 引号）
4. LLM Repair：让 LLM 带 schema context 重写出错组件
5. Checkpoint Rollback：兜底 —— 把丢弃的文本以未校验 text 事件吐给消费方，至少不丢内容

每一步都会发 repair_applied 事件，带着 layer、original、repaired 字段，全程可审计。

六、几个值得展开的设计

1. Token-boundary safe

为什么不在流式阶段就插括号修 JSON？因为 LLM 的 token 切分可能把 {"a": 切成两个 token，你在中间插 } 会破坏后续 token 的拼接。fluxmend 把所有"需要插字符"的修复都推迟到 </tag> 闭合后 —— 这时候组件内容已经全部到齐，怎么改都安全。

2. 检查点（Checkpoint）

在结构边界（值完成、分隔符后）打快照 (position, fsm_state, schema_path)，修复失败时 pop 最近一个检查点回滚状态。检查点上限 64 个，长流不漏内存。

3. LLM-as-Repairer

Local Repair 修不了的语义错（比如字符串该是数字且不可强转），才把 schema JSON + 出错文本一起丢给 LLM，prompt 大致是：

Fix the following JSON to match the schema.Schema: <schemaJSON>Broken JSON: <fullcomponenttext>Rules:1. Output ONLY the fixed JSON, no explanation2. Field names must match the schema exactly3. Field types must be correct4. Do not add or remove fields5. Output valid JSON

LLM 返回的修复结果会逐字符回放进 FSM，每个字符都必须能转移、最终状态必须落在结构边界 —— 不通过就降级到 Checkpoint Rollback。流永远不会因为 LLM 修复失败而崩溃。

4. 异步修复模式（不阻塞流）

多组件流里，如果一个 </tag> 触发 LLM Repair，后续组件都得等。开 async_repair=True：

guard = Fluxmend(schemas=[("shop", Shop), ("map", MapMark)],try_times=2,llm_client=client,async_repair=True,# ← 组件 close 时立刻发 component_end(verified="pending"))

主 feed() 循环立刻返回，后续组件正常处理；待修复组件在后台单线程池里跑，close() 时统一 flush 事件。FIFO 顺序，结果仍然按出现顺序对齐。

5. 生产级：FluxmendPool

Web 框架（FastAPI / Flask / Django）里多请求并发，直接 FluxmendPool：

from fluxmend import FluxmendPoolpool = FluxmendPool(schemas=[("shop", Shop), ("map", MapMark)],try_times=2,llm_client=client,handlers={"shop": process_shop},pool_size=10,)# 同步with pool.acquire() as guard:for chunk in text_stream:guard.feed(chunk)result = guard.close()# 异步（FastAPI / Starlette）asyncwith pool.aacquire() as guard:asyncfor chunk in text_stream:await guard.afeed(chunk)result = await guard.aclose()

Grammars 在 pool init 时编译一次，每个 checkout 拿到的是 reset() 过的实例 —— 无跨请求状态泄漏，线程安全。

6. Per-tag Handler：在 LLM 流里插入函数调用，但不打断流

这是 fluxmend 一个挺关键、但容易被忽略的卖点。每个 tag 可以挂一个 handler 函数，模型边吐边校验，到了 </tag> 闭合时自动跑 handler，但 LLM 的文本流不会因此中断。

defprocess_shop(shop: dict) ->dict:return {"id": shop["id"], "name_upper": shop["name"].upper()}guard = Fluxmend(schemas=[("shop", Shop), ("map", MapMark)],handlers={"shop": process_shop},# map 保持默认行为)

为什么不会中断 LLM 流？ 关键是校验/解析/handler 执行 和 LLM 文本流是解耦的：

LLM stream chunk ──▶ guard.feed(chunk) ──▶ events（text / component_* / handler 调用）│└─ LLM 自己继续往下吐，不知道下游干了什么

1. LLM 那边该怎么流还怎么流（你 for chunk in llm_stream 照常迭代）
2. feed(chunk) 内部走 FSM，遇到 <tag> 进入组件态、遇到 </tag> 触发 handler
3. handler 同步执行完才会返回后续 events，但它跑的是用户代码、不是 LLM 调用，所以只要你的 handler 不阻塞（不发 HTTP / 不查 DB / 不调 LLM），流就不会卡

也就是说，handler 是一个 "在 LLM 流的某个时点插入一段你的逻辑" 的钩子，但它本身不会去打断 LLM 的流 —— LLM 那条 stream 该来多少 chunk 还是来多少。

唯一会"卡流"的情况 —— handler 里干慢 IO：

defprocess_shop(shop: dict) ->dict:# ❌ 这种会阻塞流resp = requests.post("https://api.xxx.com/save", json=shop)return resp.json()# ❌ 这种也会阻塞，等于在 LLM 流里串了第二个 LLM 调用defprocess_shop(shop: dict) ->dict:return llm.translate(shop["name"])

handler 是同步的（在 feed() 调用栈里跑），慢操作会把后续 chunk 的处理推迟。LLM stream 本身没断（SDK 那边 buffer 还在攒），但你的消费侧会感觉到延迟。正确做法：handler 里只做纯计算 / 内存操作（重命名、补默认值、转大写、过滤字段），慢 IO 推到流结束之后或塞进队列异步处理。

和厂商原生 function calling 的对比 —— 这套"prompt 引导 tag + handler 钩子"的路子，本质是用文本流模拟 function call，但保留文本流的连续性：

适合"我想要 function call 的语义，但不想让 LLM 流被打断 / 不想绑厂商"的场景。

handler 抛异常会被吞掉，存 ""（保持长度对齐前端 zip），同时发 handler_error 事件方便诊断 —— 不让一个组件的 handler bug 拖垮整段流。

七、适用场景

1. Agent 框架：用 prompt 引导模型吐 <tool>{...}</tool>，handler 在 </tool> 闭合时触发工具执行 —— 文本流不中断、不绑厂商
2. 替代 native function calling：当你需要 function call 的语义，但不想让 tool_call 打断 markdown 文本流，或想多组件混合、字段名纠错时
3. Chat UI / 富组件渲染（核心场景）：LLM 流式吐 markdown，中间嵌 <chart> / <card> / <metric> / <map> 等结构化块，前端按 tag 渲染对应组件 —— markdown 走打字机效果，结构化块校验通过才进组件，坏了自动修
4. RAG 管道：修复嵌入在 markdown 响应里的元数据块
5. Chat UI：实时显示已校验文本，修复事件丢日志
6. 批量评测：跑 1000 次 LLM 调用，统计修复率、审计失败 case
7. 多模态流：一段流里 JSON + XML + 自由文本混着来，按 tag 分别校验

八、装一下

pip install fluxmend# 可选 extras：pip install "fluxmend[cfg]"# lark CFG 支持pip install "fluxmend[dev]"# pytest, mypy, ruff, pre-commit

仓库地址：github.com/luvrix/flux…（开源 Apache-2.0）

九、为什么我觉得值得推荐

市面上做"LLM 结构化输出"的方案，要么绑死某家厂商（OpenAI Structured Output）、要么只做事后修补（json-repair）、要么直接限制模型能力（constrained decoding）。fluxmend 是少数把"流式校验 + 分层修复 + 多格式"三件事一起认真做的库：

1. 不绑厂商：LLMClient Protocol 一接就行
2. 不阻塞流：字符级 FSM + 异步修复
3. 不放弃语义：schema-aware 修复能改字段名、改类型
4. 不丢内容：兜底 Checkpoint Rollback 仍然把原文以 text 事件吐出
5. 可审计：每一步修复都有事件，调起 Agent bug 来不再抓瞎

如果你也在做 Agent / RAG / 工具调用，正在被 LLM 流式 JSON 折腾，强烈建议试一下。