在实际的线上大模型项目里JSON 输出格式飘忽不定往往比回答内容质量差更早暴露问题。同一个提示词上一轮还返回合法 JSON下一轮就可能带上 Markdown 代码块、多余解释、尾随逗号甚至直接在嵌套对象的中间截断。只要下游代码用json.loads一解析整个流程就会被打断。只靠反复微调 Prompt很难把这类问题彻底解决因为模型生成是概率采样只要输出没有在解码阶段被约束格式错误就一定会出现。本文从提示词约束、输出校验、后处理修复、重试降级四个层级给出一个可落地的完整稳定化方案。1. 先定位JSON 输出为什么会在线上反复出错1.1 模型输出是采样结果不是模板渲染结果大模型生成文本时每次选择一个 token 都按概率分布采样而不是直接渲染一段固定的模板。temperature越高采样随机性越大同样输入产生不同 token 序列的概率就越高。JSON 是一种严格的结构化语法要求模型在长序列里始终保持括号配对、引号闭合、逗号正确这对生成式模型来说本质上是在概率空间里逼近一个离散结构。只要允许模型自由解码就存在格式失败的概率。这个概率很小的时候单条测试看不出来但线上一天调用几十万次失败样本就会持续累积。所以问题不是“为什么会有错误输出”而是“系统必须把每一次错误输出都当作预期事件来处理”。1.2 线上最常见的六类 JSON 输出故障根据实际项目里常见失败样本可以归纳为六类输出被 Markdown 代码块包裹形如json ... 。JSON 前后带有解释性文字比如“以下是结果{...}”。语法细节错误尾随逗号、缺少逗号、单引号字符串、未转义换行。输出被截断因为max_tokens不够或触发了长度上限JSON 结构不完整。字段缺失、字段名偏离、字段类型不对例如confidence输出成字符串0.85。模型把整个 JSON 当成字符串又序列化了一次形成嵌套转义。这里的问题可以先用一张表归类故障现象典型原因影响范围输出带代码块模型默认按 Markdown 返回直接json.loads失败前后有说明文字提示词未约束“只输出 JSON”解析失败或解析到错误片段逗号、引号、括号异常采样噪声、截断语法解析失败结构完整但字段缺失模型按自己的理解省略字段解析成功但业务逻辑异常字段类型不符合预期模型对类型理解偏差下游类型转换崩溃整体被多次序列化模型把 JSON 当字符串解析后类型不是对象1.3 Prompt 微调解决不了全部问题需要分层兜底Prompt 微调可以显著降低格式失败率但它不是保证。模型没有所谓“严格遵守提示词”的确定性能力提示词越长模型注意力越可能分散同一模型升级版本后输出风格也可能变化。更关键的是就算模型输出了合法 JSON它也不一定符合业务字段要求。因此工程上要做的是分层兜底用提示词和模型能力提高首次成功率用校验确认输出是否真的可用用后处理把可修复问题救回来用重试吸收随机波动用降级保证系统不崩溃。这才是“跳出 Prompt 微调思路”的真正含义。2. 四层方案的整体设计2.1 每一层负责一个明确职责在动手写代码前先把四层职责定义清楚避免层与层之间职责重叠。层级核心动作解决什么问题失败后怎么办提示词约束输出协议、结构化输出参数、采样参数提高首次格式正确率交给校验层输出校验语法解析、Schema 校验、业务规则校验确认输出是否真的可用交给修复或重试后处理修复提取 JSON 片段、修复括号、容错解析把可修复问题自动救回修复后必须重新校验重试降级错误分类、退避重试、降级熔断吸收随机波动保障可用性返回降级结果并告警这四层不能互相替代。只做校验不做修复失败率会偏高只做修复不做校验可能出现“看似成功、实际字段不对”的假正常只做重试不做分类则会把不可重试的错误反复重试浪费成本和响应时间。2.2 一次完整调用如何穿越四层一次请求应该按照固定顺序执行组装提示词同时设置模型支持的结构化输出参数。调用模型拿到原始文本。后处理模块从原始文本中提取 JSON 片段并做保守修复。语法解析确认这是一段合法 JSON。Schema 校验确认字段和类型符合业务要求。校验成功后把原始文本和解析结果一并返回调用方。校验失败时记录原始输出和错误类型。如果错误可重试退避后重新发起调用。达到最大重试次数后走降级策略。这个链路可以用一个统一入口封装所有线上代码都通过这个入口调用模型不直接散落openai.ChatCompletion之类的调用。3. 第一层提示词约束与模型能力配合使用3.1 提示词里把“输出协议”写清楚提示词里写 JSON 格式不是简单写一句“返回 JSON”而是要把输出协议讲完整。至少包括字段列表、每个字段的类型、字段含义、必须遵守的约束、禁止做什么。一个示例提示词模板SYSTEM_PROMPT 你是一个数据提取助手。只输出 JSON 对象不要输出任何解释、Markdown 标记或结束语。 JSON_PROMPT_TEMPLATE 请根据用户问题结合给定上下文输出一个 JSON 对象。 要求 1. 只能输出一个合法 JSON 对象不要输出 Markdown 代码块标记。 2. 输出必须包含以下字段 - answer: string对用户问题的简要回答 - confidence: number回答置信度取值范围 0 到 1 - sources: array of string信息来源列表可以为空数组 3. 如果没有足够信息回答问题answer 填写“无法回答”confidence 填写 0。 4. 不要输出字段之外的任何键。 用户问题{question} 上下文 {context} def build_messages(question: str, context: str) - list[dict]: user_content JSON_PROMPT_TEMPLATE.format(questionquestion, contextcontext) return [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_content}, ]这里的关键是“不要输出字段之外的任何键”和“没有足够信息时也按固定结构返回”。前者避免模型自行增加字段后者保证极端情况下 JSON 结构仍然完整。3.2 优先开启模型自带的结构化输出能力不要只靠文字约束很多模型和推理框架已经提供了结构化输出能力优先使用这些能力比在提示词里反复强调格式更可靠。常见方式包括OpenAI 兼容接口的response_format{type: json_object}或更严格的 JSON Schema 约束。Ollama 的format: json参数。vLLM、SGLang、llama.cpp 等推理框架的guided_json、grammar、结构化解码能力。对话模型通过 function calling / tool calling 返回结构化参数。要注意的是json_object模式通常只保证输出是合法 JSON不保证字段符合你的业务 Schema。所以即便开启了结构化输出校验层仍然不能省。3.3 参数选择temperature、max_tokens、stop除了提示词采样参数对格式稳定性影响很大。参数推荐值作用调大后的影响调小后的风险temperature0 到 0.3控制随机性更容易出现格式噪声回答多样性下降但对于结构化提取不是风险max_tokens按正常输出长度加 30% 余量控制最大生成长度浪费等待时间和费用输出中途截断JSON 不完整stop按任务设置终止标记提前结束生成可能提前截断答案需要仔细验证不会丢内容对于结构化提取类任务temperature建议直接设为 0 或一个很小的值。max_tokens不能只按业务预估长度设置要预留 JSON 前后可能出现的额外噪声空间。3.4 一份可直接改用的 JSON 提示词模板实际项目里提示词模板通常要支持传入动态字段。可以用一个字典来定义字段再统一渲染成提示词OUTPUT_FIELDS { answer: string对用户问题的简要回答, confidence: number置信度0 到 1, sources: array of string信息来源列表, } def build_schema_description(fields: dict[str, str]) - str: lines [输出必须包含以下字段] for name, desc in fields.items(): lines.append(f - {name}: {desc}) return \n.join(lines) def build_user_prompt(question: str, context: str) - str: schema_desc build_schema_description(OUTPUT_FIELDS) return ( 只输出一个 JSON 对象不要输出解释。\n\n f{schema_desc}\n\n f用户问题{question}\n\n f上下文{context} )这样以后字段变更只需要改OUTPUT_FIELDS提示词模板自动同步避免“改需求只改代码不改提示词”的低级问题。4. 第二层用语法校验和 Schema 校验把失败挡在业务逻辑之前4.1 语法校验先确认能否被标准解析器解析模型输出经过后处理之后第一步是要确认它是一段合法 JSON。这一步的关键是不要直接使用默认异常而是要包装成我们自己的可重试错误并保留解析器错误信息。import json class RetryableError(Exception): 可重试的错误重试可能成功。 class NotRetryableError(Exception): 不可重试的错误重试大概率还是失败。 def safe_loads(text: str): try: return json.loads(text) except json.JSONDecodeError as exc: raise RetryableError(fJSON 语法解析失败: {exc}) from exc语法解析只能判断“这段文本是不是 JSON”不能判断“这个 JSON 是否符合业务要求”。很多线上问题恰恰是解析成功但字段缺失直到业务层报KeyError才发现。4.2 Schema 校验字段缺失和类型错误才是真正的业务风险用 JSON Schema 描述业务对输出的要求再用jsonschema库校验。import jsonschema SCHEMA { type: object, required: [answer, confidence, sources], properties: { answer: {type: string}, confidence: {type: number, minimum: 0, maximum: 1}, sources: {type: array, items: {type: string}}, }, additionalProperties: False, } def validate_data(data): try: jsonschema.validate(instancedata, schemaSCHEMA) except jsonschema.ValidationError as exc: raise RetryableError(fSchema 校验失败: {exc}) from excSchema 的几个要点required列表里的字段一个都不能少。additionalProperties: False会拒绝模型自行添加的额外字段保持输出协议严格。minimum、maximum这类约束能挡住置信度 1.5、概率 -0.2 这类明显错误。如果业务允许兼容再考虑放宽不要一开始就把 schema 写得过松。4.3 每次校验失败都必须留下原始输出校验失败时最怕只记录“JSON 格式错误”这样的摘要没有原始文本。没有原始输出后续就无法判断是模型问题、提示词问题还是截断问题。建议每次失败都记录结构化日志def log_failure(raw_text, exc, model, attempt, prompt_hash): failure { model: model, attempt: attempt, raw_text: raw_text, error: str(exc), prompt_hash: prompt_hash, } # 写入日志或消息队列供后续分析 print(json.dumps(failure, ensure_asciiFalse))需要注意raw_text可能包含用户敏感信息写入日志前要按公司规范做脱敏。建议在开发环境全量记录在生产环境只记录脱敏后的摘要和错误码需要排查时再根据请求 ID 查完整原始输出。5. 第三层后处理与容错解析能修复的尽量修复5.1 从非 JSON 噪音中提取 JSON 片段后处理的第一项任务是提取。模型即使被告知只输出 JSON偶尔还是会在前面加“结果如下”或者用代码块包裹。提取策略是先剥离 Markdown 代码块再取第一个{到最后一个}之间的内容。import re def extract_json(raw_text: str) - str: text raw_text.strip() fence re.search(r(?:json)?\s*(.*?)\s*, text, re.DOTALL) if fence: text fence.group(1).strip() start text.find({) end text.rfind(}) if start ! -1 and end ! -1 and end start: text text[start:end 1] return text利用“取最后一个}”的策略可以把尾部解释性文字丢掉。但如果模型在 JSON 后面又跟了一个对象这个策略可能截到错误位置所以提取之后仍必须经过语法和 Schema 校验。5.2 修复截断、括号不平衡和多余逗号后处理的核心是“保守修复”。只处理能明确判断的问题不猜测模型意图。def repair_json(text: str) - str: # 去掉尾随逗号例如 {a: 1,} 或 [1, 2,] text re.sub(r,\s*([}\]]), r\1, text) # 补全截断场景下未闭合的引号和括号 stack [] in_string False escaped False for ch in text: if in_string: if escaped: escaped False elif ch \\: escaped True elif ch : in_string False continue if ch : in_string True elif ch in {[: stack.append(ch) elif ch in }]: if stack: stack.pop() if in_string: text for token in reversed(stack): text } if token { else ] return text这个修复函数处理的是最简单也最常见的两类问题多余逗号和因为截断导致的未闭合括号。它不处理字段缺失、类型错误这类语义问题修复之后仍然必须走语法校验和 Schema 校验严禁“修复完直接返回”。5.3 容错解析适合兜底不能替代前两层有些团队会用json5、demjson3、pydantic等容错解析器直接解析模型输出。这类工具能接受单引号、尾随逗号、注释等非标准 JSON 语法开发时确实省事。但容错解析要谨慎。它接受更宽语法不代表业务数据结构正确它内部做了很多隐性修复会让你更难定位模型输出问题。更推荐的做法是后处理先把输出规整成标准 JSON再用标准解析器解析这样线上返回给其它服务的数据始终是标准 JSON不会因为解析器差异导致跨语言消费问题。6. 第四层重试与降级策略6.1 先给错误分类哪些重试有用哪些重试是浪费重试不是所有错误都能处理。错误分类是重试策略的第一步。错误类型示例是否重试原因格式随机性错误JSON 语法错误、schema 缺字段、截断应该重试重新采样可能输出合法结果服务暂时性故障429 限流、5xx、超时应该重试等待后可能恢复输入或参数错误400 参数错误、context length 超限不重试重试大概率同样失败内容策略拦截提示词触发内容安全规则不重试需要改输入而不是重试业务条件不具备输入信息本身缺少答案字段来源不重试这是输入问题不是输出问题代码里用异常类型区分而不是靠字符串匹配错误信息。def classify_exception(exc: Exception): if isinstance(exc, RetryableError): return True return False6.2 退避重试要给随机抖动重试间隔不能固定否则多个请求同时失败后会同时重试形成雪崩。推荐指数退避加随机抖动。import random def backoff_delay(attempt: int, base: float 1.0, cap: float 8.0, jitter: float 0.3) - float: exp min(base * (2 ** (attempt - 1)), cap) return exp random.uniform(0, exp * jitter)attempt从 1 开始第一次重试间隔约 1 秒第二次约 2 秒第三次约 4 秒最大不超过 8 秒。抖动的作用是防止同一批请求在同一个时间点同时重试。重试次数要有限制不能无限循环。一般结构化提取任务建议 2 到 3 次重试。重试次数太多单次请求耗时会过长用户体验和成本都不可控。6.3 降级和熔断要考虑可用性重试耗尽之后必须有降级路径。常见降级方式切换到备用模型或更保守的模型版本。使用最近一次成功返回的缓存结果。返回一个固定结构的兜底 JSON比如{answer: 暂时无法回答, confidence: 0, sources: []}。直接把错误抛给上层由调用方决定如何处理。如果系统连续失败次数超过阈值要先熔断不再进入模型调用直接走降级等冷却期后再恢复。熔断能避免大模型服务出现故障时整个业务系统被拖垮。7. 完整实现一个可运行的 Python 稳定输出管道7.1 模块划分与目录结构把前面四层落到代码里建议按模块拆分方便测试和替换。llm_json_pipeline/ ├── prompt.py # 提示词模板 ├── client.py # 模型请求封装 ├── validator.py # 语义解析和 Schema 校验 ├── repair.py # 后处理修复 ├── pipeline.py # 四层主流程 └── main.py # 示例入口模块划分的原则提示词、模型调用、校验、修复互不依赖哪个模块想换都能单独替换。7.2 请求封装与提示词组装client.py用一个统一的LLMClient封装模型调用调用方传入 prompts返回原始文本。from openai import OpenAI class LLMClient: def __init__(self, api_key: str, base_url: str, model: str): self.client OpenAI(api_keyapi_key, base_urlbase_url) self.model model def chat( self, messages: list[dict], temperature: float 0.1, max_tokens: int 1024, response_format: dict | None None, ) - str: kwargs { model: self.model, messages: messages, temperature: temperature, max_tokens: max_tokens, } if response_format: kwargs[response_format] response_format resp self.client.chat.completions.create(**kwargs) return resp.choices[0].message.content注意openaiSDK 的版本不同参数名可能有差异。本地部署的模型如果暴露了 OpenAI 兼容接口也可以复用这个封装。7.3 校验、修复、重试主流程pipeline.py把四层串起来import logging import time logger logging.getLogger(__name__) class LLMJsonPipeline: def __init__( self, client: LLMClient, schema: dict, primary_model: str, fallback_model: str | None None, max_attempts: int 3, base_delay: float 1.0, ): self.client client self.schema schema self.primary_model primary_model self.fallback_model fallback_model self.max_attempts max_attempts self.base_delay base_delay def run(self, question: str, context: str): messages build_messages(question, context) last_error None for attempt in range(1, self.max_attempts 1): try: raw_text self.client.chat( messagesmessages, temperature0.1, max_tokens1024, response_format{type: json_object}, ) text extract_json(raw_text) text repair_json(text) data safe_loads(text) validate_data(data, self.schema) logger.info(attempt %s succeeded, attempt) return data, raw_text except NotRetryableError as exc: raise exc except Exception as exc: last_error exc logger.warning(attempt %s failed: %s, attempt, exc) if attempt self.max_attempts: time.sleep(backoff_delay(attempt, baseself.base_delay)) return self._fallback(question, context, last_error) def _fallback(self, question, context, last_error): if self.fallback_model: try: raw_text self.client.chat( messagesbuild_messages(question, context), temperature0.0, max_tokens1024, ) text extract_json(raw_text) text repair_json(text) data safe_loads(text) validate_data(data, self.schema) return data, raw_text except Exception as exc: logger.error(fallback model failed: %s, exc) degraded { answer: 暂时无法回答, confidence: 0, sources: [], } logger.error(all attempts failed: %s, last_error) return degraded, 主流程里每个失败分支都会记录日志成功分支返回(data, raw_text)调用方如果需要排查可以拿到原始输出。7.4 运行效果与日志观察在main.py里跑一个示例if __name__ __main__: logging.basicConfig(levellogging.INFO) client LLMClient( api_keyyour-api-key, base_urlhttp://localhost:8000/v1, modelprimary-model, ) pipeline LLMJsonPipeline( clientclient, schemaSCHEMA, primary_modelprimary-model, fallback_modelNone, max_attempts3, base_delay1.0, ) data, raw pipeline.run( question今天上海适合户外跑步吗, context上海今日多云气温 22 到 28 度空气质量良。, ) print(data)正常情况下会看到INFO attempt 1 succeeded {answer: 适合户外跑步, confidence: 0.85, sources: []}如果第一次输出格式有问题日志会是WARNING attempt 1 failed: JSON 语法解析失败: Expecting , delimiter... WARNING attempt 2 failed: JSON 语法解析失败: ... INFO attempt 3 succeeded通过日志可以直观看到重试是否生效以及是否出现了大面积失败。8. 线上排错从现象倒推根因8.1 按这条链路逐层排查线上出现 JSON 输出异常时不要直接改提示词按下面的顺序排查先确认原始输出有没有被完整记录。没有原始输出后面所有分析都是猜测。再确认是不是截断。看是否每次都在同一个位置截断如果是检查max_tokens是否不足。然后确认提示词是否真的按预期传给模型。检查有没有别的前置逻辑改写或拼接了提示词。接着确认结构化输出参数是否真的生效。有些模型版本对response_format的支持不完全参数传了也可能被忽略。然后检查重试是否在同一个模型、同一组参数下重复避免“重试只是浪费时间”。最后确认降级路径是否正常。降级结果有没有被当作正常结果返回给调用方。8.2 常见现象、原因与处理对照表问题现象可能原因检查方式处理建议返回合法 JSON 但缺字段提示词和 schema 不一致对比提示词模板与 Schema 的 required 字段统一用字段定义生成提示词和 Schema每次都在同一位置截断max_tokens不足看 raw_text 是否以半截结构结尾增大max_tokens或压缩上下文长度输出带代码块模型默认 Markdown 风格看 raw_text 开头是否是 后处理提取代码块同时提示词强调不要代码块偶尔多一个逗号采样随机噪声看错误日志中的 JSONDecodeError后处理修复尾随逗号再重试重试 3 次全部失败模型降级、输入不适合该模型统计连续失败率检查输入长度、内容策略必要时换模型结构化输出模式仍出现非法 JSON模型版本或参数名不支持直接打印请求参数确认当前模型是否真的支持该参数成功但业务结果不对Schema 太松检查 schema 的 type、required收紧密集字段约束增加业务规则校验排查时建议给每次请求分配一个 request_id从入口到重试、降级都带上这样日志才能串联起来。9. 生产落地建议与扩展方向9.1 学习环境和生产环境要区别对待本地实验阶段可以直接用json.loads解析重点观察模型输出长什么样。开发环境可以详细记录原始输出方便分析失败样本。测试环境要加入失败注入模拟模型输出带代码块、截断、缺字段等情况验证四层逻辑是否都生效。生产环境还要额外考虑原始输出日志脱敏避免敏感信息落库。记录成功率、各类失败率、平均耗时、重试次数等指标。对降级结果打上明确标记避免业务层把兜底 JSON 当成真实回答。控制重试成本和总耗时为单次请求设置整体超时。大模型服务压力大时优先熔断而不是无限重试。9.2 上线前检查清单检查项确认标准四层是否都实现提示词、校验、修复、重试降级缺一不可Schema 与业务字段一致required、type、取值范围都明确原始输出有记录每个失败样本都能找到原始文本和请求参数重试配置可调重试次数、退避基数、抖动参数通过配置控制降级路径无副作用兜底 JSON 不会进入下游写入流程指标齐全能统计成功率、失败原因分布、重试成功率灰度计划先让一部分请求走新逻辑观察指标再全量9.3 进阶方向如果四层方案已经稳定下一步可以往更底层去优化。对自建推理服务可以引入受控解码或语法约束生成让模型在解码阶段就不可能生成非法 JSON。对闭源模型可以优先使用 function calling / tool calling 替代裸 JSON 输出把结构化结果交给模型自身机制管理。对提示词和 Schema可以做版本管理每次调整都能对比格式成功率变化。对复杂业务建议用pydantic定义输出模型再自动生成 JSON Schema保证类型定义和校验逻辑不脱节。只要保证每个失败场景都有兜底路径JSON 输出从概率问题就会变成可观测、可控制、可恢复的工程问题。
大模型JSON输出稳定化：从提示词到重试降级的四层方案
在实际的线上大模型项目里JSON 输出格式飘忽不定往往比回答内容质量差更早暴露问题。同一个提示词上一轮还返回合法 JSON下一轮就可能带上 Markdown 代码块、多余解释、尾随逗号甚至直接在嵌套对象的中间截断。只要下游代码用json.loads一解析整个流程就会被打断。只靠反复微调 Prompt很难把这类问题彻底解决因为模型生成是概率采样只要输出没有在解码阶段被约束格式错误就一定会出现。本文从提示词约束、输出校验、后处理修复、重试降级四个层级给出一个可落地的完整稳定化方案。1. 先定位JSON 输出为什么会在线上反复出错1.1 模型输出是采样结果不是模板渲染结果大模型生成文本时每次选择一个 token 都按概率分布采样而不是直接渲染一段固定的模板。temperature越高采样随机性越大同样输入产生不同 token 序列的概率就越高。JSON 是一种严格的结构化语法要求模型在长序列里始终保持括号配对、引号闭合、逗号正确这对生成式模型来说本质上是在概率空间里逼近一个离散结构。只要允许模型自由解码就存在格式失败的概率。这个概率很小的时候单条测试看不出来但线上一天调用几十万次失败样本就会持续累积。所以问题不是“为什么会有错误输出”而是“系统必须把每一次错误输出都当作预期事件来处理”。1.2 线上最常见的六类 JSON 输出故障根据实际项目里常见失败样本可以归纳为六类输出被 Markdown 代码块包裹形如json ... 。JSON 前后带有解释性文字比如“以下是结果{...}”。语法细节错误尾随逗号、缺少逗号、单引号字符串、未转义换行。输出被截断因为max_tokens不够或触发了长度上限JSON 结构不完整。字段缺失、字段名偏离、字段类型不对例如confidence输出成字符串0.85。模型把整个 JSON 当成字符串又序列化了一次形成嵌套转义。这里的问题可以先用一张表归类故障现象典型原因影响范围输出带代码块模型默认按 Markdown 返回直接json.loads失败前后有说明文字提示词未约束“只输出 JSON”解析失败或解析到错误片段逗号、引号、括号异常采样噪声、截断语法解析失败结构完整但字段缺失模型按自己的理解省略字段解析成功但业务逻辑异常字段类型不符合预期模型对类型理解偏差下游类型转换崩溃整体被多次序列化模型把 JSON 当字符串解析后类型不是对象1.3 Prompt 微调解决不了全部问题需要分层兜底Prompt 微调可以显著降低格式失败率但它不是保证。模型没有所谓“严格遵守提示词”的确定性能力提示词越长模型注意力越可能分散同一模型升级版本后输出风格也可能变化。更关键的是就算模型输出了合法 JSON它也不一定符合业务字段要求。因此工程上要做的是分层兜底用提示词和模型能力提高首次成功率用校验确认输出是否真的可用用后处理把可修复问题救回来用重试吸收随机波动用降级保证系统不崩溃。这才是“跳出 Prompt 微调思路”的真正含义。2. 四层方案的整体设计2.1 每一层负责一个明确职责在动手写代码前先把四层职责定义清楚避免层与层之间职责重叠。层级核心动作解决什么问题失败后怎么办提示词约束输出协议、结构化输出参数、采样参数提高首次格式正确率交给校验层输出校验语法解析、Schema 校验、业务规则校验确认输出是否真的可用交给修复或重试后处理修复提取 JSON 片段、修复括号、容错解析把可修复问题自动救回修复后必须重新校验重试降级错误分类、退避重试、降级熔断吸收随机波动保障可用性返回降级结果并告警这四层不能互相替代。只做校验不做修复失败率会偏高只做修复不做校验可能出现“看似成功、实际字段不对”的假正常只做重试不做分类则会把不可重试的错误反复重试浪费成本和响应时间。2.2 一次完整调用如何穿越四层一次请求应该按照固定顺序执行组装提示词同时设置模型支持的结构化输出参数。调用模型拿到原始文本。后处理模块从原始文本中提取 JSON 片段并做保守修复。语法解析确认这是一段合法 JSON。Schema 校验确认字段和类型符合业务要求。校验成功后把原始文本和解析结果一并返回调用方。校验失败时记录原始输出和错误类型。如果错误可重试退避后重新发起调用。达到最大重试次数后走降级策略。这个链路可以用一个统一入口封装所有线上代码都通过这个入口调用模型不直接散落openai.ChatCompletion之类的调用。3. 第一层提示词约束与模型能力配合使用3.1 提示词里把“输出协议”写清楚提示词里写 JSON 格式不是简单写一句“返回 JSON”而是要把输出协议讲完整。至少包括字段列表、每个字段的类型、字段含义、必须遵守的约束、禁止做什么。一个示例提示词模板SYSTEM_PROMPT 你是一个数据提取助手。只输出 JSON 对象不要输出任何解释、Markdown 标记或结束语。 JSON_PROMPT_TEMPLATE 请根据用户问题结合给定上下文输出一个 JSON 对象。 要求 1. 只能输出一个合法 JSON 对象不要输出 Markdown 代码块标记。 2. 输出必须包含以下字段 - answer: string对用户问题的简要回答 - confidence: number回答置信度取值范围 0 到 1 - sources: array of string信息来源列表可以为空数组 3. 如果没有足够信息回答问题answer 填写“无法回答”confidence 填写 0。 4. 不要输出字段之外的任何键。 用户问题{question} 上下文 {context} def build_messages(question: str, context: str) - list[dict]: user_content JSON_PROMPT_TEMPLATE.format(questionquestion, contextcontext) return [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_content}, ]这里的关键是“不要输出字段之外的任何键”和“没有足够信息时也按固定结构返回”。前者避免模型自行增加字段后者保证极端情况下 JSON 结构仍然完整。3.2 优先开启模型自带的结构化输出能力不要只靠文字约束很多模型和推理框架已经提供了结构化输出能力优先使用这些能力比在提示词里反复强调格式更可靠。常见方式包括OpenAI 兼容接口的response_format{type: json_object}或更严格的 JSON Schema 约束。Ollama 的format: json参数。vLLM、SGLang、llama.cpp 等推理框架的guided_json、grammar、结构化解码能力。对话模型通过 function calling / tool calling 返回结构化参数。要注意的是json_object模式通常只保证输出是合法 JSON不保证字段符合你的业务 Schema。所以即便开启了结构化输出校验层仍然不能省。3.3 参数选择temperature、max_tokens、stop除了提示词采样参数对格式稳定性影响很大。参数推荐值作用调大后的影响调小后的风险temperature0 到 0.3控制随机性更容易出现格式噪声回答多样性下降但对于结构化提取不是风险max_tokens按正常输出长度加 30% 余量控制最大生成长度浪费等待时间和费用输出中途截断JSON 不完整stop按任务设置终止标记提前结束生成可能提前截断答案需要仔细验证不会丢内容对于结构化提取类任务temperature建议直接设为 0 或一个很小的值。max_tokens不能只按业务预估长度设置要预留 JSON 前后可能出现的额外噪声空间。3.4 一份可直接改用的 JSON 提示词模板实际项目里提示词模板通常要支持传入动态字段。可以用一个字典来定义字段再统一渲染成提示词OUTPUT_FIELDS { answer: string对用户问题的简要回答, confidence: number置信度0 到 1, sources: array of string信息来源列表, } def build_schema_description(fields: dict[str, str]) - str: lines [输出必须包含以下字段] for name, desc in fields.items(): lines.append(f - {name}: {desc}) return \n.join(lines) def build_user_prompt(question: str, context: str) - str: schema_desc build_schema_description(OUTPUT_FIELDS) return ( 只输出一个 JSON 对象不要输出解释。\n\n f{schema_desc}\n\n f用户问题{question}\n\n f上下文{context} )这样以后字段变更只需要改OUTPUT_FIELDS提示词模板自动同步避免“改需求只改代码不改提示词”的低级问题。4. 第二层用语法校验和 Schema 校验把失败挡在业务逻辑之前4.1 语法校验先确认能否被标准解析器解析模型输出经过后处理之后第一步是要确认它是一段合法 JSON。这一步的关键是不要直接使用默认异常而是要包装成我们自己的可重试错误并保留解析器错误信息。import json class RetryableError(Exception): 可重试的错误重试可能成功。 class NotRetryableError(Exception): 不可重试的错误重试大概率还是失败。 def safe_loads(text: str): try: return json.loads(text) except json.JSONDecodeError as exc: raise RetryableError(fJSON 语法解析失败: {exc}) from exc语法解析只能判断“这段文本是不是 JSON”不能判断“这个 JSON 是否符合业务要求”。很多线上问题恰恰是解析成功但字段缺失直到业务层报KeyError才发现。4.2 Schema 校验字段缺失和类型错误才是真正的业务风险用 JSON Schema 描述业务对输出的要求再用jsonschema库校验。import jsonschema SCHEMA { type: object, required: [answer, confidence, sources], properties: { answer: {type: string}, confidence: {type: number, minimum: 0, maximum: 1}, sources: {type: array, items: {type: string}}, }, additionalProperties: False, } def validate_data(data): try: jsonschema.validate(instancedata, schemaSCHEMA) except jsonschema.ValidationError as exc: raise RetryableError(fSchema 校验失败: {exc}) from excSchema 的几个要点required列表里的字段一个都不能少。additionalProperties: False会拒绝模型自行添加的额外字段保持输出协议严格。minimum、maximum这类约束能挡住置信度 1.5、概率 -0.2 这类明显错误。如果业务允许兼容再考虑放宽不要一开始就把 schema 写得过松。4.3 每次校验失败都必须留下原始输出校验失败时最怕只记录“JSON 格式错误”这样的摘要没有原始文本。没有原始输出后续就无法判断是模型问题、提示词问题还是截断问题。建议每次失败都记录结构化日志def log_failure(raw_text, exc, model, attempt, prompt_hash): failure { model: model, attempt: attempt, raw_text: raw_text, error: str(exc), prompt_hash: prompt_hash, } # 写入日志或消息队列供后续分析 print(json.dumps(failure, ensure_asciiFalse))需要注意raw_text可能包含用户敏感信息写入日志前要按公司规范做脱敏。建议在开发环境全量记录在生产环境只记录脱敏后的摘要和错误码需要排查时再根据请求 ID 查完整原始输出。5. 第三层后处理与容错解析能修复的尽量修复5.1 从非 JSON 噪音中提取 JSON 片段后处理的第一项任务是提取。模型即使被告知只输出 JSON偶尔还是会在前面加“结果如下”或者用代码块包裹。提取策略是先剥离 Markdown 代码块再取第一个{到最后一个}之间的内容。import re def extract_json(raw_text: str) - str: text raw_text.strip() fence re.search(r(?:json)?\s*(.*?)\s*, text, re.DOTALL) if fence: text fence.group(1).strip() start text.find({) end text.rfind(}) if start ! -1 and end ! -1 and end start: text text[start:end 1] return text利用“取最后一个}”的策略可以把尾部解释性文字丢掉。但如果模型在 JSON 后面又跟了一个对象这个策略可能截到错误位置所以提取之后仍必须经过语法和 Schema 校验。5.2 修复截断、括号不平衡和多余逗号后处理的核心是“保守修复”。只处理能明确判断的问题不猜测模型意图。def repair_json(text: str) - str: # 去掉尾随逗号例如 {a: 1,} 或 [1, 2,] text re.sub(r,\s*([}\]]), r\1, text) # 补全截断场景下未闭合的引号和括号 stack [] in_string False escaped False for ch in text: if in_string: if escaped: escaped False elif ch \\: escaped True elif ch : in_string False continue if ch : in_string True elif ch in {[: stack.append(ch) elif ch in }]: if stack: stack.pop() if in_string: text for token in reversed(stack): text } if token { else ] return text这个修复函数处理的是最简单也最常见的两类问题多余逗号和因为截断导致的未闭合括号。它不处理字段缺失、类型错误这类语义问题修复之后仍然必须走语法校验和 Schema 校验严禁“修复完直接返回”。5.3 容错解析适合兜底不能替代前两层有些团队会用json5、demjson3、pydantic等容错解析器直接解析模型输出。这类工具能接受单引号、尾随逗号、注释等非标准 JSON 语法开发时确实省事。但容错解析要谨慎。它接受更宽语法不代表业务数据结构正确它内部做了很多隐性修复会让你更难定位模型输出问题。更推荐的做法是后处理先把输出规整成标准 JSON再用标准解析器解析这样线上返回给其它服务的数据始终是标准 JSON不会因为解析器差异导致跨语言消费问题。6. 第四层重试与降级策略6.1 先给错误分类哪些重试有用哪些重试是浪费重试不是所有错误都能处理。错误分类是重试策略的第一步。错误类型示例是否重试原因格式随机性错误JSON 语法错误、schema 缺字段、截断应该重试重新采样可能输出合法结果服务暂时性故障429 限流、5xx、超时应该重试等待后可能恢复输入或参数错误400 参数错误、context length 超限不重试重试大概率同样失败内容策略拦截提示词触发内容安全规则不重试需要改输入而不是重试业务条件不具备输入信息本身缺少答案字段来源不重试这是输入问题不是输出问题代码里用异常类型区分而不是靠字符串匹配错误信息。def classify_exception(exc: Exception): if isinstance(exc, RetryableError): return True return False6.2 退避重试要给随机抖动重试间隔不能固定否则多个请求同时失败后会同时重试形成雪崩。推荐指数退避加随机抖动。import random def backoff_delay(attempt: int, base: float 1.0, cap: float 8.0, jitter: float 0.3) - float: exp min(base * (2 ** (attempt - 1)), cap) return exp random.uniform(0, exp * jitter)attempt从 1 开始第一次重试间隔约 1 秒第二次约 2 秒第三次约 4 秒最大不超过 8 秒。抖动的作用是防止同一批请求在同一个时间点同时重试。重试次数要有限制不能无限循环。一般结构化提取任务建议 2 到 3 次重试。重试次数太多单次请求耗时会过长用户体验和成本都不可控。6.3 降级和熔断要考虑可用性重试耗尽之后必须有降级路径。常见降级方式切换到备用模型或更保守的模型版本。使用最近一次成功返回的缓存结果。返回一个固定结构的兜底 JSON比如{answer: 暂时无法回答, confidence: 0, sources: []}。直接把错误抛给上层由调用方决定如何处理。如果系统连续失败次数超过阈值要先熔断不再进入模型调用直接走降级等冷却期后再恢复。熔断能避免大模型服务出现故障时整个业务系统被拖垮。7. 完整实现一个可运行的 Python 稳定输出管道7.1 模块划分与目录结构把前面四层落到代码里建议按模块拆分方便测试和替换。llm_json_pipeline/ ├── prompt.py # 提示词模板 ├── client.py # 模型请求封装 ├── validator.py # 语义解析和 Schema 校验 ├── repair.py # 后处理修复 ├── pipeline.py # 四层主流程 └── main.py # 示例入口模块划分的原则提示词、模型调用、校验、修复互不依赖哪个模块想换都能单独替换。7.2 请求封装与提示词组装client.py用一个统一的LLMClient封装模型调用调用方传入 prompts返回原始文本。from openai import OpenAI class LLMClient: def __init__(self, api_key: str, base_url: str, model: str): self.client OpenAI(api_keyapi_key, base_urlbase_url) self.model model def chat( self, messages: list[dict], temperature: float 0.1, max_tokens: int 1024, response_format: dict | None None, ) - str: kwargs { model: self.model, messages: messages, temperature: temperature, max_tokens: max_tokens, } if response_format: kwargs[response_format] response_format resp self.client.chat.completions.create(**kwargs) return resp.choices[0].message.content注意openaiSDK 的版本不同参数名可能有差异。本地部署的模型如果暴露了 OpenAI 兼容接口也可以复用这个封装。7.3 校验、修复、重试主流程pipeline.py把四层串起来import logging import time logger logging.getLogger(__name__) class LLMJsonPipeline: def __init__( self, client: LLMClient, schema: dict, primary_model: str, fallback_model: str | None None, max_attempts: int 3, base_delay: float 1.0, ): self.client client self.schema schema self.primary_model primary_model self.fallback_model fallback_model self.max_attempts max_attempts self.base_delay base_delay def run(self, question: str, context: str): messages build_messages(question, context) last_error None for attempt in range(1, self.max_attempts 1): try: raw_text self.client.chat( messagesmessages, temperature0.1, max_tokens1024, response_format{type: json_object}, ) text extract_json(raw_text) text repair_json(text) data safe_loads(text) validate_data(data, self.schema) logger.info(attempt %s succeeded, attempt) return data, raw_text except NotRetryableError as exc: raise exc except Exception as exc: last_error exc logger.warning(attempt %s failed: %s, attempt, exc) if attempt self.max_attempts: time.sleep(backoff_delay(attempt, baseself.base_delay)) return self._fallback(question, context, last_error) def _fallback(self, question, context, last_error): if self.fallback_model: try: raw_text self.client.chat( messagesbuild_messages(question, context), temperature0.0, max_tokens1024, ) text extract_json(raw_text) text repair_json(text) data safe_loads(text) validate_data(data, self.schema) return data, raw_text except Exception as exc: logger.error(fallback model failed: %s, exc) degraded { answer: 暂时无法回答, confidence: 0, sources: [], } logger.error(all attempts failed: %s, last_error) return degraded, 主流程里每个失败分支都会记录日志成功分支返回(data, raw_text)调用方如果需要排查可以拿到原始输出。7.4 运行效果与日志观察在main.py里跑一个示例if __name__ __main__: logging.basicConfig(levellogging.INFO) client LLMClient( api_keyyour-api-key, base_urlhttp://localhost:8000/v1, modelprimary-model, ) pipeline LLMJsonPipeline( clientclient, schemaSCHEMA, primary_modelprimary-model, fallback_modelNone, max_attempts3, base_delay1.0, ) data, raw pipeline.run( question今天上海适合户外跑步吗, context上海今日多云气温 22 到 28 度空气质量良。, ) print(data)正常情况下会看到INFO attempt 1 succeeded {answer: 适合户外跑步, confidence: 0.85, sources: []}如果第一次输出格式有问题日志会是WARNING attempt 1 failed: JSON 语法解析失败: Expecting , delimiter... WARNING attempt 2 failed: JSON 语法解析失败: ... INFO attempt 3 succeeded通过日志可以直观看到重试是否生效以及是否出现了大面积失败。8. 线上排错从现象倒推根因8.1 按这条链路逐层排查线上出现 JSON 输出异常时不要直接改提示词按下面的顺序排查先确认原始输出有没有被完整记录。没有原始输出后面所有分析都是猜测。再确认是不是截断。看是否每次都在同一个位置截断如果是检查max_tokens是否不足。然后确认提示词是否真的按预期传给模型。检查有没有别的前置逻辑改写或拼接了提示词。接着确认结构化输出参数是否真的生效。有些模型版本对response_format的支持不完全参数传了也可能被忽略。然后检查重试是否在同一个模型、同一组参数下重复避免“重试只是浪费时间”。最后确认降级路径是否正常。降级结果有没有被当作正常结果返回给调用方。8.2 常见现象、原因与处理对照表问题现象可能原因检查方式处理建议返回合法 JSON 但缺字段提示词和 schema 不一致对比提示词模板与 Schema 的 required 字段统一用字段定义生成提示词和 Schema每次都在同一位置截断max_tokens不足看 raw_text 是否以半截结构结尾增大max_tokens或压缩上下文长度输出带代码块模型默认 Markdown 风格看 raw_text 开头是否是 后处理提取代码块同时提示词强调不要代码块偶尔多一个逗号采样随机噪声看错误日志中的 JSONDecodeError后处理修复尾随逗号再重试重试 3 次全部失败模型降级、输入不适合该模型统计连续失败率检查输入长度、内容策略必要时换模型结构化输出模式仍出现非法 JSON模型版本或参数名不支持直接打印请求参数确认当前模型是否真的支持该参数成功但业务结果不对Schema 太松检查 schema 的 type、required收紧密集字段约束增加业务规则校验排查时建议给每次请求分配一个 request_id从入口到重试、降级都带上这样日志才能串联起来。9. 生产落地建议与扩展方向9.1 学习环境和生产环境要区别对待本地实验阶段可以直接用json.loads解析重点观察模型输出长什么样。开发环境可以详细记录原始输出方便分析失败样本。测试环境要加入失败注入模拟模型输出带代码块、截断、缺字段等情况验证四层逻辑是否都生效。生产环境还要额外考虑原始输出日志脱敏避免敏感信息落库。记录成功率、各类失败率、平均耗时、重试次数等指标。对降级结果打上明确标记避免业务层把兜底 JSON 当成真实回答。控制重试成本和总耗时为单次请求设置整体超时。大模型服务压力大时优先熔断而不是无限重试。9.2 上线前检查清单检查项确认标准四层是否都实现提示词、校验、修复、重试降级缺一不可Schema 与业务字段一致required、type、取值范围都明确原始输出有记录每个失败样本都能找到原始文本和请求参数重试配置可调重试次数、退避基数、抖动参数通过配置控制降级路径无副作用兜底 JSON 不会进入下游写入流程指标齐全能统计成功率、失败原因分布、重试成功率灰度计划先让一部分请求走新逻辑观察指标再全量9.3 进阶方向如果四层方案已经稳定下一步可以往更底层去优化。对自建推理服务可以引入受控解码或语法约束生成让模型在解码阶段就不可能生成非法 JSON。对闭源模型可以优先使用 function calling / tool calling 替代裸 JSON 输出把结构化结果交给模型自身机制管理。对提示词和 Schema可以做版本管理每次调整都能对比格式成功率变化。对复杂业务建议用pydantic定义输出模型再自动生成 JSON Schema保证类型定义和校验逻辑不脱节。只要保证每个失败场景都有兜底路径JSON 输出从概率问题就会变成可观测、可控制、可恢复的工程问题。
