线上大模型项目最让人头疼的问题之一,就是JSON输出格式不稳定。同一个Prompt今天能返回标准JSON,明天可能多出一段解释文字,后天可能在字符串里混入未转义换行,最后直接导致下游解析失败、任务中断、数据入库报错。很多人第一反应是继续调Prompt,但其实这个问题更适合用工程手段解决:提示词约束、格式校验、后处理修复、重试降级,四层配合才能形成一套稳定方案。这篇文章会围绕这条主线,给出可以落地的代码示例和通用排查思路。
先说结论:不要指望大模型每次都输出完美JSON,而是要把“输出不可靠”当成一个默认前提,在解析链路里做好防御。下面会按“为什么输出会飘、先怎样约束、再怎样校验、坏了怎样修复、还不行怎样重试”这个顺序展开。适合正在做线上 AI 应用、数据管道、Agent 工具调用、批量文本解析的开发者阅读。
1. 核心方案速览
| 能力项 | 说明 |
|---|---|
| 问题场景 | 线上大模型接口返回的 JSON 格式不稳定,导致下游解析失败 |
| 方案定位 | 不依赖单次 Prompt 调优,以工程手段做多层兜底 |
| 核心分层 | Prompt 输出协议、JSON 语法校验、Schema 校验、破损修复、重试与降级 |
| 是否依赖微调 | 否,纯调用层与解析层改造 |
| 是否依赖特定模型 | 否,OpenAI、Claude、本地开源模型均可适用 |
| 是否支持批量任务 | 是,可套用队列与日志记录方式处理大批量文本 |
| 是否提供接口 API | 方案可嵌入现有服务,输出为统一解析结果对象 |
| 主要开发语言 | Python 为主,思路可迁移到 Java、TypeScript、Go |
这套方案的重点是“可落地”。它不是一篇概念文章,而是把线上调用大模型返回结果的防御流程拆开,每一层都给出具体的代码形态、判断标准、失败处理方式。
2. 适用场景与使用边界
这套方案适合几类场景:第一类是数据抽取应用,比如从合同、简历、工单中提取结构化字段,大模型返回的 JSON 必须稳定映射到数据库字段;第二类是 Agent 工具调用,模型生成 JSON 格式的函数参数,一旦格式错误,工具调用直接失败;第三类是批量内容处理,比如每天跑数千条文本摘要、分类、实体识别,如果解析错误率偏高,会消耗大量人工成本。
它也有限制。它不能解决模型本身“能力不足”的问题。如果模型在特定任务上始终生成错误内容,比如必填字段缺失、语义理解错误、字段值乱写,那么格式兜底只能保证“JSON 合法”,不能保证“业务内容正确”。这种场景需要回到数据标注和微调方向,不能用解析逻辑硬扛。
同时要明确数据安全边界。如果你使用的是第三方云上大模型 API,注意不要将未脱敏的隐私信息、密钥、版权文本直接传入;使用本地部署模型时,做好模型文件的授权确认和访问控制。涉及真实用户数据时,先做好脱敏、授权和审计。
3. 为什么大模型 JSON 输出会飘忽不定
在构建防御方案之前,先理解问题来源。大模型生成文本本质上是逐 token 采样,同一个 Prompt 在不同温度、不同采样参数下会产生不同的文本分布。即使设置了较低温度,模型也可能在 JSON 对象之外补充“好的,以下是我生成的 JSON:”这类说明文字,或者在 JSON 末尾追加多余注释。这些在人类看来无害的文本,对json.loads来说就是致命错误。
更常见的原因是输出截断。线上接口往往有max_tokens限制,生成长 JSON 时可能在中途被截断,导致缺少右括号、字符串未闭合、数组未结束。另一个高频问题是转义处理不当:模型在字符串字段里插入换行符\n时,有时会输出真实换行符而不是转义序列,导致 JSON 解析直接失败。
此外,模型可能把 JSON 包在 Markdown 代码块里:
{ "name": "test" }这种情况非常常见,尤其是使用了带补全接口或聊天接口的模型。还有一类问题是编码相关的,比如中文标点被混入、全角冒号替代半角冒号、字符串值里出现控制字符等。这些不确定因素叠加在一起,就解释了为什么光靠 Prompt 很难根除问题——你无法控制模型的每一次采样过程。
4. 第一层:提示词约束与输出协议
提示词虽然不能 100% 保证输出稳定,但它决定了问题的基础概率。一个好的输出协议会把“输出什么格式”和“不要输出什么”写得非常具体。
4.1 固定 JSON Schema 说明
可以在 System Prompt 中固定 JSON Schema,并要求模型严格按键名输出。下面是一个可复用的模板,需要按实际任务替换字段和示例。
你是一个数据结构化助手。你只能输出一个 JSON 对象,不要输出其他任何文字、解释或 Markdown 代码块。 输出格式必须严格符合以下 JSON Schema: { "type": "object", "properties": { "title": {"type": "string"}, "tags": {"type": "array", "items": {"type": "string"}}, "summary": {"type": "string"} }, "required": ["title", "tags", "summary"] } 要求: 1. 输出内容必须是合法 JSON,不能包含注释。 2. 字符串值中的换行必须使用转义序列 \\n。 3. 不要使用代码块包裹 JSON。 4. 只输出 JSON 本身。这段提示词把返回协议拆成了三部分:格式定义、字段要求、易错点提醒。相比笼统的“请返回 JSON”,它能减少模型自由发挥的空间。
4.2 Few-shot 示例
对大模型来说,给出一个输入输出对往往比长篇描述更有效。可以在 Prompt 里附加一条输入文本和一条标准输出示例。关键是示例的 JSON 必须完全规范,最好与真实业务字段一致,否则模型会照搬示例中的错误格式。
4.3 使用接口层强制 JSON 模式
如果使用 OpenAI 等支持结构化输出的接口,可以直接启用 JSON 输出模式或结构化输出功能。这类模式会在采样层面约束模型输出为合法 JSON,能大幅降低格式错误概率,但不能完全消除截断和 schema 不匹配问题,所以后面的校验和后处理仍然需要保留。
一个通用调用示例,具体参数需要按服务商文档调整:
import openai client = openai.OpenAI( api_key="your-api-key", base_url="https://your-endpoint/v1" ) response = client.chat.completions.create( model="gpt-4o-mini", response_format={"type": "json_object"}, messages=[ {"role": "system", "content": "只输出 JSON 对象,不要解释。"}, {"role": "user", "content": "提取标题、标签、摘要"} ] ) raw = response.choices[0].message.content print(raw)需要注意的是,不同服务商的response_format参数名称和取值可能不同,接入前先看文档。本地部署模型通常通过推理引擎的grammar或json_schema参数实现类似能力,效果也取决于模型和引擎版本。
4.4 控制采样参数
temperature调低可以降低输出随机性,一般设置为 0 或接近 0,更严格的任务可以直接用 greedy decoding。部分接口还支持top_p、seed参数,固定 seed 能提高同参数下的可复现性,但并不能保证不同调用之间绝对一致。
5. 第二层:响应校验模块设计
Prompt 只是降低概率,真正要兜住线上稳定性,需要一个独立的校验模块。校验模块承担两个职责:判断输出是否是合法 JSON;判断 JSON 结构是否满足业务要求。
5.1 基础 JSON 语法校验
最简单的做法是直接json.loads,并捕获解析异常。但这种粗暴方式无法给出足够定位信息。建议封装一个统一解析函数,能返回错误阶段和错误片段。
import json def parse_json_safe(raw_text: str): """ 尝试把模型输出解析为 JSON。 返回: (data, error) 成功时 data 为解析结果,error 为 None; 失败时 data 为 None,error 为错误信息。 """ if not raw_text or not isinstance(raw_text, str): return None, "empty or non-str output" try: return json.loads(raw_text), None except json.JSONDecodeError as e: return None, f"JSONDecodeError at line {e.lineno} col {e.colno}: {e.msg}"5.2 Schema 校验
语法校验只能保证“能解析”,不能保证“结构对”。比如模型返回了一个合法的 JSON 数组,但业务需要的是对象;或者返回了对象但缺少title字段。这些都需要用 JSON Schema 校验。
from jsonschema import validate, ValidationError # 以业务需要的数据结构为例,按实际任务替换 SCHEMA = { "type": "object", "properties": { "title": {"type": "string"}, "tags": { "type": "array", "items": {"type": "string"}, "minItems": 1 }, "summary": {"type": "string"} }, "required": ["title", "tags", "summary"], "additionalProperties": False } def validate_with_schema(data): try: validate(instance=data, schema=SCHEMA) return True, None except ValidationError as e: return False, e.message这里额外设置了additionalProperties: False,可以防止模型输出无关字段。但在实际业务中,如果模型偶尔会多输出字段,直接拒绝可能导致重试率上升,建议谨慎使用。如果只是需要忽略多余字段,可以把该配置去掉。
5.3 字段级业务校验
Schema 只能约束类型,约束不了业务范围。比如tags数组里必须是合法标签、summary不能为空字符串、日期字段需要符合YYYY-MM-DD格式。这些校验建议单独写函数,便于扩展和测试。
def validate_business_fields(data): if not data.get("title", "").strip(): return False, "title is empty" if not data.get("summary", "").strip(): return False, "summary is empty" if len(data.get("tags", [])) > 5: return False, "too many tags" return True, None如果模型在 JSON 合法但内容不合法时,不应该直接进入下游逻辑,应该走后续的修复或重试分支。
6. 第三层:后处理修复损坏 JSON
即便加了 Prompt 约束,线上仍可能出现 5% 到 20% 的格式偏离。后处理修复的价值在于,把高频的小毛病直接修掉,避免一出现问题就重新调用模型,既节省 token 又降低延迟。
6.1 剥离代码块与前后缀
最典型的问题是模型把 JSON 放在 Markdown 代码块里,或者前后有解释性文字。可以用正则或字符串切片提取核心 JSON 片段。
import re def extract_json_block(raw_text: str): """从模型输出中提取包含 JSON 的候选片段。""" # 1. 优先匹配 ```json ... ``` pattern = re.compile(r"```(?:json)?\s*([\s\S]*?)```", re.IGNORECASE) matches = pattern.findall(raw_text) if matches: return matches[-1].strip(), "code_block" # 2. 没有代码块时,截取从第一个 { 或 [ 到最后一个 } 或 ] start = min( [idx for idx in (raw_text.find("{"), raw_text.find("[")) if idx != -1], default=-1 ) end = max( raw_text.rfind("}"), raw_text.rfind("]") ) if start >= 0 and end > start: return raw_text[start:end + 1].strip(), "bracket_slice" # 3. 没有候选片段,返回原文交给解析函数判断 return raw_text.strip(), "raw"6.2 常见 JSON 破损修复
解析失败后,可以按错误类型做多级修复。下面这个函数处理了一批常见问题,每次修复后再尝试解析,直到成功或所有策略耗尽。
def repair_json_candidate(candidate: str): """ 针对常见 JSON 破损做多轮修复。 注意:这是尽力修复,不是万能方案。 """ # 修复控制字符 candidate = re.sub(r"[\x00-\x1f]", "", candidate) # 去掉可能拼接的尾部逗号 candidate = re.sub(r",\s*([}\]])", r"\1", candidate) # 把单引号替换为双引号(粗略方案,只适合简单结构) candidate = re.sub(r"'", '"', candidate) # 把全角冒号替换为半角冒号 candidate = candidate.replace(":", ":") # 尝试补齐缺失结尾括号(只处理最后一位是逗号或冒号的情况) candidate = candidate.rstrip().rstrip(",").rstrip(":") return candidate def safe_extract_json(raw_text: str): """ 核心入口:尝试提取并解析 JSON。 成功返回 (data, repaired_flag, error) """ data, err = parse_json_safe(raw_text) if data is not None: return data, False, None fallback, source = extract_json_block(raw_text) if fallback != raw_text: data, err = parse_json_safe(fallback) if data is not None: return data, True, None repaired = repair_json_candidate(fallback) data, err = parse_json_safe(repaired) if data is not None: return data, True, None return None, False, err这里要强调一下:修复策略不能写得过重。过度启发式替换带来的风险是,它可能会把字符串里的合法内容改坏。比如字符串中出现了“don't”这样的英文缩写,粗暴的单引号替换会破坏内容。因此修复逻辑应该放在解析失败之后,而且任何修复后的结果都必须再经过一次 schema 校验。
6.3 截断内容补齐尝试
针对max_tokens截断导致的 JSON 不完整,如果任务字段有限,可以尝试写一个简单的补齐函数。比如模型输出到{"title": "abc", "tags": ["a"被截断,此时可以尝试补全数组和对象闭合。
def complete_truncated_json(candidate: str): """针对常见的截断位置尝试补齐括号,不保证成功。""" opening = candidate.count("{") + candidate.count("[") closing = candidate.count("}") + candidate.count("]") diff = opening - closing if diff > 0: candidate += "}" * diff return candidate return candidate这种方式只适用于“单纯缺右括号”的截断场景。如果截断发生在字符串值中间,补全后依然无法解析,此时应该走重试而不是继续修补。
6.4 修复后的验证闭环
所有修复结果都必须回到同一套解析函数和 schema 校验函数验证。不要让“修复成功”成为绕过校验的口子。建议把修复结果、修复策略、是否通过校验都写入日志,便于后续统计哪种修复策略在真实业务里最高频、哪种策略可能引入错误。
7. 第四层:重试与降级策略
修复不是万能的,当修复后依然解析失败,或 schema 校验不通过时,需要考虑重新调用模型。重试不是无脑循环,而是要设计成有节奏、有上限、有降级的机制。
7.1 按错误类型决定是否重试
建议把错误分成三类:可重试错误、不可重试错误、业务校验错误。可重试错误包括空输出、JSON 解析失败、截断、字段类型不对等;不可重试错误包括接口鉴权失败、余额不足、请求参数不合法、服务端限流等。业务校验错误要区分情况:如果模型输出内容与业务规则冲突,重试一次说不定能生成为合规数据,但第三次如果仍失败,应停止重试并转入人工处理。
RETRYABLE_ERROR_KEYWORDS = [ "JSONDecodeError", "truncated", "empty", "out of memory", ] def is_retryable(error: str) -> bool: if not error: return False return any(keyword.lower() in error.lower() for keyword in RETRYABLE_ERROR_KEYWORDS)7.2 通用重试模板
推荐使用指数退避,限制最大重试次数,并记录每次重试的原因和耗时。
import time def call_with_retry( generate_fn, schema, max_retries=3, backoff_base=1.0, backoff_max=10.0 ): """ generate_fn: 无参数函数,返回原始模型输出字符串。 schema: JSON Schema,用于校验。 """ last_error = None for attempt in range(max_retries): try: raw_text = generate_fn() except Exception as e: last_error = str(e) if not is_retryable(last_error): raise wait = min(backoff_base * (2 ** attempt), backoff_max) time.sleep(wait) continue data, repaired, err = safe_extract_json(raw_text) if data is not None and schema is not None: ok, schema_err = validate_with_schema(data) if not ok: last_error = f"schema error: {schema_err}" wait = min(backoff_base * (2 ** attempt), backoff_max) time.sleep(wait) continue elif data is None: last_error = err wait = min(backoff_base * (2 ** attempt), backoff_max) time.sleep(wait) continue return data, repaired, attempt raise RuntimeError(f"call_with_retry exceeded max_retries: {last_error}")这个模板把原始调用、解析、校验、重试合并成一个入口,业务方不需要关心中间细节。可以根据实际需要,把generate_fn换成你自己的 API 调用函数。
7.3 多候选结果选择
重试是串行的,有一定延迟开销。如果接口允许一次传入多个候选结果,或者你可以并发调用多次大模型,可以用“多候选选择”策略:生成 n 份输出,先对每个候选做解析和 schema 校验,选择第一个或评分最高的合法结果。该策略对接口成本和延迟都有压力,适用于对延迟不敏感、对稳定性要求极高的任务。
7.4 降级策略
当重试次数耗尽后,不要让任务直接失败,应该走预定义降级方案。常见做法是:把原始输出和错误信息写入失败表,返回一个可控的默认结构,由后续人工或规则流程补全。降级结构需要保证下游不会因缺字段而崩溃。
FALLBACK_RESULT = { "title": "", "tags": [], "summary": "", "_parse_error": None } def downgrade_fallback(raw_text, error): result = dict(FALLBACK_RESULT) result["_parse_error"] = error result["_raw_output"] = raw_text return result8. 批量任务处理与接口集成
线上场景往往不是单条调用,而是批量任务。批处理时,要把“格式解析”和“任务调度”分开,避免一条坏数据阻塞整个队列。
8.1 批量任务循环示例
import json import logging from pathlib import Path logger = logging.getLogger(__name__) def process_batch(input_items, generate_fn, schema, output_path, max_retries=3): """ input_items: list[dict],包含每条待处理任务的原始字段。 generate_fn: function(template, item) -> str,返回模型输出。 """ results = [] failed_items = [] for idx, item in enumerate(input_items, start=1): try: data, repaired, attempt = call_with_retry( lambda: generate_fn(item), schema, max_retries=max_retries ) results.append({ "index": idx, "item_id": item.get("id"), "parse_result": data, "repaired": repaired, "retry_count": attempt }) logger.info("item %s ok, retry=%s", item.get("id"), attempt) except Exception as e: logger.exception("item %s failed: %s", item.get("id"), e) failed_items.append({ "index": idx, "item_id": item.get("id"), "error": str(e), "raw_output": getattr(e, "raw_output", None) }) # 成功结果与失败结果分开保存,失败项可后续重跑 Path(output_path).mkdir(parents=True, exist_ok=True) with open(Path(output_path) / "success.jsonl", "w", encoding="utf-8") as f: for r in results: f.write(json.dumps(r, ensure_ascii=False) + "\n") with open(Path(output_path) / "failed.jsonl", "w", encoding="utf-8") as f: for r in failed_items: f.write(json.dumps(r, ensure_ascii=False) + "\n") return results, failed_items批量任务的关键是:每一条都独立 catch 异常,失败项写failed.jsonl,下一次可以只重跑失败文件。这样即便中间有几十条格式异常,也不会拖垮整批任务。
8.2 接口服务接入示例
如果要把这套解析逻辑接入一个 Web API,可以用以下结构:
from flask import Flask, request, jsonify app = Flask(__name__) @app.post("/api/extract") def extract(): payload = request.get_json(force=True) text = payload.get("text", "") def generate_with_retry(): return call_model_api(text) try: data, repaired, attempt = call_with_retry( generate_with_retry, SCHEMA, max_retries=3 ) return jsonify({ "code": 0, "data": data, "repaired": repaired, "retry_count": attempt }) except Exception as e: return jsonify({ "code": 1, "message": str(e) }), 502 if __name__ == "__main__": app.run(host="127.0.0.1", port=8000)代码里的call_model_api需要替换为你的真实模型调用函数。接口返回中带上repaired和retry_count字段,便于调用方了解这次结果是否需要复核。
8.3 日志与可观测性
每条调用建议至少记录以下信息:任务 ID、输入摘要、原始输出长度、解析结果、修复标记、重试次数、耗时、是否走了降级。有了这些数据,才能判断方案是否真的稳定,也才能发现新的高频错误类型。
9. 资源占用与性能观察建议
这套方案本身只涉及文本处理和少量正则运算,对 CPU 和内存占用非常低,主要成本集中在模型调用环节。运行时需要重点观察三个指标。
第一是首次解析成功率。它代表模型输出的原始质量,如果这个指标长期低于 80%,说明 Prompt 约束或模型选型需要调整。第二是修复成功率。在首次解析失败后,多少比例能通过后处理修复救回,这决定了 token 成本和延迟。第三是重试率。如果重试率过高,说明模型的输出稳定性和 Prompt 约束都有问题,要做结构性优化。
延迟方面,一次模型调用通常需要数秒,后处理和校验代码应该在毫秒级完成。如果后处理超过 100ms,要检查是否在循环中频繁做了大规模正则匹配或重复解析。批处理时,建议使用线程池或异步调用并发发送模型请求,但要注意接口限流和超时设置。
避免把重试写成无限循环。没有上限的重试会显著增加 token 消耗和延迟,也可能触发服务商限流。通常 2 到 3 次重试已经能覆盖大部分瞬时波动,再高基本是无效成本。
10. 常见问题与排查方法
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 首次解析成功率很低 | Prompt 约束不清晰,模型自由度太高 | 打开原始输出日志,观察常见错误形态 | 增加 JSON Schema 与 few-shot 示例,降低 temperature |
| 模型输出包含 Markdown 代码块 | 模型默认偏好以代码块展示 | 检查原始输出是否包含 ``` 包裹 | 使用 extract_json_block 剥离代码块 |
| 字符串中出现未转义换行 | 模型把真实换行输出到 JSON 字符串 | 查看解析异常位置行号列号 | 在 Prompt 中显式要求使用 \n,并在修复阶段过滤控制字符 |
| JSON 结构合法但缺字段 | 模型理解偏差,或 schema 描述不充分 | 检查 schema 校验返回的错误字段 | 在 Prompt 中补充字段说明和必填强调 |
| 截断导致 JSON 不完整 | max_tokens 设置过小或文本过长 | 检查输出长度是否接近 max_tokens | 调大 max_tokens,或先压缩输入文本,再用补齐策略 |
| 重试次数耗尽 | 模型持续输出错误格式或限流 | 查看重试日志中的累计错误原因 | 增加降级逻辑,标记失败项人工处理 |
| API 调用返回限流 | 请求频率过高或并发数过大 | 查看接口返回的状态码和 retry_after | 增加并发控制、指数退避或排队机制 |
| 修复后内容被改坏 | 启发式修复误伤了合法字符串 | 抽样对比修复前后的解析结果 | 缩小修复范围,最多只做一两次替换,不做深度改写 |
11. 最佳实践与使用建议
这套方案能不能真正稳定,取决于几个工程习惯。
第一,所有 Prompt 和 schema 要版本化。不要只把 Prompt 写在代码字符串里,建议放到独立配置文件或数据库中,便于回滚和对比测试。schema 变更时要同步升级 prompt 示例,避免两者不一致。
第二,建立回归测试集。准备 30 到 50 条带有真实业务特征的输入样本,记录它们在不同模型版本、不同 Prompt 版本下的解析成功率。每次修改 Prompt 或解析逻辑后,跑一遍回归集,防止“修好 A 类错误、搞坏 B 类错误”。
第三,解析函数要保持无状态、可独立测试。把extract_json_block、repair_json_candidate、validate_with_schema拆成纯函数,单测覆盖常见坏样本,避免在业务代码里维护一大坨嵌套逻辑。
第四,不要把修复函数写得越来越长。修复策略每增加一条,都可能引入新的误伤风险。建议限制修复函数的替换次数,把所有修复策略限制在 5 到 10 个高频场景内。超出范围的错误,宁可重试或降级。
第五,内容合规与授权同样重要。如果产品涉及提取他人文章、简历、聊天记录等信息,必须确保有合法数据来源和用户授权。模型输出结果在对外展示前,也要经过内容安全过滤和人工复核,尤其是涉及敏感人物、品牌、医疗、金融等高风险内容时。
12. 总结与下一步
这次讨论的方案,核心是把大模型 JSON 输出不稳定问题从“Prompt 玄学”变成“工程可治理”。先通过 Prompt 输出协议降低初犯概率,再用校验模块识别问题,接着用后处理修复常见破损,最后用重试和降级兜底。四层各司其职,缺一不可。真正上线时,最先要做的是把原始输出日志完整记下来,统计一周内所有解析失败的类型和频率,针对性优先修复 TOP 3 问题。最容易踩的坑是把重试当成万能药,以及在后处理里塞入过多的启发式替换逻辑。
后续可以继续扩展的方向包括:接入结构化输出能力或本地模型的 grammar 约束、引入 Prometheus 监控指标、在批量任务中增加基于解析成功率的分流策略,甚至针对高频坏样本构建微调数据集。先把四层防御跑通,再去谈进一步压测和优化。