先说一个让我凌晨2点爬起来改代码的事故
我们有个客服工单自动分类服务，用GPT-4o-mini把用户提交的工单内容分类成「退款/技术故障/账号问题/其他」四类，并抽取关键信息。上线第一个版本，prompt长这样：
# 线上v1代码，伪代码示意
def classify_with_llm(content):
prompt = f"""
你是一个客服工单分类器。请对下面的工单内容进行分类，并输出JSON。
工单内容：{content}
输出格式：{"category": "退款", "summary": "用户要求退款", "priority": "high"}
只输出JSON，不要输出其他内容。
"""
resp = openai.chat.completions.create(
model="gpt-4o-mini-2024-07-18",
messages=[{"role": "user", "content": prompt}],
temperature=0.2
)
return json.loads(resp.choices[0].message.content)
上线第一周没出事。第二周某个凌晨，监控告警：json.loads 解析失败率突然飙到38%。拉日志一看，模型输出长这样：
# 实际线上日志摘录
"好的，我来帮你分析这个工单。\n\n```json\n{\n \"category\": \"退款\",\n \"summary\": \"用户要求退款\",\n \"priority\": \"high\"\n}\n```\n\n以上就是我的分析结果。"
没加markdown围栏的，有的在JSON前面加「结果如下：」，有的把注释写进JSON里，有的key加了引号有的没加。一个json.loads全炸。
这篇文章就是从那晚之后，我花了三天时间把「让LLM输出稳定JSON」这件事彻底搞清楚后写出来的。数据全部来自真实压测。
三个方案，跑同一批数据
先交代测试环境，省得你们跑出来的数据和我对不上：
| 项目 | 配置 |
|---|---|
| 模型 | gpt-4o-mini-2024-07-18 |
| OpenAI SDK | openai 1.35.15 |
| Python | 3.11.7 |
| 压测工具 | ApacheBench 2.3 |
| 测试样本 | 200条真实历史工单（脱敏后） |
| 并发 | 20并发 × 10轮 |
| 温度 | 0.2（所有方案统一） |
任务任务固定：给一段工单文本（80~300字不等），输出以下结构化字段：
category：枚举值，退款/技术故障/账号问题/其他summary：字符串，一句话摘要priority：枚举值，low/medium/high
下面三个方案跑下来，结果差异非常大。
方案一：零样本直出（基线）
就是我开头放的那版代码。用我们线上跑了两周的真实数据做统计，不说理论，直接上数据：
| 指标 | 数值 |
|---|---|
| JSON合法率（json.loads可直接解析） | 61% |
| 平均响应耗时 | 4.2s（含重试） |
| 平均重试次数 | 0.39次/请求 |
| 平均token用量 | 1264 tokens/请求 |
| 输出包含markdown围栏占比 | 23% |
| 输出带前后缀解释文本占比 | 19% |
token为什么这么高？因为解析失败后我们做了「把错误喂回给模型让它修正」的重试逻辑，一来一回token翻倍。
零样本方案的问题本质：LLM的tokenizer和采样逻辑不保证输出结构的完整性。你说「输出JSON」，模型认为的JSON和json.loads要求的JSON不是同一个东西。「JSON」这个概念在训练数据里包含了markdown围栏、解释性文本、注释等大量噪声。
方案二：嵌套示例（Few-Shot）
很直觉的改进：给模型看2~3个完整的输入输出例子。代码长这样：
from openai import OpenAI
import json
client = OpenAI()
FEW_SHOT_EXAMPLES = [
{
"content": "我2月3号买了一个蓝牙耳机，用了不到一周右耳朵就没声音了。订单号是ORD-2024-0231，希望能给我换一个或者退款。",
"output": {
"category": "技术故障",
"summary": "蓝牙耳机右耳无声，要求换货或退款",
"priority": "medium"
}
},
{
"content": "我的账号突然登不上了，说是什么密码错误，我确定密码没记错。手机号是138****8877，请帮我查一下。",
"output": {
"category": "账号问题",
"summary": "账号无法登录，密码确认无误",
"priority": "high"
}
},
{
"content": "你们能不能把运费退给我？我收到货发现尺寸不对，退货运费花了18块钱，这个应该你们出吧？",
"output": {
"category": "退款",
"summary": "退货产生的运费应由商家承担",
"priority": "low"
}
}
]
def classify_few_shot(content: str) -> dict:
messages = [
{"role": "system", "content": "你是一个客服工单分类器。请参考示例，对用户工单进行分类，输出JSON。"}
]
for ex in FEW_SHOT_EXAMPLES:
messages.append({"role": "user", "content": f"工单：{ex['content']}"})
messages.append({"role": "assistant", "content": json.dumps(ex["output"], ensure_ascii=False)})
messages.append({"role": "user", "content": f"工单：{content}"})
resp = client.chat.completions.create(
model="gpt-4o-mini-2024-07-18",
messages=messages,
temperature=0.2,
)
text = resp.choices[0].message.content
return json.loads(text.strip())
跑同一批200条数据：
| 指标 | 数值 |
|---|---|
| JSON合法率 | 84% |
| 平均响应耗时 | 1.8s（含重试） |
| 平均重试次数 | 0.16次/请求 |
| 平均token用量 | 897 tokens/请求 |
| 输出包含markdown围栏占比 | 8% |
合法率提升了23个百分点，token降了29%。但还没有达到生产可用的线。真实流量下20%的解析失败率，依然要写一堆纠错逻辑。
更关键的问题是：补一个例子，token就会增加大约150~200。这会导致KV-Cache命中率下降——因为每个请求的prompt不一样，prompt越长，KV-Cache能缓存的部分就越少，每次请求都要重新跑前缀的prefill，直接推高首字延迟。
我们把方案二的prompt平均长度和KV-Cache命中率做了个统计：输出提示词平均1128 tokens，KV-Cache命中的输入token比例只有12%。方案一比它更低，只有约5%的命中率。
方案三：Few-Shot × JSON Schema约束
核心思路是两条腿走路：
- 用
response_format参数把模型输出限制在合法JSON结构内 - 用Few-Shot示例保证内容质量稳定
OpenAI的API从 gpt-4o-mini-2024-07-18 开始支持 response_format: {"type": "json_schema"}，它是用约束解码技术（constrained decoding）实现的——在采样阶段就屏蔽掉不合法JSON的词元。这是结构性保证，不是prompt里写几个字能比的。
先定义JSON Schema：
{
"name": "ticket_classification",
"strict": true,
"schema": {
"type": "object",
"properties": {
"category": {
"type": "string",
"enum": ["退款", "技术故障", "账号问题", "其他"]
},
"summary": {
"type": "string",
"description": "一句话概括用户的问题，不超过30个字"
},
"priority": {
"type": "string",
"enum": ["low", "medium", "high"]
}
},
"required": ["category", "summary", "priority"],
"additionalProperties": false
}
}
注意：strict: true 意味着所有字段都必须出现在 required 里，且禁止 additionalProperties。同时拒绝包含anyOf/oneOf这类混合类型——因为这个模型版本的约束解码只支持非空的object类型，不支持嵌套的派生类型限制。
完整实现代码：
from openai import OpenAI
import json
client = OpenAI()
TICKET_SCHEMA = {
"name": "ticket_classification",
"strict": True,
"schema": {
"type": "object",
"properties": {
"category": {
"type": "string",
"enum": ["退款", "技术故障", "账号问题", "其他"]
},
"summary": {
"type": "string",
"description": "一句话概括用户的问题，不超过30个字"
},
"priority": {
"type": "string",
"enum": ["low", "medium", "high"]
}
},
"required": ["category", "summary", "priority"],
"additionalProperties": False
}
}
# 压缩版few-shot示例：不需要太多，2个足够
FEW_SHOT_COMPACT = [
{
"user": "耳机右耳无声，订单ORD-2024-0231，要求换货或退款",
"assistant": {"category": "技术故障", "summary": "蓝牙耳机右耳无声，要求换货或退款", "priority": "medium"}
},
{
"user": "账号登不上，密码确认无误，手机号138****8877",
"assistant": {"category": "账号问题", "summary": "账号无法登录，密码确认无误", "priority": "high"}
}
]
def classify_structured(content: str, max_retries: int = 2) -> dict:
messages = [
{"role": "system", "content": "你是客服工单分类器。"},
]
for ex in FEW_SHOT_COMPACT:
messages.append({"role": "user", "content": f"工单：{ex['user']}"})
messages.append({"role": "assistant", "content": json.dumps(ex["assistant"], ensure_ascii=False)})
messages.append({"role": "user", "content": f"工单：{content}"})
for attempt in range(max_retries):
try:
resp = client.chat.completions.create(
model="gpt-4o-mini-2024-07-18",
messages=messages,
temperature=0.2,
response_format={"type": "json_schema", "json_schema": TICKET_SCHEMA}
)
text = resp.choices[0].message.content
data = json.loads(text) # 正常情况下不会失败
# 附加校验：枚举值和必填字段
if data.get("category") not in ["退款", "技术故障", "账号问题", "其他"]:
raise ValueError(f"invalid category: {data.get('category')}")
if not data.get("summary", "").strip():
raise ValueError("summary is empty")
return data
except Exception as e:
if attempt == max_retries - 1:
raise
print(f"retry {attempt + 1} after error: {e}")
raise RuntimeError("unreachable")
跑同一批200条样本：
| 指标 | 数值 |
|---|---|
| JSON合法率 | 97% |
| 平均响应耗时 | 0.3s（不含重试） |
| 平均重试次数 | 0.03次/请求 |
| 平均token用量 | 382 tokens/请求 |
| KV-Cache命中率 | 89% |
剩余3%的失败集中在summary字段超长或包含极端字符（换行、emoji）的情况，不是JSON解析失败，是业务校验失败。
三个方案横向对比
| 合法率 | 平均耗时 | 平均token | KV-Cache命中 | 重试率 | |
|---|---|---|---|---|---|
| 零样本直出 | 61% | 4.2s | 1264 | 5% | 39% |
| 嵌套示例 | 84% | 1.8s | 897 | 12% | 16% |
| Few-Shot × Schema | 97% | 0.3s | 382 | 89% | 3% |
方案三的耗时优势是叠加出来的：合法率高→不需要重试→prompt短→KV-Cache命中率超高→不用重复prefill。三个因素互相强化。
为什么Schema约束能省这么多token和耗时
关键在于response_format=json_schema改的是采样逻辑，不是prompt逻辑。
具体来说，OpenAI在服务端做了约束解码（constrained decoding）：给模型加了一个token-level的过滤器，凡是会导致JSON语法非法的token直接禁止采样。模型每一步只能从「合法延续」集合里挑选token，所以输出一定是合法JSON。模型不需要在自己的输出里考虑「要不要加markdown围栏」「key要不要加引号」这些事。
这带来两个直接影响：
- 输出长度大幅缩短。跑完200条统计，方案三平均输出78 tokens，方案一平均输出156 tokens，方案二平均输出121 tokens。输出短了，TTFT没变但TPOT时间少了。
- 重试逻辑基本可以删掉。方案二因为有16%的重试，每个失败请求平均多花3~5秒；方案三的重试率降到3%，且几乎都是业务校验失败，整个调用链路的P99延迟降了60%以上。
另外我们发现一个有意思的数据：方案三里那些让模型从schema推导出逻辑的请求，响应长度其实和方案的性能也有关。如果把schema写得很啰嗦（比如description写一大段），模型想「按照schema输出」的意图会变弱，输出内容会偏长——token和时间反而增加。Schema应该短，约束应该靠参数结构本身，不靠描述文字。
生产级改造：加个守护解析器
方案三还挂了一个「守卫层」：即使response_format保证了JSON合法，内容里也可能出现业务层不可用的数据（category枚举不匹配、summary空白、多出了不可见字符）。我写了一个独立的解析器，不依赖OpenAI SDK：
# guard_parser.py
import json
import re
from typing import Any
VALID_CATEGORIES = {"退款", "技术故障", "账号问题", "其他"}
VALID_PRIORITIES = {"low", "medium", "high"}
def clean_model_output(text: str) -> str:
"""清洗模型输出，去掉可能的噪声字符。"""
text = text.strip()
# 去除包裹的markdown围栏（如果有，兼容backup场景）
fence_pattern = re.compile(r"^```(?:json)?\s*(.*?)\s*```$", re.DOTALL)
m = fence_pattern.match(text)
if m:
text = m.group(1).strip()
# 如果response_format没生效，可能有前后缀文字，找到第一个{和最后一个}
start = text.find("{")
end = text.rfind("}")
if start != -1 and end != -1 and end > start:
text = text[start:end + 1]
return text
def parse_ticket_output(text: str) -> dict[str, Any]:
cleaned = clean_model_output(text)
try:
data = json.loads(cleaned)
except json.JSONDecodeError:
# 最后兜底：把常见的单引号/注释等情况修掉
cleaned = re.sub(r"//.*", "", cleaned)
cleaned = cleaned.replace("'", '"')
data = json.loads(cleaned)
if not isinstance(data, dict):
raise ValueError(f"output is not an object: {type(data)}")
if data.get("category") not in VALID_CATEGORIES:
raise ValueError(f"invalid category: {data.get('category')}")
if data.get("priority") not in VALID_PRIORITIES:
raise ValueError(f"invalid priority: {data.get('priority')}")
if not isinstance(data.get("summary"), str) or len(data["summary"].strip()) == 0:
raise ValueError("summary is empty or not a string")
# 截断超长summary，保证下游数据库字段长度
data["summary"] = data["summary"].strip()[:100]
return data
这个解析器不做重试，只做隔离。重试逻辑在上层重试，比如用 tenacity 或者直接在一个 while 循环里重试。解析器只负责把输出变成可靠的结构化数据。
压测：20并发下的性能表现
生产环境不能只看单次延迟，得看并发表现。用ab压了10轮，每轮20并发：
ab -n 2000 -c 20 -T application/json -p test_payload.json https://your-api.example.com/classify
压测结果：
| 指标 | 方案一（基线） | 方案三（Schema约束） |
|---|---|---|
| QPS | 8 | 31 |
| 平均延迟 | 2417ms | 226ms |
| P95延迟 | 4930ms | 413ms |
| P99延迟 | 7914ms | 682ms |
| 错误率 | 14.2% | 0.3% |
QPS翻了接近4倍。这还是在加了业务校验和日志中间件的情况下跑的。延迟主要省在prefill上——方案一每个请求要处理超过1000 token的prompt，方案三只有382个token，加上KV-Cache命中率高，首token延迟降了一半以上。
成本侧也顺带算了笔账：方案三单request费用约 $0.00015（按gpt-4o-mini的定价 $0.15/百万输入 + $0.60/百万输出计算），方案一约 $0.00047。按每天10000次调用算，一个月能省大约 $96。
上线时的接入配置
我们用的是Gin框架，接入方式很轻，只加了一个中间件：
# config.yaml
llm:
model: gpt-4o-mini-2024-07-18
temperature: 0.2
max_retries: 2
response_format:
type: json_schema
json_schema:
name: ticket_classification
strict: true
schema:
type: object
properties:
category:
type: string
enum: ["退款", "技术故障", "账号问题", "其他"]
summary:
type: string
description: "一句话概括用户的问题，不超过30个字"
priority:
type: string
enum: ["low", "medium", "high"]
required: [category, summary, priority]
additionalProperties: false
server:
port: 8080
read_timeout: 5s
write_timeout: 10s
如果你不是用OpenAI官方SDK，而是走兼容网关（比如OneAPI、LiteLLM），需要注意：很多网关只透传 response_format: {"type": "json_object"}，不认 json_schema。这种情况可以退一步用 json_object 模式加一个strict提示词，合法率实测大约90%左右，比不设强但弱于 json_schema。上线前一定要做一次Schema感知测试。
避坑指南（每一条都是真金白银换来的）
坑1：schema写得太复杂，模型直接拒绝输出
最开始我们用了一个非常完整的JSON Schema——所有字段都标了 required，summary字段加 minLength 和 maxLength，还有 pattern 正则要求不能包含特殊字符。结果模型在遇到含换行符或emoji的工单时，输出直接变成 {"error": "..."} 或者干脆返回空内容。
原因是约束解码在遇到「不可能满足」的约束时，采样空间会退化，模型被迫输出低概率token，整个输出质量崩塌。解决方法：约束只用来保证结构合法，不做内容级校验。内容级校验放到解析器里。summary长度超了就在解析器截断，不要放进schema。
坑2：正则解析不覆盖所有情况
我们在方案一和方案二阶段写过一个 extract_json 函数，期望用正则从模型输出里抠出JSON。实际遇到的问题：有的输出里JSON前后有解释性的中文，正则匹配到 } 之前的部分包含额外文本；有的模型把JSON拆成多行并加了注释，正则直接崩。最后这个函数被我们彻底删了——靠正则兜底本身就是一种错误的防御姿势。正确做法是把约束前置到结构保证层面（response_format），解析器只做清洗和校验。
坑3：temperature=0≠确定性
调优阶段我们把temperature设成了0，以为输出会完全确定。实际上OpenAI的GPU推理存在非确定性（浮点累加顺序、不同batchsize下的kernel选择都会影响结果）。我们用同一批200条样本跑了两轮，temperature=0的条件下，有7条输出的summary措辞不一样。所以不要在生产环境做「输出完全一致」的假设。需要幂等的话，正确的姿势是在业务层加缓存，key可以设为工单内容的hash，output存数据库，下次直接读。
坑4：模型版本不支持json_schema参数，静默降级
有段时间我们在测试环境用的 gpt-3.5-turbo-1106，文档写的是支持 response_format 的，但传 {"type": "json_schema", ...} 时报错 BadRequestError: Unsupported response format type。OpenAI的版本差异在这里很隐蔽：json_object 是大多数模型都支持的，json_schema 是从 gpt-4o-mini-2024-07-18 才开始对特定模型开放。如果你的服务要兼容多个模型，必须做一个模型能力探测，把 json_schema 降级到 json_object，而不是直接报错。
坑5：KV-Cache命中率被prompt长度悄悄毁掉
方案二里我们用了完整的500字工单示例，每轮请求都要重新处理那1000多字的系统prompt。后来查网关日志发现，同一个用户连续请求时，系统prompt完全没有命中缓存，因为我们在每个请求里用 f-string 动态拼接了用户ID之类的内容，导致前面部分也变了。架构上应该把静态前缀和动态部分隔离开，让 messages 数组的system prompt部分保持完全一致。这个改动直接把我们的成本降低了约40%。
坑6：schema里的enum和你的业务枚举不对齐
我们试过把 category 的枚举定义为 ["refund", "tech", "account", "other"] 这样的英文，输出合法率97%但业务代码不认识，还得做映射。后来直接把枚举值改成了中文业务词，省掉了一层转换。这个不疼，但提醒一点：schema里的枚举就是你的API契约，它最好直接等于业务层的枚举，不要引入第二套表示法。
总结一下这个方案的可复制路径
如果你也在做LLM结构化输出，直接抄作业：
- 能用
response_format: json_schema就用，不要纠结prompt怎么写。 - schema定义得短，只放结构约束，不放内容校验。
- Few-Shot例子保留1~3个，足够稳定，多了只会增加token。
- 解析器做成独立模块，负责清洗、校验、截断。
- 重试逻辑放上层，控制在2次以内，超过就fail fast，不要死循环。
这套方案上线后，我们工单分类服务的解析失败率从38%降到了1.2%，平均延迟从4.2s降到了0.3s，每个月API成本降了接近40%。如果这篇文章能帮你少走一点弯路，那这力气就没白费。