大模型应用中的结构化输出稳定性工程实践：从 JSON Schema 约束、失败重试策略到解析兜底与异常样本回放

在业务里接大模型，很多问题不出在“答得对不对”，而是出在“能不能稳定给出机器可消费的结果”。接口要落库，要走规则引擎，要进工单系统，这时候一句自然语言解释没法直接用。真正卡人的，往往是结构化输出的稳定性。

我最近在一个信息抽取服务里把这件事重新做了一遍，目标很直接：让模型稳定输出符合约束的 JSON，对解析失败、字段缺失、类型漂移这类问题有统一处理办法，并且把坏样本沉淀下来做回放复测。说实话，前期我以为加个“请按 JSON 返回”就差不多了，实测并不是这样。

这篇文章给一套可复现方案，包含四块：

• 用 JSON Schema 收紧输出边界
• 用分级重试控制失败成本
• 用解析兜底扛住线上脏输出
• 用异常样本回放做持续修正

场景不复杂。输入是一段用户文本，输出是工单分类结果：

• category: 工单大类
• priority: 优先级
• sentiment: 情绪判断
• summary: 50 字以内摘要
• need_callback: 是否需要人工回访

短句先说结论：结构化输出稳定性不是靠单点提示词，而是靠约束、校验、重试、回放一起做。

---

一、问题是怎么暴露出来的

先看最开始的版本，Prompt 很简单：

请阅读用户反馈内容，提取分类、优先级、情绪、摘要和是否需要回访，按 JSON 格式输出。

线上跑了两天，问题集中在这几类：

• 返回 ```json 包裹，解析器直接报错
• 字段名漂移，比如 needCallback、callback_needed
• 布尔值写成 "yes"、"否"
• summary 超长，甚至带解释语句
• category 输出了训练集中没定义过的新值
• 少字段，偶发缺 priority

我把 500 条线上样本回捞后做了统计，初版结果如下：

指标	数值
首次解析成功率	81.6%
Schema 校验通过率	74.2%
可直接入库率	71.8%
平均重试次数	0

71.8% 没法上线。很现实。

因为后面接的是工单系统，任何一条坏 JSON 都不是“小误差”，而是整条流程中断。单看模型回答内容，很多其实“意思是对的”，但工程上依然是失败。

---

二、先把输出约束写清楚：JSON Schema 是第一道门

如果你的目标是结构化消费，先别急着堆 few-shot，先把目标格式定义成可验证对象。我的做法是直接维护一份 Schema，作为 Prompt、解析器、测试集校验的统一基准。

1.1 Schema 定义

下面是一个可直接使用的 Schema 示例：

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "type": "object",
  "required": ["category", "priority", "sentiment", "summary", "need_callback"],
  "additionalProperties": false,
  "properties": {
    "category": {
      "type": "string",
      "enum": ["billing", "technical", "logistics", "product", "other"]
    },
    "priority": {
      "type": "string",
      "enum": ["P0", "P1", "P2", "P3"]
    },
    "sentiment": {
      "type": "string",
      "enum": ["positive", "neutral", "negative"]
    },
    "summary": {
      "type": "string",
      "maxLength": 50
    },
    "need_callback": {
      "type": "boolean"
    }
  }
}

这里有两个点很实用。

一个是 additionalProperties: false。它能直接拦住“模型自己多想了一列”的情况。另一个是枚举值收紧，不让 priority 输出成 high、urgent 这类自由文本。

1.2 Prompt 中怎么嵌入 Schema

我不建议把整份 Schema 原样全贴进去，太长时会影响生成稳定性。常见做法是抽出关键约束，保留字段、类型、枚举、长度限制，再配一个合法示例。

我在线上用的 Prompt 模板大概长这样：

你是信息抽取器。任务是从用户反馈中提取结构化字段。

输出要求：
1. 只输出一个 JSON 对象，不要输出 Markdown，不要输出解释。
2. 字段必须完整，且只能包含以下字段：
   - category: billing | technical | logistics | product | other
   - priority: P0 | P1 | P2 | P3
   - sentiment: positive | neutral | negative
   - summary: 字符串，50字以内
   - need_callback: 布尔值 true 或 false
3. 不允许输出额外字段。
4. 如果信息不足，也必须给出最合理值，不要留空。

合法示例：
{"category":"billing","priority":"P2","sentiment":"negative","summary":"用户反馈扣费异常，要求核查账单","need_callback":true}

待处理文本：
{{input_text}}

实测看，“只输出一个 JSON 对象” 和 “不要 Markdown” 这两句要写得很硬。很多模型默认喜欢包一层代码块，这会让上游解析变脆。

---

三、不要把所有失败都当成一种失败

结构化输出失败，处理方式不能一刀切。我一般把失败拆成三层：

• 生成层失败：模型没按 JSON 输出
• 语法层失败：看着像 JSON，但解析不过
• 语义层失败：能解析，但不符合 Schema

分层后，重试策略才好设计。

2.1 失败分类器

先给一个简单的 Python 分类器：

import json
from jsonschema import validate, ValidationError


def classify_output(raw_text: str, schema: dict):
    raw_text = raw_text.strip()

    if not raw_text.startswith("{"):
        return "generation_failure", None, "output_not_json_object"

    try:
        obj = json.loads(raw_text)
    except json.JSONDecodeError as e:
        return "syntax_failure", None, str(e)

    try:
        validate(instance=obj, schema=schema)
    except ValidationError as e:
        return "semantic_failure", obj, e.message

    return "success", obj, None

这段逻辑很短，但线上很有用。因为每一种失败都对应不同修复动作。别混着处理。

2.2 重试不是重复调用一遍

很多系统的“重试”其实只是把同一个 Prompt 再发一次，这样收益不稳定，还会白白烧 token。我的做法是按失败类型切换策略。

生成层失败

症状：输出有解释、有前后缀、有代码块。

处理：

• 把系统提示再收紧
• 附上“错误原因”回灌给模型
• 强制要求“仅返回 JSON 对象”

示例重试 Prompt：

上一次输出不符合要求，原因是：输出不是纯 JSON 对象。
请修正。

要求：
- 仅返回 JSON 对象
- 不要输出 ```json
- 不要解释
- 字段必须为 category, priority, sentiment, summary, need_callback

待处理文本：
{{input_text}}

语法层失败

症状：缺逗号、单双引号混用、布尔值写错。

处理：

• 优先走本地修复器
• 修不动再进模型二次纠错

语义层失败

症状：字段值越界、缺字段、类型不对。

处理：

• 把校验错误原文传回模型
• 指定只修复错误字段
• 保留已经正确的字段，减少二次漂移

重试 Prompt 示例：

你上一次输出的 JSON 已经可以解析，但不符合字段约束。
错误信息：
{{validation_error}}

请在保留正确字段含义的前提下，输出修正后的完整 JSON。
只返回 JSON 对象。
原始文本：
{{input_text}}

上一次输出：
{{previous_json}}

这一步很管用。因为模型知道自己错在哪，比盲重试更稳。

---

四、解析兜底别写成一团 if-else

线上场景里，总会遇到“差一点就能用”的输出。比如：

```json
{"category":"billing","priority":"P1","sentiment":"negative","summary":"用户反馈重复扣费","need_callback":"true"}

这类结果严格说不合格，但直接丢弃也可惜。我的建议是加一层轻量解析兜底，原则是：**只做确定性修复，不做猜测性修复。**

### 3.1 兜底清洗器

```python
import json
import re


def sanitize_raw_output(text: str) -> str:
    text = text.strip()

    # 去掉 markdown 代码块
    text = re.sub(r"^```json\s*", "", text)
    text = re.sub(r"^```\s*", "", text)
    text = re.sub(r"\s*```$", "", text)

    # 截取首个 JSON 对象
    start = text.find("{")
    end = text.rfind("}")
    if start != -1 and end != -1 and end > start:
        text = text[start:end + 1]

    return text


def normalize_fields(obj: dict) -> dict:
    alias_map = {
        "needCallback": "need_callback",
        "callback_needed": "need_callback"
    }

    normalized = {}
    for k, v in obj.items():
        k = alias_map.get(k, k)
        normalized[k] = v

    if "need_callback" in normalized and isinstance(normalized["need_callback"], str):
        val = normalized["need_callback"].strip().lower()
        if val in {"true", "yes", "1", "是"}:
            normalized["need_callback"] = True
        elif val in {"false", "no", "0", "否"}:
            normalized["need_callback"] = False

    return normalized

这个清洗器做的事情很有限：去代码块、截 JSON 主体、修复少量别名、归一化布尔值。别做太多。

因为修得越多，越容易把真正的错误掩盖掉。到最后你看到“成功率很高”，其实是本地逻辑偷偷改了业务语义。

3.2 完整处理主流程

from jsonschema import validate, ValidationError


def process_output(raw_text: str, schema: dict):
    cleaned = sanitize_raw_output(raw_text)

    try:
        obj = json.loads(cleaned)
    except json.JSONDecodeError:
        return {
            "status": "failed",
            "stage": "syntax",
            "data": None
        }

    obj = normalize_fields(obj)

    try:
        validate(instance=obj, schema=schema)
        return {
            "status": "success",
            "stage": "validated",
            "data": obj
        }
    except ValidationError as e:
        return {
            "status": "failed",
            "stage": "semantic",
            "error": e.message,
            "data": obj
        }

工程里我一般会把 raw_text、cleaned_text、normalized_obj 都打日志，这样后面排查不会靠猜。

---

五、一套可复现的重试编排

我比较推荐“最多两次模型调用”的控制方式：首调一次，失败后按类型补一次。超过两次，收益开始掉，延迟和成本反而上来。

4.1 状态机设计

用户输入
  -> 首次生成
    -> 解析与校验
      -> 成功，返回
      -> 失败分类
        -> 本地清洗可修复，重新校验
          -> 成功，返回
          -> 失败，进入定向重试
        -> 不可修复，进入定向重试
          -> 再次解析与校验
            -> 成功，返回
            -> 失败，降级处理并记录异常样本

4.2 参考代码

import time


def call_llm(prompt: str) -> str:
    # 伪代码，替换成你的模型调用
    raise NotImplementedError


def build_retry_prompt(input_text: str, failure_type: str, error_message: str, previous_output: str):
    if failure_type == "generation_failure":
        return f"""
上一次输出不是纯 JSON 对象，原因：{error_message}
请重新输出。
只返回一个 JSON 对象，不要解释，不要 Markdown。
字段：category, priority, sentiment, summary, need_callback
输入文本：{input_text}
""".strip()

    return f"""
上一次输出存在结构化错误。
错误信息：{error_message}
请修正后返回完整 JSON。
只返回 JSON 对象。
输入文本：{input_text}
上一次输出：{previous_output}
""".strip()


def run_pipeline(input_text: str, schema: dict):
    t0 = time.time()

    prompt = build_initial_prompt(input_text)
    raw1 = call_llm(prompt)

    cleaned1 = sanitize_raw_output(raw1)
    failure_type, obj1, error1 = classify_output(cleaned1, schema)

    if failure_type == "success":
        return {"ok": True, "data": obj1, "retry": 0, "latency_ms": int((time.time()-t0)*1000)}

    if failure_type in {"syntax_failure", "semantic_failure"}:
        try:
            parsed = json.loads(cleaned1)
            normalized = normalize_fields(parsed)
            validate(instance=normalized, schema=schema)
            return {"ok": True, "data": normalized, "retry": 0, "latency_ms": int((time.time()-t0)*1000)}
        except Exception:
            pass

    retry_prompt = build_retry_prompt(input_text, failure_type, error1, cleaned1)
    raw2 = call_llm(retry_prompt)
    cleaned2 = sanitize_raw_output(raw2)
    failure_type2, obj2, error2 = classify_output(cleaned2, schema)

    if failure_type2 == "success":
        return {"ok": True, "data": obj2, "retry": 1, "latency_ms": int((time.time()-t0)*1000)}

    return {
        "ok": False,
        "data": None,
        "retry": 1,
        "latency_ms": int((time.time()-t0)*1000),
        "final_error": error2,
        "raw_outputs": [raw1, raw2]
    }

这里少了 build_initial_prompt，你可以直接用前面那版模板。重点不在调用 SDK，而在处理流程。

---

六、降级策略要提前定，不要等线上报错再补

不是所有失败都值得继续卡住主流程。如果业务允许，建议准备一层降级输出。比如：

• category 无法判断时回 other
• priority 默认 P2
• need_callback 默认 false
• summary 取原文前 30 字做截断摘要

我会把降级结果单独打标：

{
  "category": "other",
  "priority": "P2",
  "sentiment": "neutral",
  "summary": "用户反馈内容较长，系统暂未完成结构化提取",
  "need_callback": false,
  "_degraded": true
}

如果后续系统不能接受额外字段，那 _degraded 不入主对象，单放元数据。这个细节要提前跟消费方约好。

有个小缺点也得承认：降级会吞掉部分真实问题，所以我只建议在强 SLA 的同步接口里用，离线任务最好保留失败态，后面统一修复。

---

七、异常样本回放，才是稳定性往上走的关键

前面这些做完，能把失败率压下去，但如果不做样本回放，问题会重复出现。今天修了布尔值，明天冒出新字段；今天修了代码块，后天开始输出注释。

我一般会把异常样本分成两类：

• 结构异常：JSON 格式、字段名、字段类型问题
• 业务异常：字段值虽合法，但和人工期望不符

本文先讲结构异常回放。

6.1 异常样本表设计

建议最少存这些字段：

字段	说明
sample_id	样本 ID
input_text	原始输入
prompt_version	Prompt 版本
model_name	模型名
raw_output	原始输出
cleaned_output	清洗后输出
failure_stage	generation/syntax/semantic
error_message	错误详情
created_at	失败时间

落库后，每次改 Prompt 或改清洗器，都可以把历史异常样本跑一遍。

6.2 回放脚本示例

from typing import List, Dict


def replay_samples(samples: List[Dict], schema: dict):
    results = []
    for sample in samples:
        result = run_pipeline(sample["input_text"], schema)
        results.append({
            "sample_id": sample["sample_id"],
            "ok": result["ok"],
            "retry": result.get("retry", 0),
            "final_error": result.get("final_error")
        })
    return results


def summarize_replay(results: List[Dict]):
    total = len(results)
    success = sum(1 for x in results if x["ok"])
    retry_used = sum(1 for x in results if x["retry"] > 0)

    return {
        "total": total,
        "success_rate": round(success / total, 4) if total else 0,
        "retry_rate": round(retry_used / total, 4) if total else 0
    }

回放别只看总成功率。还要看哪些错误类型下降了，哪些还在原地。

---

八、实测结果对比

我用同一批 500 条真实工单文本做了三轮对比，模型保持一致，只改工程策略。

7.1 版本定义

• V0：仅 Prompt 要求输出 JSON
• V1：增加字段枚举约束 + Schema 校验
• V2：增加本地清洗 + 失败分类重试 + 异常样本回放修正

7.2 核心指标

指标	V0	V1	V2
首次解析成功率	81.6%	89.4%	90.2%
Schema 校验通过率	74.2%	86.8%	95.6%
可直接入库率	71.8%	84.9%	94.1%
平均耗时	820ms	860ms	1040ms
平均 token 成本	1.00x	1.05x	1.18x

这个结果我自己比较满意。V2 的耗时和成本都有增加，但换来了 20 多个点的可入库率提升，在线上是划算的。

再拆一下 V2 中失败样本的去向：

处理结果	占比
首次即成功	90.2%
本地清洗后成功	2.9%
定向重试后成功	4.0%
最终失败并降级	2.9%

没想到的是，本地清洗吃掉了不少低级问题，性价比很高。真正需要二次模型调用的比例，比一开始预估的低。

---

九、监控指标怎么埋

如果你只看接口 200 成功率，那结构化输出的问题会被藏起来。建议单独埋这些指标：

• llm_structured_first_pass_success_rate
• llm_structured_schema_pass_rate
• llm_structured_retry_rate
• llm_structured_degrade_rate
• llm_structured_field_missing_rate
• llm_structured_enum_violation_rate

我自己常看的看板分法是按 Prompt 版本、模型版本、业务来源分桶。这样能快速看出问题是某次模板更新引入的，还是某类输入天然更脏。

举个例子，后面我们发现投诉类文本的 summary 超长比例明显高于普通咨询，原因不是模型突然变差，而是用户原文里常带大量时间线和情绪句，摘要约束更容易被冲掉。找到这个点后，单独给投诉类样本补了两个 few-shot，summary 越界率从 8.7% 降到 3.1%。

---

十、一个更稳的工程落地建议

如果你准备在生产环境里接结构化输出，我建议按下面的顺序上：

1）先定 Schema，再写 Prompt

很多团队反过来做，先让模型跑起来，后面再补字段规范。这样会积累很多历史兼容包袱。

2）解析、校验、重试逻辑独立成中间层

别散落在业务代码里。后面接第二个抽取任务时，能直接复用。

3）异常样本固定回放周期

一周一次就够用。频率太高，团队容易疲劳；太低，坏样本会越积越多。

4）把“成功”定义清楚

不是模型回了内容就算成功，也不是 JSON 能 parse 就算成功。要以“可被下游直接消费”为准。

---

十一、总结

大模型做结构化输出，最怕的是“看起来差不多能用”。工程上，差一点就是不能用。

一套更稳的方案，我建议至少包含这些部分：

• 明确 Schema，约束字段名、类型、枚举和长度
• 对失败做分层，不同类型走不同修复策略
• 本地只做确定性兜底，不替模型乱猜
• 重试要带错误信息，不要盲重试
• 异常样本要回放，把修复动作沉淀成可复现流程

如果你现在的系统还停留在“提示模型输出 JSON”，可以先补上 Schema 校验和异常样本回放，这两步见效通常最快。

我自己的体感是，结构化输出稳定性一旦做顺，后面的评估、回归测试、版本对比都会轻松很多。基础打稳，后面省事不少。