大模型应用中的结构化输出稳定性治理:从 JSON Schema 约束、重试修复到线上异常兜底的工程实践

在业务里接大模型,很多问题不是“答得对不对”,而是“能不能稳定落到程序里”。文本回答看起来正常,服务一解析就报错,这种情况我遇到过太多次。很常见。

比如客服质检、工单分类、信息抽取、Agent 工具参数生成,这些场景最后都要落成结构化数据。字段缺失、类型漂移、枚举值跑偏、嵌套层级错位,线上一多就会变成事故。说实话,模型效果评估时如果只看语义正确率,基本会高估上线后的真实可用率。

这篇文章我不讲空泛方法,直接给一套我在项目里用过的可复现方案:Schema 约束 + 输出校验 + 分级重试修复 + 线上异常兜底 + 指标观测。重点放在工程细节和实测结果。


一、问题定义:结构化输出为什么总是不稳

先把问题拆开。所谓结构化输出不稳定,通常不是单一错误,而是多类错误混在一起:

  • JSON 根本无法解析,常见于多输出了说明文字、Markdown 包裹、尾逗号
  • JSON 能解析,但字段不全,比如少了 priority
  • 字段名变了,比如 user_sentiment 变成 sentiment
  • 类型不对,比如置信度应该是 float,结果给了 "high"
  • 枚举值越界,比如约束是 low|medium|high,模型回了 urgent
  • 嵌套对象结构不一致,尤其是数组里对象字段缺失

在离线测试里,这些错误看起来只是几个 bad case。到了线上,请求量一上来,问题会集中冒出来。更麻烦的是,解析失败和业务失败往往不在同一个地方暴露:模型服务返回 200,应用服务却在 JSON decode 或 schema validate 时报错。

短句说透:能生成,不等于能消费。


二、治理目标:不是追求 100% 完美,而是提高“可消费率”

我一般把目标拆成四层指标:

1)Raw JSON 成功率

模型输出是否能被标准 JSON 解析器直接解析。

2)Schema 验证通过率

解析后,是否满足字段、类型、枚举、层级等约束。

3)业务可用率

即使 Schema 通过,字段语义也可能不符合业务要求。比如摘要太长、命中标签为空、时间格式不符合下游要求。这一层要加业务规则。

4)最终成功率

加上修复、重试、兜底后的最终可用比例。这才更接近真实线上表现。

我自己的经验是,很多团队只统计最后成功率,看上去数字不错,但中间实际消耗了大量重试 token 和延迟。这样上线后成本会偏高,排障也困难。


三、整体方案:四层防线

先给架构思路:

  1. 生成前约束:Prompt 明确输出规则,配合 JSON Schema 或函数调用格式约束
  2. 生成后校验:使用统一校验器做 JSON parse + Schema validate + business rule check
  3. 失败后修复:按错误类型走轻量修复、定向重试、降级重试
  4. 线上兜底:默认值填充、人工审核队列、业务保守策略

别省这步。

很多不稳定问题,单纯改 Prompt 解决不了。因为线上请求分布比测试集复杂得多,用户输入一旦含噪、跨语种、字段歧义、上下文污染,模型输出就容易漂。


四、先从 Schema 设计入手:别把约束写得太松

结构化输出治理,第一步不是调模型,而是把 Schema 写清楚。

我建议 Schema 至少包含以下信息:

  • 必填字段 required
  • 字段类型 type
  • 枚举范围 enum
  • 字符串长度 minLength/maxLength
  • 数值范围 minimum/maximum
  • 嵌套对象定义 properties
  • 是否允许额外字段 additionalProperties

下面用一个工单分类场景举例。输入是一段用户投诉文本,模型输出结构化工单:

{
  "ticket_type": "refund",
  "priority": "high",
  "sentiment": "negative",
  "summary": "用户反馈订单已取消但仍被扣款,要求退款",
  "confidence": 0.91,
  "needs_human_review": false
}

对应的 JSON Schema:

TICKET_SCHEMA = {
    "type": "object",
    "additionalProperties": False,
    "required": [
        "ticket_type",
        "priority",
        "sentiment",
        "summary",
        "confidence",
        "needs_human_review"
    ],
    "properties": {
        "ticket_type": {
            "type": "string",
            "enum": ["refund", "delivery", "invoice", "product", "other"]
        },
        "priority": {
            "type": "string",
            "enum": ["low", "medium", "high"]
        },
        "sentiment": {
            "type": "string",
            "enum": ["positive", "neutral", "negative"]
        },
        "summary": {
            "type": "string",
            "minLength": 5,
            "maxLength": 120
        },
        "confidence": {
            "type": "number",
            "minimum": 0,
            "maximum": 1
        },
        "needs_human_review": {
            "type": "boolean"
        }
    }
}

这里我建议把 additionalProperties 设为 False。原因很直接:字段一旦放松,下游通常会悄悄吞掉异常,等业务统计出问题时才发现模型偷偷加了字段或改了字段名。

Schema 也别写过头。比如把摘要长度死卡在 30 字以内,模型会更容易出错,后续修复成本反而变高。


五、Prompt 怎么写:少说空话,多给硬约束

如果你已经有 Schema,Prompt 不要再写成大段自然语言说明。实测里,结构化任务最稳的方式是:角色固定 + 任务边界清晰 + 输出要求明确 + 错误容忍空间小

一个比较稳的模板如下:

SYSTEM_PROMPT = """
你是工单结构化抽取服务。
请基于用户输入,输出一个 JSON 对象。
要求:
1. 只能输出 JSON,不要输出任何解释、前后缀、Markdown 标记。
2. 必须包含字段:ticket_type, priority, sentiment, summary, confidence, needs_human_review。
3. 枚举值必须严格从候选集合中选择。
4. confidence 取值范围为 0 到 1。
5. summary 使用简体中文,长度不超过 120 个字符。
6. 如果信息不足,也必须返回合法 JSON,并给出最合理判断。
"""

用户输入拼进去:

def build_user_prompt(text: str) -> str:
    return f"用户投诉内容如下:\n{text}"

如果模型 API 支持 response_formatjson_schema 或 function calling,优先用原生约束。原因很简单,服务端约束通常比纯 Prompt 更稳。

不过别迷信。部分模型即使支持结构化模式,依然会出现枚举偏移、空字段、截断输出,尤其是长上下文或并发上来之后。


六、统一校验器:把错误分类,而不是只返回失败

很多服务写成这样:

  • 调模型
  • json.loads
  • 报错就重试

这太粗了。工程上更有用的做法是把失败原因标准化,不然后续很难做定向修复。

下面给一个 Python 校验器示例,用 jsonschema

import json
from dataclasses import dataclass
from typing import Any, Optional
from jsonschema import validate, ValidationError

@dataclass
class ValidateResult:
    ok: bool
    data: Optional[dict] = None
    error_type: Optional[str] = None
    error_msg: Optional[str] = None


def validate_llm_output(raw_text: str, schema: dict) -> ValidateResult:
    try:
        data = json.loads(raw_text)
    except json.JSONDecodeError as e:
        return ValidateResult(
            ok=False,
            error_type="json_parse_error",
            error_msg=str(e)
        )

    try:
        validate(instance=data, schema=schema)
    except ValidationError as e:
        return ValidateResult(
            ok=False,
            error_type="schema_validation_error",
            error_msg=e.message
        )

    business_error = check_business_rules(data)
    if business_error:
        return ValidateResult(
            ok=False,
            error_type="business_rule_error",
            error_msg=business_error
        )

    return ValidateResult(ok=True, data=data)


def check_business_rules(data: dict) -> Optional[str]:
    if len(data.get("summary", "")) > 120:
        return "summary too long"
    if data.get("ticket_type") == "refund" and data.get("confidence", 0) < 0.3:
        return "refund with too low confidence"
    return None

这一步的价值在于:后面每一种错误都能对应不同的修复策略。


七、失败修复策略:不是无脑重试,而是按类型处理

我在项目里通常把修复分成三档。

1)轻量修复

适用于明显的格式错误,比如:

  • 输出被 ```json 包裹
  • 前面多了一句“下面是结果”
  • 末尾多了无关解释

这类问题不一定要重调模型,先做一次安全清洗即可。

import re


def sanitize_raw_output(text: str) -> str:
    text = text.strip()
    text = re.sub(r"^```json", "", text, flags=re.IGNORECASE).strip()
    text = re.sub(r"^```", "", text).strip()
    text = re.sub(r"```$", "", text).strip()

    start = text.find("{")
    end = text.rfind("}")
    if start != -1 and end != -1 and end > start:
        text = text[start:end+1]

    return text

要注意,清洗逻辑别写得太激进。否则可能把原始内容截坏,排障时也看不到真实输出。

2)定向修复重试

适用于 JSON 能解析,但 Schema 不通过。

这时不要把原始任务完整重跑,而是把失败原因反馈给模型,让它只修结构,不改语义。

REPAIR_PROMPT = """
你上一次输出的 JSON 未通过校验。
请根据错误信息修复,并只输出修复后的 JSON。
不要添加解释。

校验错误:{error_msg}
原始输出:
{raw_output}
"""

修复流程示例:

def repair_with_feedback(llm, raw_output: str, error_msg: str) -> str:
    prompt = REPAIR_PROMPT.format(error_msg=error_msg, raw_output=raw_output)
    return llm.generate(prompt)

这种方式对字段缺失、枚举值不合法、布尔值写成字符串这类问题比较有效。

3)降级重试

如果修复两次还不行,就不要继续死磕同一个大模型配置了。可以降级:

  • 降低 temperature
  • 缩短上下文,只保留核心输入
  • 切到更稳的模型版本
  • 改为更保守的输出模板

我一般会把结构化任务的 temperature 固定在 00.1。生成创意文本和抽取 JSON,本来就不是一类任务。


八、一套可复现的调用封装

下面给一个完整点的服务封装,把清洗、校验、修复、重试串起来:

from typing import Callable

class StructuredOutputService:
    def __init__(self, llm_generate: Callable[[str], str], schema: dict, max_retry: int = 2):
        self.llm_generate = llm_generate
        self.schema = schema
        self.max_retry = max_retry

    def run(self, user_text: str) -> dict:
        prompt = self._build_prompt(user_text)
        raw_output = self.llm_generate(prompt)

        result = self._try_validate(raw_output)
        if result.ok:
            return self._success_response(result.data, stage="first_pass")

        current_output = raw_output
        for retry_idx in range(self.max_retry):
            repaired = self._repair(current_output, result.error_msg)
            result = self._try_validate(repaired)
            if result.ok:
                return self._success_response(result.data, stage=f"repair_{retry_idx+1}")
            current_output = repaired

        fallback = self._fallback(user_text, current_output, result.error_msg)
        return fallback

    def _build_prompt(self, user_text: str) -> str:
        return SYSTEM_PROMPT + "\n" + build_user_prompt(user_text)

    def _try_validate(self, raw_output: str):
        sanitized = sanitize_raw_output(raw_output)
        return validate_llm_output(sanitized, self.schema)

    def _repair(self, raw_output: str, error_msg: str) -> str:
        return repair_with_feedback(self, raw_output, error_msg)

    def generate(self, prompt: str) -> str:
        return self.llm_generate(prompt)

    def _success_response(self, data: dict, stage: str) -> dict:
        return {
            "status": "ok",
            "stage": stage,
            "data": data
        }

    def _fallback(self, user_text: str, raw_output: str, error_msg: str) -> dict:
        return {
            "status": "fallback",
            "stage": "manual_review",
            "data": {
                "ticket_type": "other",
                "priority": "medium",
                "sentiment": "neutral",
                "summary": user_text[:80],
                "confidence": 0.0,
                "needs_human_review": True
            },
            "debug": {
                "last_error": error_msg,
                "last_output": raw_output[:500]
            }
        }

这里有个小点要提一下:fallback 里的结构也必须满足同一份 Schema。这样下游接口就不用区分“正常结果”和“兜底结果”的数据格式。


九、线上异常兜底:别让解析失败直接变 500

线上最怕的不是有坏样本,而是坏样本把整个请求打挂。我的做法通常是:

1)返回保守结构

即使模型输出不可用,也返回一份合法结构,附带 needs_human_review=true

2)关键场景进人工队列

比如退款、高风险审核、合同抽取,这类任务不能因为模型字段不齐就自动放行。

3)保留原始输出与错误原因

线上排查时,如果只留一个“解析失败”,基本没法定位问题。至少要记录:

  • request_id
  • model_name
  • prompt_version
  • raw_output
  • sanitized_output
  • error_type
  • error_msg
  • retry_count
  • latency_ms

日志字段要统一。后面接监控和看板会省很多事。


十、监控指标怎么设:别只看成功率

结构化输出治理做上线后,我一般会把指标分成四组。

成功类指标

  • raw_json_success_rate
  • schema_pass_rate
  • business_rule_pass_rate
  • final_success_rate

成本类指标

  • avg_retry_count
  • avg_total_tokens
  • avg_latency_ms
  • p95_latency_ms

失败类指标

  • json_parse_error_ratio
  • schema_validation_error_ratio
  • business_rule_error_ratio
  • fallback_ratio

漂移类指标

  • 不同 prompt 版本通过率变化
  • 不同模型版本通过率变化
  • 不同输入长度区间通过率变化
  • 不同业务类型通过率变化

这一块我踩过坑。说实话,有一次我们只看最终成功率,结果看板很平稳,后来拆开才发现 repair_rate 已经从 8% 涨到 26%,延迟多了接近 700ms,成本也高了一截。


十一、实测对比:只靠 Prompt,和分层治理差多少

下面给一组我按类似项目方式整理的对比数据。测试集 2000 条,任务是工单结构化抽取,输入含口语、省略句、错别字和中英混杂文本。模型参数固定,temperature=0。

方案 Raw JSON 成功率 Schema 通过率 最终可用率 平均延迟 平均输出 token
仅 Prompt 约束 91.8% 84.7% 84.7% 1.21s 168
Prompt + 清洗 95.6% 88.9% 88.9% 1.24s 168
Prompt + 清洗 + Schema 校验 + 1次修复 95.6% 88.9% 94.8% 1.63s 224
Prompt + 清洗 + Schema 校验 + 2次修复 + 兜底 95.6% 88.9% 99.2% 1.88s 251

这个结果说明两件事:

  • 只看初次输出,结构化通过率往往没想象中高
  • 修复和兜底能显著提高最终可用率,但会带来延迟和 token 成本

所以工程上不能只追求“最终数字最好看”,还要看你的业务能不能接受额外 600ms 到 800ms 的耗时。


十二、常见错误案例与处理方式

案例 1:输出带解释文本

原始输出:

以下是分析结果:
{
  "ticket_type": "refund",
  "priority": "high",
  "sentiment": "negative",
  "summary": "用户反馈重复扣款,申请退款",
  "confidence": 0.93,
  "needs_human_review": false
}

处理:轻量清洗,一般可恢复。

案例 2:枚举值漂移

原始输出:

{
  "ticket_type": "refund_request",
  "priority": "urgent",
  "sentiment": "negative",
  "summary": "用户要求退款",
  "confidence": 0.85,
  "needs_human_review": false
}

处理:Schema 校验失败,回传错误信息做定向修复。

案例 3:字段缺失

原始输出:

{
  "ticket_type": "invoice",
  "summary": "用户要求补开发票",
  "confidence": 0.76
}

处理:重试修复,要求补全缺失字段。如果连续失败,走兜底。

案例 4:布尔值和数值类型错误

原始输出:

{
  "ticket_type": "delivery",
  "priority": "medium",
  "sentiment": "neutral",
  "summary": "用户咨询物流进展",
  "confidence": "0.66",
  "needs_human_review": "false"
}

处理:看场景。如果你要求严格,直接判失败并修复;如果业务允许,也可以加一层安全类型转换,但要写审计日志。


十三、生产环境里的几个细节建议

1)Prompt 要版本化

结构化输出很吃模板稳定性。建议把 prompt_version 打进日志和结果表,不然线上回溯很痛苦。

2)Schema 变更要兼容旧数据

如果新增字段,先让下游兼容,再切模型输出。别直接硬切。

3)分任务设不同容忍度

信息抽取、标签分类、工具参数生成,容错标准不一样。工具调用参数错一个字段,后果可能比标签分类严重得多。

4)把错误样本沉淀成回归集

每周把解析失败、修复失败、人工驳回样本收进测试集。这个习惯很值钱。

5)高风险任务别省人工兜底

这条很现实。模型再稳,也会遇到分布外输入。完全自动化有时不是最合适的选择。

唯一的局限是,这套方案会增加一些服务复杂度,特别是日志、重试、回归集维护这些部分,需要工程上多花点时间。但比起线上随机报错,我觉得这笔成本是值得付的。


十四、一个更稳的落地顺序

如果你现在手里已经有一个“能跑但经常解析失败”的 LLM 服务,我建议按这个顺序改:

  1. 先补 Schema 和统一校验器
  2. 接入原始输出清洗
  3. 增加基于错误信息的定向修复
  4. 上线 fallback 和人工审核标记
  5. 接监控面板,拆分各阶段通过率
  6. 把失败样本沉淀成回归测试集

顺序别反。

很多团队一开始就去换模型、换 Prompt、调采样参数,最后发现问题真正卡在“没有可观测性”和“没有标准兜底”。


十五、结语

大模型接入业务后,结构化输出稳定性是一个很典型的工程问题。它不神秘,也不是靠一条 Prompt 就能彻底解决。更实用的做法是把流程拆清楚:前面约束输出,中间严格校验,失败后按类型修复,最后给线上留兜底。

如果你正在做信息抽取、工单分类、Agent 工具调用、审核结果落库这类场景,建议优先看一眼自己的 schema_pass_ratefallback_ratio。很多问题其实早就在线上出现了,只是还没被单独统计出来。

我自己的判断是,结构化输出治理做得越早,后面的评估、成本控制和线上排障都会轻松很多。先把“可消费率”做稳,再去谈更高层的业务指标,会更踏实一些。

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐