第8章:结构化输出与 JSON 结果约束
1. 项目背景
某银行的风控系统需要从客服通话转写文本中提取结构化信息:客户姓名、身份证号、投诉类型、涉及金额。技术团队使用LLM做信息抽取,在测试集上准确率高达95%,但上线后实际业务中只有72%的请求返回了合法JSON——剩下的28%中,有的输出了"好的,根据您的描述,我提取了以下信息:{…}"这样的前缀文本,有的JSON缺少闭合括号,有的直接把整个分析过程(CoT)写了出来。
业务方的下游系统是严格的JSON解析器——任何非JSON输出都导致整个风控流程中断,需要人工介入。技术团队被迫在模型输出后加了一层"正则清洗+JSON修复"的补丁逻辑,但两个月后发现这个补丁越写越复杂,已经到了难以维护的程度。
痛点:LLM天然是文本生成器,不是结构化数据生成器。temperature=0可以增加确定性,但不能保证JSON格式。问题的本质在于:标准的Chat Completion API中,模型输出是一个自由文本流,JSON只是文本的一种可能形式。要从根本上解决这个问题,需要在采样层面就施加格式约束,而不是事后修补。
vLLM提供了多种结构化输出方案:response_format(OpenAI兼容的JSON模式)、guided_json(JSON Schema约束)、guided_regex(正则约束)、guided_choice(选项约束)。本章将从业务驱动的信息抽取场景出发,逐步实现一个高鲁棒性的结构化输出服务。
2. 项目设计
(场景:早会白板前,小胖正在展示上周的"JSON修复代码量增长曲线"——一条陡峭的上坡线。)
小胖:“大师你看,我们的JSON修复补丁从最初的50行涨到了现在800行!上周加了’支持修复嵌套花括号’和’自动补全缺少的引号’两个功能后,代码已经没人看得懂了。”
大师:“这说明你们一直在下游解决问题,而不是从源头解决。LLM输出JSON不稳定,根因是采样过程没有受到格式约束。想象一下,你让一个厨师做麻婆豆腐,但只告诉他要’麻辣鲜香’——厨师可能做出来的东西能吃,但每次的配菜、火候、摆盘都可能不同。structured_outputs相当于给了厨师一张精确的菜谱——每一步必须放什么,少了一步就不行。”
小白:“vLLM具体是怎么实现格式约束的?是事后校验还是真的在生成时限制了Token?”
大师:“有两层机制。第一层是采样约束(Guided Decoding)——在每个Token采样前,根据目标Schema计算出当前状态下合法的下一个Token集合,排除所有不合法的Token。比如正在生成JSON的键{"name"——下一个Token必须是":" ,不能是别的。这是从源头保证格式正确。”
小白:“那第二层呢?”
大师:“第二层是结构提示注入——vLLM在system prompt后面自动追加格式说明,或者在调用apply_chat_template时将格式要求嵌入Prompt。这层的效果取决于模型对格式指令的理解能力。两层配合,比单靠任何一层都可靠。”
小胖:“那如果我就是想要模型一句话总结+JSON字段这种混合输出呢?JSON模式会不会阻止它输出前面的自由文本?”
大师:“这正是guided_json的局限——它会强制整个输出必须是合法JSON。如果你的业务确实需要’自由文本 + JSON’的混合输出,可以用guided_regex定义一个匹配你预期格式的正则,或者把JSON字段嵌入到一个更大的文本模板中,让它以’包裹格式’输出。但这会降低JSON的成功率——因为自由文本部分不受约束。”
小白:“vLLM有哪几种具体的约束方式?分别适合什么场景?”
大师:(在白板上列了一个表)
| 约束方式 | 参数 | 适用场景 | 可靠性 |
|---|---|---|---|
| JSON模式 | response_format={"type": "json_object"} |
简单JSON,模型支持 | 中等 |
| JSON Schema | guided_json=json_schema_string |
复杂结构,字段有类型约束 | 高 |
| 正则约束 | guided_regex=regex_pattern |
自定义格式,非JSON | 高 |
| 选项约束 | guided_choice=["A","B","C"] |
分类、路由决策 | 极高 |
| 无约束(兜底) | 事后JSON修复 | 任何场景 | 低 |
小胖:“那个guided_choice听起来最靠谱!是不是跟多选题一样?”
大师:“对,本质上就是限制模型的输出必须是指定选项之一。比如客服意图分类——['退货', '换货', '投诉', '咨询']——模型只能输出这四个词中的一个,连多余的空格都不会有。这种场景的可靠性接近100%。”
小白:“JSON Schema约束和正则约束的性能有差异吗?”
大师:“约束越严格,采样越慢——因为每个Token生成前都要计算合法Token集合。但差异很小(通常<5%),远小于格式失败带来的重试和修复成本。格式约束是一次性投入,换取长期的准确性。”
小胖:“那失败兜底怎么做?就算99%成功率,每天10万次调用也有1000次失败。”
大师:"兜底策略分四层:
- 采样层兜底:如果guided模式无法应用(模型不支持),退化为temperature=0的普通采样
- 解析层兜底:尝试正则提取JSON、修复常见错误(缺引号、单引号、尾部逗号)
- 校验层兜底:用JSON Schema验证结构完整性,缺失字段填默认值
- 业务层兜底:标记该条为’需人工处理’,返回错误码给上游,不中断主流程
永远不要让格式问题变成业务中断。"
3. 项目实战
3.1 环境准备
依赖:
- vLLM OpenAI兼容服务已启动
- 支持
response_format/guided_json的模型(推荐Qwen2.5-7B-Instruct) - OpenAI Python SDK
# 启动vLLM服务
vllm serve ./models/Qwen2.5-7B-Instruct --host 0.0.0.0 --port 8000 \
--gpu-memory-utilization 0.85 --max-model-len 4096
3.2 分步实现
步骤1:实现四种结构化输出的对比测试
目标:对同一批信息抽取需求,分别用无约束、JSON模式、JSON Schema、正则约束四种方式对比成功率。
# 保存为 structured_output_compare.py
# 四种结构化输出策略的对比实验
import json
import time
from openai import OpenAI
CLIENT = OpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")
MODEL = "Qwen2.5-7B-Instruct"
# 测试数据:工单信息抽取
TEST_CASES = [
"我叫张三,手机号13800138000,订单号ORD-2024-8891,申请退货,原因是屏幕有坏点。",
"李四,电话13912345678,工单TK-001,投诉物流太慢,包裹等了5天还没到。",
"王五 15000001111 订单 ORD-001 换货 颜色发错了。",
"用户赵六反映说买的耳机左耳没声音,联系电话是13600002222,单号TK-2025-0003",
"客户投诉:孙七,18700003333,订单ORD-5566,收到的商品与图片不符,要求退款。",
]
# JSON Schema定义(字段和类型约束)
EXTRACTION_SCHEMA = {
"type": "object",
"properties": {
"customer_name": {
"type": "string",
"description": "客户姓名"
},
"phone": {
"type": "string",
"description": "联系电话"
},
"order_id": {
"type": "string",
"description": "订单号或工单号"
},
"action_type": {
"type": "string",
"enum": ["退货", "换货", "投诉", "咨询", "退款"],
"description": "客户诉求类型"
},
"reason": {
"type": "string",
"description": "具体原因描述"
},
},
"required": ["customer_name", "phone", "order_id", "action_type", "reason"],
}
SYSTEM_PROMPT = """你是一个专业的信息抽取助手。从用户输入中提取关键信息,严格按JSON格式输出。
不要输出任何JSON之外的内容,不要加解释、不要加markdown代码块标记。"""
# ===== 策略1:无约束(普通Chat)=====
def extract_baseline(user_input: str) -> dict:
"""基准策略:普通Chat,temperature=0"""
response = CLIENT.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_input},
],
temperature=0,
max_tokens=300,
)
return {
"raw": response.choices[0].message.content,
"strategy": "baseline",
}
# ===== 策略2:JSON模式(response_format)=====
def extract_json_mode(user_input: str) -> dict:
"""OpenAI兼容的JSON模式"""
try:
response = CLIENT.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_input},
],
temperature=0,
max_tokens=300,
response_format={"type": "json_object"},
)
return {
"raw": response.choices[0].message.content,
"strategy": "json_mode",
}
except Exception as e:
return {"raw": "", "strategy": "json_mode", "error": str(e)}
# ===== 策略3:JSON Schema(guided_json)=====
def extract_guided_json(user_input: str) -> dict:
"""JSON Schema约束,通过extra_body传递"""
try:
response = CLIENT.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_input},
],
temperature=0,
max_tokens=300,
extra_body={
"guided_json": json.dumps(EXTRACTION_SCHEMA),
},
)
return {
"raw": response.choices[0].message.content,
"strategy": "guided_json",
}
except Exception as e:
return {"raw": "", "strategy": "guided_json", "error": str(e)}
# ===== 策略4:正则约束(guided_regex)=====
def extract_guided_regex(user_input: str) -> dict:
"""用正则约束输出格式"""
# 构建一个正则,匹配预期的输出模式
# 这里做一个简化:要求字段按特定顺序排列
import re
try:
response = CLIENT.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_input},
],
temperature=0,
max_tokens=300,
extra_body={
"guided_regex": (
r'\{\s*"customer_name":\s*"[^"]*",\s*'
r'"phone":\s*"[^"]*",\s*'
r'"order_id":\s*"[^"]*",\s*'
r'"action_type":\s*"(?:退货|换货|投诉|咨询|退款)",\s*'
r'"reason":\s*"[^"]*"\s*\}'
),
},
)
return {
"raw": response.choices[0].message.content,
"strategy": "guided_regex",
}
except Exception as e:
return {"raw": "", "strategy": "guided_regex", "error": str(e)}
# ===== JSON解析和验证 =====
def parse_and_validate(raw: str) -> dict:
"""尝试解析JSON,包含容错处理"""
import re
result = {"valid_json": False, "fields_complete": False, "parsed": None}
if not raw:
return result
content = raw.strip()
# 尝试1:直接解析
for attempt in [content]:
try:
parsed = json.loads(attempt)
result["valid_json"] = True
result["parsed"] = parsed
break
except json.JSONDecodeError:
pass
# 尝试2:提取 {...} 部分
if not result["valid_json"]:
match = re.search(r'\{[^{}]*"customer_name"[^{}]*\}', content, re.DOTALL)
if match:
try:
parsed = json.loads(match.group())
result["valid_json"] = True
result["parsed"] = parsed
except json.JSONDecodeError:
pass
# 尝试3:修复常见的JSON问题后重试
if not result["valid_json"]:
fixed = content
fixed = re.sub(r"'", '"', fixed) # 单引号→双引号
fixed = re.sub(r',\s*}', '}', fixed) # 尾部逗号
try:
parsed = json.loads(fixed)
result["valid_json"] = True
result["parsed"] = parsed
except json.JSONDecodeError:
pass
# 验证字段完整性
if result["valid_json"]:
required = ["customer_name", "phone", "order_id", "action_type", "reason"]
result["fields_complete"] = all(
k in result["parsed"] and result["parsed"][k] for k in required
)
return result
# ===== 主对比实验 =====
def run_compare():
strategies = {
"baseline (无约束)": extract_baseline,
"json_mode (JSON模式)": extract_json_mode,
"guided_json (Schema)": extract_guided_json,
"guided_regex (正则)": extract_guided_regex,
}
results = {name: [] for name in strategies}
for case_idx, case_text in enumerate(TEST_CASES):
print(f"\n{'='*60}")
print(f"测试案例 {case_idx + 1}: {case_text[:60]}...")
for strategy_name, strategy_fn in strategies.items():
print(f"\n [{strategy_name}]")
output = strategy_fn(case_text)
parse_result = parse_and_validate(output.get("raw", ""))
status = "OK" if parse_result["valid_json"] else "FAIL"
fields = "complete" if parse_result["fields_complete"] else "incomplete"
print(f" JSON合法: {status}, 字段: {fields}")
if parse_result["valid_json"]:
print(f" 解析结果: {json.dumps(parse_result['parsed'], ensure_ascii=False)}")
else:
raw_preview = output.get("raw", "")[:100].replace("\n", " ")
print(f" 原始输出: {raw_preview}")
results[strategy_name].append({
"case_idx": case_idx,
"valid_json": parse_result["valid_json"],
"fields_complete": parse_result["fields_complete"],
})
time.sleep(0.3)
# 汇总统计
print(f"\n{'='*70}")
print("四种策略对比汇总")
print(f"{'='*70}")
print(f"{'策略':<25} {'JSON合法率':<15} {'字段完整率':<15} {'综合得分'}")
print(f"{'-'*70}")
for strategy_name in strategies:
items = results[strategy_name]
total = len(items)
valid = sum(1 for i in items if i["valid_json"])
complete = sum(1 for i in items if i["fields_complete"])
score = (valid + complete) / (2 * total) * 100
print(f"{strategy_name:<25} {valid}/{total} ({valid/total*100:.0f}%) "
f"{complete}/{total} ({complete/total*100:.0f}%) "
f"{score:.0f}%")
if __name__ == "__main__":
run_compare()
运行结果示例:
======================================================================
四种策略对比汇总
======================================================================
策略 JSON合法率 字段完整率 综合得分
----------------------------------------------------------------------
baseline (无约束) 3/5 (60%) 2/5 (40%) 50%
json_mode (JSON模式) 4/5 (80%) 3/5 (60%) 70%
guided_json (Schema) 5/5 (100%) 5/5 (100%) 100%
guided_regex (正则) 5/5 (100%) 4/5 (80%) 90%
步骤2:实现生产级结构化抽取服务(含兜底)
目标:封装一个健壮的信息抽取服务,包含多层兜底和业务降级。
# 保存为 extraction_service.py
# 生产级结构化信息抽取服务,含多层兜底
import json
import re
import logging
from typing import Optional, Dict, Any, List
from openai import OpenAI
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class StructuredExtractor:
"""结构化信息抽取服务"""
def __init__(self, client: OpenAI, model: str, schema: dict):
self.client = client
self.model = model
self.schema = schema
self.schema_json = json.dumps(schema)
def extract(self, text: str, strategy: str = "guided_json") -> Dict[str, Any]:
"""
主抽取方法
strategy: "guided_json" | "guided_regex" | "json_mode" | "baseline"
返回: {"success": bool, "data": dict, "error": str, "strategy_used": str}
"""
# 第1层:尝试使用指定策略抽取
raw, actual_strategy = self._try_extract(text, strategy)
# 第2层:解析JSON
parsed = self._parse_json(raw)
if parsed["valid_json"] and parsed["fields_complete"]:
return {
"success": True,
"data": parsed["parsed"],
"strategy_used": actual_strategy,
"raw": raw,
}
# 第3层:降级策略——如果第1层失败,尝试其他策略
logger.warning(f"策略{actual_strategy}失败,尝试降级策略")
fallback_strategies = [s for s in ["guided_regex", "json_mode", "baseline"]
if s != actual_strategy]
for fb_strategy in fallback_strategies:
raw, _ = self._try_extract(text, fb_strategy)
parsed = self._parse_json(raw)
if parsed["valid_json"] and parsed["fields_complete"]:
return {
"success": True,
"data": parsed["parsed"],
"strategy_used": f"{actual_strategy}->{fb_strategy}(降级)",
"raw": raw,
}
# 第4层:字段级降级——缺失字段填充默认值
if parsed["valid_json"] and not parsed["fields_complete"]:
data = self._fill_defaults(parsed["parsed"])
return {
"success": True,
"data": data,
"strategy_used": f"{actual_strategy}(字段降级)",
"warnings": ["部分字段使用默认值"],
"raw": raw,
}
# 第5层:彻底失败,返回错误标记
return {
"success": False,
"data": None,
"error": "所有策略均失败,无法提取结构化信息",
"raw": raw,
"strategy_used": actual_strategy,
}
def _try_extract(self, text: str, strategy: str) -> tuple:
"""执行具体的抽取策略"""
system_prompt = (
"你是一个信息提取助手。从用户输入中提取关键信息,"
"严格遵守指定的输出格式。不要输出任何非格式内容。"
)
common_params = {
"model": self.model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": text},
],
"temperature": 0,
"max_tokens": 400,
}
try:
if strategy == "guided_json":
response = self.client.chat.completions.create(
**common_params,
extra_body={"guided_json": self.schema_json},
)
elif strategy == "json_mode":
response = self.client.chat.completions.create(
**common_params,
response_format={"type": "json_object"},
)
elif strategy == "baseline":
# 在system prompt中追加格式说明
common_params["messages"][0]["content"] += (
f"\n\n输出格式(JSON):{self.schema_json}"
)
response = self.client.chat.completions.create(**common_params)
elif strategy == "guided_regex":
response = self.client.chat.completions.create(
**common_params,
extra_body={"guided_regex": self._build_regex()},
)
else:
raise ValueError(f"Unknown strategy: {strategy}")
return response.choices[0].message.content, strategy
except Exception as e:
logger.error(f"策略 {strategy} 执行异常: {e}")
return "", strategy
def _build_regex(self) -> str:
"""根据Schema自动构建正则(简化版)"""
return (
r'\{\s*"customer_name":\s*"[^"]*",\s*'
r'"phone":\s*"[^"]*",\s*'
r'"order_id":\s*"[^"]*",\s*'
r'"action_type":\s*"(?:退货|换货|投诉|咨询|退款)",\s*'
r'"reason":\s*"[^"]*"\s*\}'
)
def _parse_json(self, raw: str) -> dict:
"""JSON解析 + 容错修复"""
if not raw:
return {"valid_json": False, "fields_complete": False, "parsed": None}
content = raw.strip()
# 移除markdown代码块标记
content = re.sub(r'^```(?:json)?\s*', '', content)
content = re.sub(r'\s*```$', '', content)
for attempt in [content]:
try:
parsed = json.loads(attempt)
return self._validate(parsed)
except json.JSONDecodeError:
pass
# 正则提取
match = re.search(r'\{[^{}]*\}', content, re.DOTALL)
if match:
try:
parsed = json.loads(match.group())
return self._validate(parsed)
except json.JSONDecodeError:
pass
# 修复单引号、尾部逗号后重试
fixed = re.sub(r"'", '"', content)
fixed = re.sub(r',\s*\}', '}', fixed)
try:
parsed = json.loads(fixed)
return self._validate(parsed)
except json.JSONDecodeError:
pass
return {"valid_json": False, "fields_complete": False, "parsed": None}
def _validate(self, parsed: dict) -> dict:
"""验证必填字段和类型"""
required = self.schema.get("required", [])
fields_ok = True
for field in required:
if field not in parsed or not parsed[field]:
fields_ok = False
break
return {
"valid_json": True,
"fields_complete": fields_ok,
"parsed": parsed,
}
def _fill_defaults(self, data: dict) -> dict:
"""缺失字段填充默认值"""
defaults = {
"customer_name": "未知",
"phone": "",
"order_id": "",
"action_type": "咨询",
"reason": "未说明",
}
for key, default in defaults.items():
if key not in data or not data[key]:
data[key] = default
return data
# ===== 使用示例 =====
if __name__ == "__main__":
client = OpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")
extractor = StructuredExtractor(
client=client,
model="Qwen2.5-7B-Instruct",
schema=EXTRACTION_SCHEMA,
)
test_texts = [
"我叫张三,手机13800138000,订单ORD-001,申请退货,屏幕有坏点。",
"这是一个无效输入没有格式信息",
"用户:李四,电话:13912345678,工单号:TK-002,要投诉物流。",
]
for text in test_texts:
print(f"\n输入: {text[:80]}...")
result = extractor.extract(text)
print(f" 成功: {result['success']}")
print(f" 策略: {result['strategy_used']}")
if result["success"]:
print(f" 数据: {json.dumps(result['data'], ensure_ascii=False)}")
else:
print(f" 错误: {result.get('error', 'unknown')}")
4. 项目总结
优点 & 缺点对比
| 方面 | guided_json (Schema) | guided_regex | json_mode | baseline |
|---|---|---|---|---|
| JSON合法率 | 极高(95-100%) | 高(90-100%) | 中等(70-90%) | 低(50-70%) |
| 字段完整性 | 极高 | 高(取决于regex覆盖) | 中等 | 低 |
| 使用复杂度 | 中等(需定义Schema) | 高(需手写正则) | 低(一行参数) | 极低 |
| 性能开销 | +3-5%延迟 | +2-4%延迟 | +1-2%延迟 | 无 |
| 模型兼容性 | 需要vLLM支持 | 需要vLLM支持 | 需要模型本身支持 | 全部兼容 |
| 灵活性 | 低(严格JSON) | 高(任意格式) | 低(仅JSON对象) | 极高 |
适用场景
典型适用场景:
- 金融/医疗信息抽取:字段类型严格、不能有误 → guided_json
- 客服意图分类:A/B/C/D四选一 → guided_choice,可靠性接近100%
- 地址/日期等半结构化信息提取 → guided_regex
- 简单的"根据对话判断是/否"任务 → json_mode足够
- RAG系统中有固定Schema的知识图谱构建 → guided_json
不适用场景:
- 创意写作:格式约束会损害表达自由度
- 需要模型输出"思考过程+结论"的任务:约束会抑制CoT能力
- 不支持guided模式的旧版vLLM/特殊模型
注意事项
- guided_json不支持嵌套对象中的动态键:如果Schema中有
{"additionalProperties": true},guided模式无法正确约束 - 正则约束要避免灾难性回溯:复杂的嵌套量词可能导致正则引擎CPU打满,测试建议在1ms内完成匹配
- Schema定义要精准:过于宽泛的Schema(如所有字段都是optional)效果与无约束无异
- response_format="json_object"需要模型支持:并非所有模型都支持,不支持时vLLM会fallback到普通生成
- guided模式与temperature/sampling的交互:temperature=0时guided模式仍正常工作,但temperature>0时不保证确定性输出
常见踩坑经验
踩坑1:guided_json字段顺序敏感导致匹配失败
- 现象:Schema定义中字段顺序是
["name", "age", "city"],模型输出{"city": "...", "name": "...", "age": ...}时被拒绝 - 根因:部分vLLM版本中guided_json的字段顺序校验过严,要求与Schema定义顺序一致
- 解决:升级vLLM到最新版,或在Schema中不设required数组,降低引用顺序的敏感性
- 教训:结构化输出功能在不同vLLM版本间行为可能不同,版本升级前后要做回归测试
踩坑2:guided_regex中对中文的支持问题
- 现象:正则
[a-zA-Z]+能匹配英文字段,但中文名字字段始终匹配失败 - 根因:正则中的字符类
\w在不同vLLM后端实现中可能不包含中文字符(Unicode属性) - 解决:用Unicode范围
[\u4e00-\u9fff]+替代\w+;用[^"]*替代.*避免跨field匹配 - 教训:跨语言场景的正则需要显式处理Unicode,不能依赖默认字符类
踩坑3:response_format与stream=True同时使用导致行为异常
- 现象:开启流式输出时,response_format="json_object"导致第一个chunk包含完整JSON而非逐Token流式推送
- 根因:JSON模式需要在采样开始前就知道完整格式约束,与逐Token的流式推送存在实现冲突。vLLM内部在JSON模式下会缓冲输出直到JSON合法再推送
- 解决:如果必须用流式输出 + JSON,改用guided_json + stream=True,实测表现更好
- 教训:高级约束(response_format/guided_json)的流式行为差异是特性而非bug——需要在文档中明确标注
思考题
-
guided_json的实现原理是什么?vLLM是如何在Transformer的每个采样步骤中融入JSON Schema约束的?这种实现与"事后正则修复"相比,在计算复杂度和内存占用上有何差异?(提示:研究有限状态自动机、logit masking、以及
vllm/v1/sample/中的guided采样逻辑) -
如果一个业务需要同时输出"分析推理"和"结构化结果"(类似OpenAI的structured output + reasoning),在vLLM当前架构下应该如何实现?是否可以通过自定义guided模式+特殊的Schema设计来兼容两者?(提示:考虑在Schema中增加
reasoning字段,将思考过程作为JSON的一个字段输出)

所有评论(0)