线上大模型项目最烦的问题之一,就是模型明明在跑,JSON 输出却“随缘”:字段顺序变了、多一个逗号、少一个括号、把"status": "success"写成"status": success、甚至整段返回一段 Markdown 代码块包着 JSON。日志里看到json.loads抛JSONDecodeError,第一反应是去改 Prompt,改完这轮好了,下一轮又崩。
这次我们不把精力全押在 Prompt 上。真正能在生产环境里扛住的方案,是把“输出稳定”当成一个完整链路来做:提示词约束 + 模型结构化输出 + 宽松解析 + 数据校验 + 后处理修复 + 重试退避。这篇文章会按这套链路拆开讲,并给出一套可以直接落到代码里的 Python 实现。
适合正在接入大模型 API、做批量数据抽取、做 Agent 工具调用、做内容结构化落库的开发者。本文不是某个模型的安装教程,而是一整套“降低 JSON 解析失败率”的工程方法,项目代码结构也可以直接搬到自己的服务里。
1. 核心能力速览
| 能力项 | 说明 |
|---|---|
| 问题类型 | 大模型返回 JSON 格式不稳定、字段缺失、类型错误、嵌套结构错乱 |
| 核心思路 | 不依赖纯 Prompt 微调,采用提示词 + 结构化输出 + 校验 + 后处理 + 重试的多层方案 |
| 适用场景 | 线上 API 调用、批量数据抽取、Agent 工具调用、数据落库、工作流编排 |
| 关键依赖 | Python 3.9+、openai / anthropic / 任意 HTTP 接口、pydantic、jsonschema |
| 模型要求 | 优先支持 JSON Mode / Function Calling / 结构化输出的模型;普通 Chat 模型也能降级使用 |
| 部署方式 | 可作为独立服务模块、中间件或 SDK 封装,不限制框架 |
| API 能力 | 支持统一封装,屏蔽各家模型接口差异 |
| 批量任务 | 支持批量请求、并发控制、失败重试、结果落盘 |
| 可观测性 | 记录每次解析失败原因、重试次数、最终成功状态 |
2. 适用场景与使用边界
这套方案适合以下几类场景:
- 批量数据抽取:从非结构化文本里抽取实体、关系、情感标签,返回结果需要落库。
- Agent 工具参数生成:模型需要输出调用工具的
function_call参数,JSON 不合法就直接调用失败。 - 内容结构化:把模型生成的文章、评论、客服话术转成固定字段,对接下游系统。
- 多模型切换:同一套业务代码要兼容 OpenAI、Anthropic、本地部署模型,接口差异统一屏蔽。
不适合的场景也要说清楚:如果业务要求 JSON 输出必须 100% 可用,且完全不允许任何代价,那任何层级的重试和后处理都只是降低概率,不能消灭异常。关键业务建议在最终环节增加人工确认或二次校验。
使用边界方面,文章里涉及的数据最好是脱敏或测试数据,不要向线上模型接口发送未经授权的用户隐私、商业机密、版权素材。批量调用第三方模型接口时要注意 API 限额和调用频率,合理设置退避。涉及国内模型和海外模型时,按各平台的使用条款来,不要用测试账号去压测付费接口。
3. 环境准备与前置条件
3.1 基础环境
建议 Python 3.9 以上,安装以下依赖:
pip install openai pydantic jsonschema tenacity如果你用的是 Anthropic 或其他模型,再补对应 SDK:
pip install anthropic如果你用的是本地模型或自建代理,那么只需要requests就够了。
3.2 模型能力确认
动手前先确认你用的模型支持哪些能力:
| 能力 | 说明 | 支持示例 |
|---|---|---|
| JSON Mode | 模型始终输出合法 JSON | OpenAI gpt-4o / gpt-4-turbo 的response_format={"type": "json_object"} |
| Function Calling | 模型按工具定义返回结构化参数 | OpenAI / Anthropic / 部分本地模型 |
| 结构化输出 | 严格按 Schema 输出 | OpenAI 的 Structured Outputs,strict: true |
| 普通对话 | 只能靠 Prompt 约束 | 传统 Chat 模型 |
如果模型本身支持 JSON Mode 或 Function Calling,优先走协议级约束。如果不支持,才回到 Prompt 约束 + 后处理修复的老路。
3.3 准备测试素材
准备一份测试文本集合,建议 10 到 20 条不同风格的数据:
- 短文本:一句话抽取(如“帮我订一张明天北京到上海的高铁票,二等座”)。
- 长文本:一篇新闻或一段客服聊天记录。
- 边界文本:空字符串、纯标点、超长文本、重复内容。
这套素材不要只用来跑通“成功案例”,要保存一批“历史上解析失败过的真实样本”。后面调校验逻辑时,这些失败样本能不能恢复,比新样本能不能跑通更重要。
4. 第一层:提示词约束与输出结构定义
先说结论:Prompt 不是不重要,而是不能作为唯一手段。Prompt 要做好,但要做好“即使模型自由发挥,也能从结构上约束住”的准备。
4.1 编写结构化提示词
示例任务:从用户输入中抽取数据库查询条件。
你是数据抽取助手。根据用户输入,输出 JSON,必须满足以下结构: { "table": "users", "filters": [ {"field": "age", "operator": ">", "value": 18} ], "limit": 10 } 要求: 1. 只输出 JSON,不要输出任何解释文字。 2. 不要使用 Markdown 代码块包裹。 3. 字段名必须与示例完全一致。 4. filters 数组至少包含一个元素。 5. value 根据字段类型决定:字符串用双引号,数字不用引号。注意几个细节:
- 明确要求“只输出 JSON”,同时给一个完整示例。
- 要求“不要使用 Markdown 代码块包裹”。不少模型偏好把 JSON 放进
```json代码块,这对人工阅读没问题,但对程序解析就是 bug。 - 给字段类型和示例值,避免模型把
{"value": "18"}输出成字符串。
4.2 Prompt 约束的局限
Prompt 约束的失败模式很固定:
- 模型偶尔输出解释文字,比如“好的,这是你要的 JSON:”。
- 字段名被改名,比如
filters变成conditions。 - 数组嵌套层级不对。
- 值类型漂移。
- 多语言混排。
所以 Prompt 能做的只是“降低失败概率”,不能保证格式。下面所有层都是在处理 Prompt 没拦住的情况。
5. 第二层:模型协议级约束
这一层是“让模型在协议层面避免产生非法 JSON”。
5.1 JSON Mode
在 OpenAI 兼容接口中,可以这样开启:
from openai import OpenAI client = OpenAI() resp = client.chat.completions.create( model="gpt-4o", response_format={"type": "json_object"}, messages=[ {"role": "system", "content": "你只输出 JSON。"}, {"role": "user", "content": "抽取用户意图"} ], ) print(resp.choices[0].message.content)JSON Mode 能有效防止模型输出普通文本,但有几个注意点:
- prompt 里必须出现“json”这个词,否则部分模型会报错。
- JSON Mode 只保证输出是合法 JSON,不能保证字段结构符合你的预期。
- 不同部署平台对 JSON Mode 的支持情况不同,本地模型用 vLLM 等框架也支持这类参数,但字段名可能有差异。
5.2 Function Calling
如果模型支持 Function Calling,直接把它当成“结构化输出协议”用,不要把它当“工具调用”理解。
tools = [ { "type": "function", "function": { "name": "extract_query_condition", "description": "抽取数据库查询条件", "parameters": { "type": "object", "properties": { "table": {"type": "string"}, "filters": { "type": "array", "items": { "type": "object", "properties": { "field": {"type": "string"}, "operator": {"type": "string"}, "value": {"type": ["string", "number", "boolean"]} }, "required": ["field", "operator", "value"] } }, "limit": {"type": "integer"} }, "required": ["table", "filters", "limit"] } } } ] resp = client.chat.completions.create( model="gpt-4o", tools=tools, tool_choice={"type": "function", "function": {"name": "extract_query_condition"}}, messages=[ {"role": "user", "content": "帮我查年龄大于18岁的用户,取前10条"} ] ) # 取工具调用参数 args = resp.choices[0].message.tool_calls[0].function.arguments print(args)这种做法下,模型返回的arguments本身就是 JSON 字符串,而且结构由parameters定义,比纯 Prompt 稳定得多。
5.3 结构化输出 Strict Mode
OpenAI 的结构化输出(Structured Outputs)要求提供json_schema:
resp = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "抽取查询条件"}], response_format={ "type": "json_schema", "json_schema": { "name": "query_condition", "strict": True, "schema": { "type": "object", "properties": { "table": {"type": "string"}, "filters": {"type": "array", "items": {"type": "object"}}, "limit": {"type": "integer"} }, "required": ["table", "filters", "limit"], "additionalProperties": False } } } )Structured Outputs 是当前模型侧最强的约束,所有字段都定义在 Schema 里。但 Strict Mode 对 Schema 有些限制,比如必须定义全部属性、禁止额外属性等。
5.4 本地模型 / 开源模型
如果你在用本地模型,优先看推理框架是否支持guided_json或相关参数。vLLM 支持guided_json,可以传入 JSON Schema 让模型解码时只生成合法 JSON。这是目前开源模型侧最强的一层约束,代码示例:
from vllm import LLM, SamplingParams llm = LLM(model="Qwen/Qwen2.5-7B-Instruct") sampling_params = SamplingParams( temperature=0.0, guided_json={ "type": "object", "properties": { "table": {"type": "string"}, "limit": {"type": "integer"} }, "required": ["table", "limit"] } ) output = llm.generate("帮我查用户表,取前10条", sampling_params) print(output[0].outputs[0].text)如果你的本地框架不支持这类语法,也可以退回到“Prompt 约束 + 后处理修复”。但说实话,不支持 guided decoding 的开源模型,生产环境里做复杂 JSON 任务会很痛苦。
6. 第三层:宽松解析与数据校验
模型协议层已经挡掉了大部分非法 JSON,但线上环境依然会有漏网之鱼。这时要进入第三层:解析和校验。
6.1 宽松 JSON 解析
不同模型输出脏数据的模式不一样,解析器要做几种“恢复型解析”:
- 去掉 Markdown 代码块包裹
```json和```。 - 截取第一个
{和最后一个}之间的内容。 - 去掉行尾多余逗号。
- 修正单引号。
写一个宽松解析函数:
import json import re def extract_json_str(text: str) -> str: if not isinstance(text, str): raise ValueError("input must be string") # 去除 Markdown 代码块 text = re.sub(r"```(?:json)?", "", text).strip() # 找到第一个 { 和最后一个 } start = text.find("{") end = text.rfind("}") if start == -1 or end == -1: raise ValueError("no json object found") return text[start:end + 1] def load_json_loose(text: str): """尝试多种方式解析 JSON""" candidates = [] # 方式1:原样解析 candidates.append(text) # 方式2:提取 { ... } 范围 try: candidates.append(extract_json_str(text)) except ValueError: pass # 方式3:去掉 Markdown 包裹后再提取 cleaned = re.sub(r"^```(?:json)?\s*", "", text.strip()) cleaned = re.sub(r"\s*```$", "", cleaned) candidates.append(cleaned) # 连续尝试 for cand in candidates: try: return json.loads(cand) except json.JSONDecodeError: continue # 方式4:去除超出尾部逗号 fixed = candidates[-1] if candidates else text fixed = re.sub(r",\s*([}\]])", r"\1", fixed) try: return json.loads(fixed) except json.JSONDecodeError: raise ValueError("all json parsing attempts failed")这个函数的思路是“多种策略逐个试”,而不是只调一次json.loads。实际项目中还可以加更多策略,比如把不标准的布尔值True/False替换成true/false。
6.2 数据校验
即使解析成功,结构也可能不符合预期。比如要求filters是数组,模型返回了一个对象。这里用 jsonschema 做结构校验:
import jsonschema from jsonschema import validate QUERY_CONDITION_SCHEMA = { "type": "object", "properties": { "table": {"type": "string"}, "filters": { "type": "array", "items": { "type": "object", "properties": { "field": {"type": "string"}, "operator": {"type": "string"}, "value": {"type": ["string", "number", "boolean"]} }, "required": ["field", "operator", "value"] } }, "limit": {"type": "integer"} }, "required": ["table", "filters", "limit"] } def validate_json(data): try: validate(instance=data, schema=QUERY_CONDITION_SCHEMA) return True, data except jsonschema.ValidationError as e: return False, str(e)校验失败后,不要急着直接丢弃,可以先看看能不能“修复”。这就进入下一层。
7. 第四层:后处理与字段级修复
后处理层的目标不是重写模型输出,而是做三类修复:
- 字段名修复:模型把
filters写成了conditions。 - 值类型修复:模型把
limit写成了字符串"10"。 - 结构补全:模型漏了
table字段,但可以从上下文推断。
7.1 字段名归一化
FIELD_ALIASES = { "table": ["table", "table_name", "from", "collection"], "filters": ["filters", "filter", "conditions", "wheres", "where"], "limit": ["limit", "max_results", "size", "count"] } def normalize_fields(data: dict) -> dict: normalized = {} for key, aliases in FIELD_ALIASES.items(): for alias in aliases: if alias in data: normalized[key] = data[alias] break # 保留未知字段 for k, v in data.items(): if k not in normalized: normalized[k] = v return normalized这样即使模型换了字段名,只要命中了别名表就能对齐。别名表要定期从失败日志里补充,这是一个长期维护的字典。
7.2 值类型修复
类型修复要“尽力而为”,不能把字符串"abc"强行转成数字123。这里只做安全转换:
def safe_cast_int(value, default=None): if isinstance(value, int) and not isinstance(value, bool): return value if isinstance(value, str): try: return int(value.strip()) except ValueError: return default return default def repair_types(data: dict) -> dict: if "limit" in data: repaired = safe_cast_int(data["limit"]) if repaired is not None: data["limit"] = repaired return data7.3 结构补全
如果filters为空或缺失,可以补默认值:某些业务场景下空数组比直接报错更合理。但要注意:补全动作必须有业务依据,不能瞎补。比如缺失table,如果场景里只有一张表,可以默认填主表;如果有多张表,就必须重试或报错。
修复之后一定要再跑一次 schema 校验,如果还不过,才进入重试。
8. 第五层:重试策略与请求退避
重试不是无脑循环。合理的重试要覆盖两种场景:
- 接口层面失败:网络错误、超时、限流、HTTP 5xx。
- 内容层面失败:JSON 解析失败、校验失败、修复后仍不合格。
8.1 重试参数设计
重试参数遵循以下原则:
- 指数退避,避免在限流时继续猛打。
- 增加随机抖动,避免多个请求同时重试造成雪崩。
- 设定最大重试次数,不能无限循环。
- 每次重试可以微调 Prompt 或采样参数,比如 temperature 降到 0。
使用tenacity:
from tenacity import ( retry, stop_after_attempt, wait_random_exponential, retry_if_exception_type ) class JSONGenerationError(Exception): pass @retry( stop=stop_after_attempt(3), wait=wait_random_exponential(min=1, max=10), retry=retry_if_exception_type(JSONGenerationError), reraise=True ) def generate_with_retry(prompt: str): # 这里按 HTTP 失败、超时、解析失败分别处理 response = call_model(prompt) try: data = load_json_loose(response) ok, error = validate_json(data) if not ok: raise JSONGenerationError(f"validation failed: {error}") return data except ValueError: raise JSONGenerationError("json parse failed")重试时要记录一 个attempt字段,方便排查。
8.2 基于内容失败的重试
复杂场景下,重试不是简单重发一遍,而是要“带着失败原因重试”:
请求:抽取查询条件 失败原因:缺少 table 字段,filters 是字符串而非数组 重试指令:请补充 table 字段,并将 filters 输出为数组这比“再生成一次”更有效。可以把错误信息作为新消息追加到对话里,让模型知道哪里错了。
def build_retry_messages(original_messages, error_msg): retry_msg = { "role": "user", "content": ( "你上一次输出的 JSON 不符合要求,错误信息:\n" f"{error_msg}\n" "请重新输出,严格满足要求的 JSON 格式。" ) } return original_messages + [retry_msg]8.3 熔断与降级
线上批量任务里,如果连续失败超过阈值,说明模型或 Prompt 出了系统性问题。这时候要熔断,而不是继续重试打空转。
class CircuitBreaker: def __init__(self, fail_threshold: int = 5, timeout: float = 60.0): self.fail_threshold = fail_threshold self.timeout = timeout self.fail_count = 0 self.last_fail_time = None self.open = False def record_failure(self): self.fail_count += 1 self.last_fail_time = time.time() if self.fail_count >= self.fail_threshold: self.open = True def record_success(self): self.fail_count = 0 self.open = False def can_request(self): if not self.open: return True if self.last_fail_time and time.time() - self.last_fail_time > self.timeout: self.open = False self.fail_count = 0 return True return False熔断后可以降级到备用模型、备用 Prompt 模板,或者发到人工队列。
9. 完整方案落地:一个可复用的 Pipeline
把上面所有层封装成一个类,方便在线上服务里直接调用。
import json import logging from datetime import datetime from typing import Any, Dict, List, Optional logger = logging.getLogger(__name__) class JSONStabilityPipeline: """ 大模型稳定 JSON 输出流水线。 使用顺序:协议约束 -> 宽松解析 -> Schema 校验 -> 后处理修复 -> 重试。 """ def __init__( self, model_func, schema: Dict[str, Any], field_aliases: Optional[Dict[str, List[str]]] = None, max_retries: int = 3, enable_circuit_breaker: bool = True, fail_threshold: int = 5, cb_timeout: float = 60.0, ): self.model_func = model_func self.schema = schema self.field_aliases = field_aliases or {} self.max_retries = max_retries self.enable_circuit_breaker = enable_circuit_breaker self.fail_threshold = fail_threshold self.cb_timeout = cb_timeout self.fail_count = 0 self.open_time = None def _can_request(self) -> bool: if not self.enable_circuit_breaker: return True if self.open_time is None: return True if datetime.now().timestamp() - self.open_time > self.cb_timeout: self.open_time = None self.fail_count = 0 return True return False def _record_failure(self): self.fail_count += 1 if self.fail_count >= self.fail_threshold: self.open_time = datetime.now().timestamp() def _record_success(self): self.fail_count = 0 def _normalize_fields(self, data: dict) -> dict: if not self.field_aliases: return data normalized = {} for key, aliases in self.field_aliases.items(): for alias in aliases: if alias in data: normalized[key] = data[alias] break for k, v in data.items(): if k not in normalized: normalized[k] = v return normalized def _repair_types(self, data: dict) -> dict: # 通用类型修复逻辑,可按业务扩展 def cast_int(value, default=None): if isinstance(value, int) and not isinstance(value, bool): return value if isinstance(value, str): try: return int(value.strip()) except ValueError: return default return default for key, value in data.items(): if isinstance(value, str): # 字符串 "true" / "false" 转布尔 if value.strip().lower() == "true": data[key] = True elif value.strip().lower() == "false": data[key] = False return data def _validate(self, data: dict): import jsonschema return jsonschema.validate(instance=data, schema=self.schema) def _parse(self, content: str) -> dict: return load_json_loose(content) def run(self, prompt: str, messages: Optional[List[dict]] = None): for attempt in range(1, self.max_retries + 1): if not self._can_request(): raise JSONGenerationError("circuit breaker open, skip request") try: content = self.model_func(prompt, messages=messages, attempt=attempt) data = self._parse(content) data = self._normalize_fields(data) data = self._repair_types(data) self._validate(data) self._record_success() return data except JSONGenerationError: raise except Exception as e: logger.warning("attempt %s failed: %s", attempt, e) self._record_failure() if attempt == self.max_retries: raise JSONGenerationError(f"max retries exceeded: {e}") raise JSONGenerationError("unreachable")调用示例:
pipeline = JSONStabilityPipeline( model_func=call_model, schema=QUERY_CONDITION_SCHEMA, field_aliases={ "table": ["table", "table_name"], "filters": ["filters", "conditions"], "limit": ["limit", "size"] }, max_retries=3, ) result = pipeline.run("帮我查年龄大于18岁的用户,取前10条") print(result)这套 Pipeline 的好处是:业务代码不需要关心模型到底怎么返回的、要不要重试、要不要修复。只管调用run(),拿到的就是校验过的、结构正确的数据。
10. 性能观察与成本控制
加了很多层的校验、修复、重试之后,最直接的担忧是“会不会变慢、变贵”。这确实是关键,但可控。
10.1 延迟观察
每一层增加的耗时如下:
- Prompt 约束:基本无额外耗时。
- JSON Mode / Function Calling:模型输出阶段可能比纯文本略慢,因为需要按语法约束生成 token。
- 宽松解析:毫秒级,可忽略。
- Schema 校验:毫秒级。
- 后处理修复:毫秒级。
- 重试:重试一次就是一次完整模型调用,这是最大的延迟成本。
所以优化重点放在“减少重试次数”,而不是优化解析函数的微秒级耗时。降低重试次数的核心方法是:优先启用协议级约束(JSON Mode / Function Calling / Structured Outputs),同时维护好失败样本库,针对性补充别名表和修复规则。
10.2 成本控制
成本主要靠控制重试次数和输入 token 长度。
重试时如果要携带上一次错误信息,会导致每次重试的输入 token 变长。可以在第二次重试后尝试换用“更简短的重试指令”,或者只携带错误摘要,不携带完整内容。
另一个可控点是批量任务。批量场景下加上并发限制,比如用asyncio.Semaphore或线程池控制并发数,避免瞬时打满 API 限额:
import asyncio async def run_batch(pipeline, prompts, max_concurrency=5): sem = asyncio.Semaphore(max_concurrency) async def worker(prompt): async with sem: return await asyncio.to_thread(pipeline.run, prompt) tasks = [worker(p) for p in prompts] results = await asyncio.gather(*tasks, return_exceptions=True) return results批量任务要输出结构化日志,至少包含:输入摘要、成功/失败、失败阶段(解析/校验/重试)、耗时、使用 token 数。这样出了问题能快速定位。
10.3 显存与硬件
如果用的是云端 API,基本不用关心显存。如果是本地模型,显存决定你能跑多大尺寸的模型,进而影响 JSON 结构化输出能力。这里不做具体数字断言,建议以实际环境测试为准。
从经验上看,本地部署场景里,如果模型支持 guided decoding 或 JSON Mode,优先开启;如果模型完全不支持,那么 7B 以下的小模型做高度结构化输出,效果通常不稳定,需要更强的后处理和重试兜底。
11. 常见问题与排查方法
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
json.loads报错 “Expecting value” | 模型输出了空内容或纯文本 | 打印原始输出,检查是否有 Markdown 代码块 | 使用宽松解析,先提取 JSON 片段 |
模型输出被```json包裹 | 模型默认偏好代码块 | 检查原始输出首尾 | 在load_json_loose里剥离代码块,或 Prompt 中明确禁止 |
| 字段名不一致 | 模型自由发挥了字段命名 | 对比输出与 Schema 定义 | 建立字段别名映射表,定期从失败样本补充 |
limit被输出成字符串 | 模型类型理解错误 | 打印 JSON 结构 | 写类型修复逻辑,int("10")安全转换,校验层兜底 |
| 多次重试仍然失败 | Prompt 约束不足或模型能力不足 | 查看连续失败的错误信息是否相同 | 切换 JSON Mode / Function Calling / 结构化输出;换更大参数模型;调整重试指令 |
| API 返回 429 | 请求过多触发限流 | 查看响应头Retry-After | 指数退避 + 随机抖动,降低并发 |
| API 返回上下文长度超限 | 重试消息叠加导致 token 增长 | 查看请求体 token 数 | 重试时截断旧消息,只保留最近一轮错误信息 |
| 批量任务中途卡住 | 某个样本触发无限重试或长时间超时 | 检查日志里是否有长时间无响应任务 | 设置单次请求超时,重试次数上限,增加熔断 |
| 本地模型不支持 JSON Mode | 推理框架版本或参数不对 | 查看推理框架文档 | 改用 guided_json / guided_choice,或退回 Prompt + 后处理 |
12. 最佳实践与使用建议
12.1 建立失败样本库
这是最容易被忽略的一点。不要只在意“这次跑通了”,要把每次解析失败、校验失败的输入和原始输出完整保存下来。每隔一段时间,用新样本集回归一次所有失败案例,看看修复规则是否真的覆盖了历史问题。
建议目录结构:
project/ ├── schema/ # JSON Schema 定义 ├── prompts/ # Prompt 模板 ├── samples/ │ ├── success/ # 成功样本 │ └── failures/ # 失败样本(包含原始输出和错误信息) ├── src/ │ ├── parser.py # 宽松解析 │ ├── validator.py # Schema 校验 │ ├── repair.py # 后处理修复 │ ├── retry.py # 重试与熔断 │ └── pipeline.py # 编排 └── logs/ # 结构化日志12.2 先小参数测试
上线前先做一轮小规模验证:
- 准备 20 条覆盖正常、边界、异常输入的数据。
- 先跑协议级约束(JSON Mode / Function Calling)。
- 确认哪些样本失败,再看失败原因属于解析失败还是校验失败。
- 根据失败原因决定是补别名表、补修复函数,还是调整 Prompt。
- 全部通过后,再逐步放大批量。
12.3 接口服务注意访问边界
如果把这套 Pipeline 封装成 HTTP 接口给内部系统调用,建议:
- 按调用方限制 QPS。
- 不要在公网裸奔,内网或带鉴权。
- 请求体大小做限制。
- 记录每个请求的模型名、输入长度、耗时、成功状态。
12.4 敏感性内容合规
发送到模型接口的数据要经过脱敏和授权检查。涉及用户个人信息、客服对话、业务文档时,确认数据使用范围符合条款。批量调用时不要使用包含敏感信息的真实用户数据作为示例放在 Prompt 里,改用脱敏样例。
12.5 保留最小可运行配置
把“最小可运行配置”固定下来,包括:
- 模型名称和版本。
- 采样参数(temperature、top_p)。
- JSON Schema 文件。
- Prompt 模板。
- 重试次数与退避参数。
这套配置放进代码仓库,避免别人接手时靠猜。
13. 总结
这次我们不把“大模型输出不稳定”看作一个 Prompt 调优问题,而是当成一个可靠性工程问题来处理。核心结论是:
- 能走协议级约束(JSON Mode、Function Calling、Structured Outputs)就优先走,不要再靠 Prompt 硬扛。
- 解析阶段要写宽松解析,兼容 Markdown 包裹、多余逗号、文本前后缀。
- 校验阶段必须用 JSON Schema,把字段、类型、必填项全量校验。
- 后处理阶段做字段名归一化、类型修复和默认值补全,但要确保修复动作有业务依据。
- 重试阶段用指数退避加随机抖动,连续失败要触发熔断,重试时携带错误信息比盲目重发更有效。
- 把失败样本库当成长期资产,持续用历史失败案例回归测试。
对于正在线上跑批量任务、Agent 工具调用、数据抽取服务的团队,这套方案能明显降低 JSON 解析失败率。最容易踩的坑是:只加重试、不加校验和后处理,导致连续重试仍然失败;或者只调 Prompt、不开启协议级约束,模型一换版本就飘。
下一步可以做的事:把这套 Pipeline 接入你现有的模型调用 SDK,用历史失败样本先做一轮回归,看看当前方案的解析失败率下降多少。然后再根据失败日志逐步补充修复规则。最后如果你的项目已经落地了这套方案,欢迎在评论区交流实际效果和踩坑记录。