先做一个动作再判断:把模型返回的原始字符串,和你在结构声明里写的字段,并排打印出来。原始文本本身已经是合法 JSON 却仍然抛解析错误,问题多半在解析器配置或字段类型;原始文本里夹着「好的,以下是……」、代码块围栏这类包裹,问题就在生成侧。两条路线的改法不一样,先分流再动手,比反复改提示词省时间。
判断顺序建议固定:先取未经解析的原始返回,再对照结构声明字段,最后才动解析器。原始返回不是合法 JSON,属于生成侧问题,优先收紧输出约束或用带修复能力的解析器;原始返回合法但仍报错,属于解析侧问题,检查字段名、类型、必填项和解析器开关。只改一侧往往不牢靠,异常分支里把原始文本落盘,才能复现和回溯。
打印模型未经解析的原始文本
最直接的验证方式是暂时不接解析器,让链的输出停在模型消息上,先看它到底吐了什么。
# 方式一:链上不挂 parser
msg = (prompt | llm).invoke({'question': '把这条订单抽成 JSON'})
print(type(msg)) # AIMessage
print(repr(msg.content)) # repr 能看见换行、反引号、首尾空格
如果链已经写成 prompt | llm | parser,可以在中间插一个打印节点,先看再解析:
from langchain_core.runnables import RunnableLambda
def dump(x):
print('--- RAW START ---')
print(repr(x.content))
print('--- RAW END ---')
return x
chain = prompt | llm | RunnableLambda(dump) | parser
多余内容通常出现在三个位置:开头一句「好的,我来帮你整理」;结尾一句总结或追问;以及 JSON 本体外面套的代码块围栏。少数情况是 JSON 内部被塞了注释符号,或者某个字符串字段里带了未转义的换行。建议用 repr 而不是直接 print,这样首尾空格和不可见字符更容易看出来。
核对结构声明的字段与模型实际输出
把结构声明和原始返回按字段逐项对齐,能区分「字段名不一致」和「类型不一致」。一段常见的结构声明:
from pydantic import BaseModel, Field
class Order(BaseModel):
order_id: str = Field(description='订单号')
amount: float = Field(description='金额,单位元')
items: list[str] = Field(default_factory=list)
对照方式建议用脚本做集合差集,而不是肉眼扫:
import json
raw_obj = json.loads(raw_text) # 这一步就失败,说明是生成侧问题
declared = set(Order.model_fields.keys())
returned = set(raw_obj.keys())
print('缺失:', declared - returned) # 模型没给的字段
print('多余:', returned - declared) # 模型自己加的字段
对不上的情况大致三类:模型把 order_id 写成 orderId 或 id;金额输出成字符串而声明是浮点数;模型额外给了一个解释性字段而结构里没有。前两类改提示词里的字段说明或加字段别名,第三类通常是模型想解释,需要在提示词里明确「不要输出未列出的字段」。另外 model_fields 是较新版本的写法,旧版本可能是 __fields__,按你实际安装的版本确认。
对比不同提示词下模型返回格式的差异
用同一段输入跑两版提示词,各自记录原始返回,差异点会很快暴露。
prompt_a = ChatPromptTemplate.from_messages([
('system', '你是订单整理助手,请把用户输入整理成 JSON,并说明你的判断理由。'),
('human', '{input}\n\n{format_instructions}'),
])
prompt_b = ChatPromptTemplate.from_messages([
('system', '你只输出 JSON,不要输出任何解释、前后缀或代码块标记。字段定义如下:\n{format_instructions}'),
('human', '{input}'),
])
# 并排记录示意(打印 repr 后粘贴到一起)
A 版: 以「好的,我来整理」开头 -> 代码块围栏 -> JSON 本体 -> 「理由:金额较小」
B 版: 直接以 { 开头,以 } 结束,无前后缀
差异一般出现在 JSON 之前的引导句、之后的解释段,以及有没有代码块围栏。如果 A 版只是外层多了文字而 JSON 本体正确,说明模型理解了结构,问题在输出约束不够紧;如果 A 版连字段都变了,说明「说明理由」这条指令影响了模型组织内容的方式。建议一次只改系统提示词,再观察原始返回的变化,不要同时改提示词和解析器,否则分不清是哪一侧起的作用。
换成可容忍杂散文本的解析方式再试
如果原始返回大体是 JSON,只是外面有说明文字或围栏,可以先换用对杂散文本更宽容的解析器,而不是立刻改结构。
from langchain_core.output_parsers import JsonOutputParser
parser = JsonOutputParser(pydantic_object=Order)
chain = prompt | llm | parser
另两条常见路线:一是带修复步骤的解析器,解析失败时把原始文本和错误一起交给模型再生成一次;二是把解析错误回填进提示词重试。两者都会额外调用一次模型,延迟和成本都要计入,重试次数建议设上限,例如 1 到 2 次,不要无限循环。导入路径随版本不同,以你所用版本为准。
# 占位写法,参数名以实际版本为准
fixing = OutputFixingParser.from_llm(parser=parser, llm=llm)
retrying = RetryOutputParser.from_llm(parser=parser, llm=llm, max_retries=1)
重试后的观察点有三个:字段值有没有被模型顺手改写,尤其是数字和 ID;修复后的对象里是否还留着原来多出的解释字段;同一条输入连续重试是否会输出不同结果。如果每次重试都在改字段值,说明内容本身与结构不匹配,属于生成侧问题,换解析器只是把报错掩盖掉。
在解析失败分支里把原始文本落盘
解析失败时原始文本如果没留下来,事后基本无法复现。可以在异常分支里统一保存:
import json, time, hashlib
from pathlib import Path
def run_with_capture(chain, inputs, tag='order'):
try:
return chain.invoke(inputs)
except Exception as e:
raw = inputs.get('__raw__', '') # 由打印节点写入的最后一条原始返回
stamp = time.strftime('%Y%m%d-%H%M%S')
digest = hashlib.md5(raw.encode('utf-8')).hexdigest()[:8]
path = Path('parse_failures') / f'{tag}-{stamp}-{digest}.json'
path.parent.mkdir(exist_ok=True)
path.write_text(json.dumps({
'tag': tag,
'raw': raw,
'error_type': type(e).__name__,
'error': str(e),
'model': getattr(llm, 'model_name', None) or getattr(llm, 'model', None),
'prompt_version': 'v2',
'parser': type(parser).__name__,
}, ensure_ascii=False, indent=2), encoding='utf-8')
raise
这里的关键是拿到原始返回本身。如果解析器挂在链的最后,异常抛出时原始文本已经丢掉,更稳的做法是让打印节点同时把内容写进一个上下文变量,或者把流程拆成「先调用模型、再单独解析」两步。文件名建议带业务标签、时间戳和原始文本的短哈希,便于去重;内容里至少保留原始文本、异常类型、模型标识、提示词版本和解析器类名,这几项决定了能不能把问题重新跑一遍。落盘目录不要放临时目录,否则复盘时找不到;原文可能含用户数据,保留周期需要结合所在环境的合规要求确认。