做抽取类需求的人多半遇到过这种场面:你写了一段自认为天衣无缝的提示词,让模型“输出 JSON 格式”,跑通时一切顺畅,等到批量跑几千条数据时,总有那么几条返回带 ```json 围栏的、带中文冒号的、字段名突然变成拼音的、甚至干脆少了一半字段的。然后你在代码里堆起正则和 try-except,越堆越厚,最后自己都看不懂。LangChain 的Pydantic(JSON)解析器就是来收这个烂摊子的——它把“模型该吐什么结构”写成一份强类型的 Schema 契约,交给模型当说明书,回来再用 Pydantic 做校验和类型转换,一次性拿到可以放心点属性的对象,而不是一个随时会炸的字符串。这篇内容我会从 Schema 设计、解析器接入、Prompt 拼装一路写到批量场景的兜底策略,踩过的坑基本都写在里面了。适合正在做信息抽取、报表结构化、工单字段解析这类工作的朋友,也适合刚接触 LangChain、想搞清楚“解析器到底解决了什么”的同学,代码可以直接抄去改。
1. 先想清楚:为什么不能让模型直接吐 JSON
1.1 自由文本解析的三种死法
先说最原始的写法。拿到模型返回的字符串,做一次json.loads,成功就完事,失败就打印日志跳过。这套逻辑在小样本上活得很好,一旦上量就开始出问题,我给归纳成三种死法。
第一种是围栏污染。你在提示词里明明写了“只输出 JSON,不要任何解释”,但模型有时候还是会在外面包一层 Markdown 代码块,前后带三个反引号。json.loads遇到反引号直接抛异常。这种问题看起来小题大做,处理起来倒也简单,剥一下就行,可一旦你补上这个逻辑,就等于承认模型不听话,后面所有字段级的意外你都得兜。
第二种是字段漂移。同一个提示词,上午跑出来 key 叫company_name,下午跑出来叫companyName,再严重点给你输出中文字段名。这是因为模型每次生成都是一次独立的采样,提示词里如果没有一份精确到 key 名的结构说明,它就会按自己的“语言习惯”去猜。你的下游代码是按固定 key 取值,一次漂移就全线报错。
第三种是类型塌缩。这个最隐蔽。比如你要一个金额字段,声明要数字,模型给你返回"1,200.50"带千分位和引号的字符串;你要一个数组,模型给你返回用中文顿号分隔的一整串。json.loads不报错,因为你拿到的是合法 JSON,但data["amount"] * 1.1这行代码运行时会告诉你字符串不能乘浮点数。这种错误往往在业务算到一半才炸,排查成本极高。
三种死法本质是同一个原因:你把结构的约束放在自然语言里,而自然语言约束的执行力取决于模型的当时状态。要彻底解决,必须把约束挪到一个能被程序校验的地方去。
1.2 解析器在整条链路里站的什么位置
LangChain 对“输出解析器”这件事的抽象其实非常朴素:它就是一个约定了两个方法的对象,一个方法负责告诉模型“你要按这个格式输出”,另一个方法负责把模型吐出来的字符串变成程序能用的东西。Pydantic 解析器是其中约束最严的一类,它的两个方法分别由get_format_instructions()和parse()承担。
这里有个很多新手会忽略的关键点:解析器不是事后清洗工具,它是提示词的一部分。get_format_instructions()返回的是一段根据你的 Pydantic 模型自动生成的 JSON Schema 说明文本,你需要把它塞进提示词模板里,模型才知道边界在哪。很多人拿到解析器直接parser.parse(response),然后抱怨“模型不按格式来”,其实模型根本没见过这份说明书,它凭什么按格式来。
和直接json.loads相比,Pydantic 解析器多做了一层事情:它拿到 JSON 之后会交给 Pydantic 的model_validate去做校验,字段缺失、类型不符、枚举值越界,全都在这一层被拦下来,抛出的是明确指向某个字段的错误信息。你能立刻知道是order_id没给,还是status给了一个不在枚举里的值。这个错误信息的质量,是后面能不能快速修好的前提。
还有个容易混淆的选择:LangChain 里其实有两个 JSON 相关的解析器,一个是轻量的JsonOutputParser,它只保证结果能解析成字典,不做 schema 校验;另一个就是本篇要讲的PydanticOutputParser,它绑定一个模型类,做严格校验和类型转换。如果你的下游代码需要点属性访问、需要 IDE 补全、需要类型提示,选后者。如果只是做个中间态透传、结构随时会变,前者更省事。
2. Pydantic 模型设计:Schema 就是你和模型签的契约
2.1 字段定义的基本功
写 Pydantic 模型这件事,看起来是 Python 语法问题,实际上是产品设计问题。你在定义字段的那一刻,就决定了模型能给你什么、不能给你什么,也决定了解析失败时你有多难修。
先说命名。字段名一律用英文小写加下划线,不要用中文,不要用驼峰。原因不是崇洋媚外,而是模型在生成 JSON 时对 snake_case 的稳定性明显高于其他风格,而且你一旦用了中文 key,format_instructions里的 schema 会变得很长,token 消耗上去了,模型理解起来反而更容易出错。真要给人看中文,在字段的description里写就行,那里才是放自然语言的地方。
然后是必填与可选的判断。Pydantic 里给字段一个默认值它就变成可选,不给就必填。这个选择要非常谨慎。凡是你下游代码会直接用的字段,一律设为必填。因为可选字段在 schema 里会被标成非 required,模型的惰性一下子就上来了,能省就省。我做过一个测试,同样的抽取任务,把三个字段从可选改成必填之后,字段完整率从 82% 提到了 97%,代价只是解析失败时需要重试而已,非常划算。
类型选择上有个小陷阱。金额、数量这类数值,用float或者Decimal,不要用str。虽然用str永远不会解析失败,但那只是把校验责任推给了下游,等到业务层做汇总时你还是要转,而且转换时的异常更难定位。宁可让解析器在入口处就报错。同理,时间字段优先用datetime,Pydantic 自带常见格式的解析能力,能省掉一堆strptime。
提示:如果你的模型对数字类型的输出不稳定,经常给带单位或千分位的字符串,别急着改类型。可以在 Field 里加
description明确写“纯数字,不带单位、不带千分位分隔符”,比改类型有效得多。
2.2 Field 约束与 description 的真正价值
Field()这个函数是 Pydantic 里信息密度最高的东西,在解析器场景下尤其如此。它的description参数会被完整写进format_instructions生成的 schema 里,也就是说,你写在那里的每一个字,模型都会读到。这是整个链路里性价比最高的“提示词投放位”。
我一般的写法是,每个字段都写 description,格式遵循“含义 + 格式要求 + 示例值”三段式。比如:
from pydantic import BaseModel, Field from typing import List, Optional from datetime import datetime from enum import Enum class Priority(str, Enum): low = "low" normal = "normal" high = "high" urgent = "urgent" class TicketItem(BaseModel): ticket_id: str = Field( description="工单编号,格式为字母TK加8位数字,例如 TK20240517" ) title: str = Field( description="工单标题,不超过30个汉字,去除首尾空格" ) priority: Priority = Field( description="优先级,只能取 low / normal / high / urgent 之一" ) amount: float = Field( description="涉及金额,纯数字,不带货币符号和千分位,无则为 0" ) created_at: datetime = Field( description="创建时间,ISO 8601 格式,例如 2024-05-17T09:30:00" ) tags: List[str] = Field( default_factory=list, description="标签列表,最多5个,每个不超过10个字符,无则为空数组" )这段代码里,description承担了三个职责:解释业务含义、约束输出格式、给出参考样例。模型看到“例如 TK20240517”这种具体样例,模仿的准确率远高于纯文字描述。这也是为什么不建议把 description 写成“工单编号”四个字就完事——那等于没说。
Field的约束参数也值得用起来。min_length、max_length、ge、le、pattern这些在 Pydantic 校验层是硬约束,一旦模型给的越界,parse会直接失败,而不是悄悄放过一个脏数据。但要注意,这些约束不会以“你不能超过多少”的形式强约束模型,它们只是出现在 schema 里作为参考,模型可能忽略。所以约束的价值在于“帮你及时发现异常”,而不是“保证模型不越界”。
2.3 嵌套模型、列表与枚举的处理策略
真实业务里的结构很少是一层平铺的。订单里有商品列表,工单里有处理记录,简历里有工作经历数组。这时候就要用嵌套模型。
嵌套在 Pydantic 里写起来很自然,定义一个子模型,然后在父模型里声明List[子模型]就行。但这里有个实际问题:嵌套结构会让format_instructions变长,而且生成的 schema 里会出现$defs和$ref这种引用语法。部分模型对$ref的理解并不可靠,容易在输出时把引用原样吐出来,或者干脆漏掉某个子结构。
我的处理办法是能扁则扁。如果子结构只有两三个字段,直接摊平到父模型里,用前缀区分,比如sender_name、sender_dept,比嵌套一个Sender对象省心。如果确实是列表结构、条数不定,那嵌套无法避免,这时候要做两件事:一是给子模型的每个字段都写清楚 description,二是给列表字段本身加一条“最多 N 条,按时间倒序”这样的约束说明。列表是模型最容易偷懒的地方,不给上限它可能只给你一条,不要求排序它给你的顺序就完全随机。
枚举是个特别好用的东西,但用法有讲究。上面代码里Priority继承自str和Enum,这样它既是枚举又能被 JSON 序列化。枚举的取值最好用英文小写,别用中文,因为format_instructions会把枚举值原样列出,英文更短、更稳定。如果业务上必须展示中文,那就用英文枚举,在下游做映射表,别为了省事把中文塞进枚举值里。
还有一种情况要特别提一下:可选嵌套。比如“用户可能留了备注,也可能没有”。如果你声明成Optional[str] = None,模型很可能直接不输出这个 key。这在 JSON 层面是合法的,但下游取值时要各种判空。我的习惯是给这类字段一个空字符串默认值,在 description 里明确写“没有则输出空字符串”,这样 key 一定会出现,下游处理起来省事很多。类型上仍然是必填,只是值可以是空字符串。
3. 解析器接入全流程:从依赖到落地
3.1 版本差异是第一个坑,先确认环境
动手之前必须先确认版本,因为 Pydantic 在 v1 到 v2 之间有不兼容改动,LangChain 的不同版本对 Pydantic 的依赖方式也不同。早期版本里你能看到from langchain_core.pydantic_v1 import BaseModel这种写法,那是兼容层的产物;较新的版本则直接依赖 Pydantic v2,你from pydantic import BaseModel就行。
我建议的做法是显式锁定版本,别让环境漂移。下面是一份我常用的依赖清单,requirements.txt直接抄:
# requirements.txt langchain-core>=0.2.0 langchain>=0.2.0 pydantic>=2.5.0,<3.0.0装完之后跑一句确认:
python -c "import pydantic; print(pydantic.VERSION)"输出以2.开头就对了。如果输出是1.x,说明你的环境里有旧包占位,这时候要么升级,要么在代码里改用兼容层导入,二选一,别混着用。混用的典型症状是定义模型时不报错,parse的时候抛出一个你完全看不懂的TypeError,排查半天发现是两套 Pydantic 在打架。
模型客户端那块我不写死具体实现,因为每个人接入的后端不一样。你只需要保证拿到的是一个符合 LangChainBaseChatModel接口的对象就行,它得支持invoke和返回带content的消息对象。我下面统一用一个build_chat_model()函数代替,你把它换成自己的初始化逻辑即可。
3.2 构建解析器与 format_instructions 长什么样
解析器的构造只有一行,但它做的事不少:
from langchain_core.output_parsers import PydanticOutputParser parser = PydanticOutputParser(pydantic_object=TicketItem)这行执行完之后,解析器内部就持有了模型类的 schema。接下来最关键的一步,是把它生成的说明书拿出来看看:
print(parser.get_format_instructions())输出大概长这样(省略了部分字段):
The output should be formatted as a JSON instance that conforms to the JSON schema below. As an example, for the schema {"properties": {"foo": {"title": "Foo", "description": "a list of strings", "type": "array", "items": {"type": "string"}}}, "required": ["foo"]} the object {"foo": ["bar", "baz"]} is a well-formatted instance of the schema. The object {"properties": {"foo": ["bar", "baz"]}} is not well-formatted. Here is the output schema: {"properties": {"ticket_id": {"description": "工单编号,格式为字母TK加8位数字,例如 TK20240517", "title": "Ticket Id", "type": "string"}, ...}, "required": ["ticket_id", "title", "priority", "amount", "created_at"], ...}强烈建议每个人第一次用的时候都把它打印出来读一遍。你会直观地看到:description原样进去了,required列表是按你定义的必填字段来的,枚举的取值也在里面。这段文本占了大概 200 到 500 个 token(结构越复杂越多),这是必须付出的成本,别心疼。
读完之后你会意识到一件事:这段说明是英文的,而你的 description 是中文的。中英混排在 schema 里完全没问题,模型两种都读得懂,但如果你发现某次抽取效果不好,可以试着把 description 也改成英文,某些模型在纯英文 schema 下的表现确实更稳。这不是玄学,是训练数据的分布问题。
还有个小细节:get_format_instructions()每次调用都会重新生成。在链路里它会被放进partial_variables,也就是说每次构造 prompt 时都会执行一次。如果你在批量场景里跑几万条,这个开销虽然不大,但也没必要重复。我一般在模块级别算一次,存成常量复用。
3.3 Prompt 拼装:partial_variables 的正确用法
把说明书塞进提示词,有两种写法。一种是手动拼字符串,把format_instructions用 f-string 插进去。这种方式能用,但一旦你要换模板、要做多轮,维护起来就很难受。LangChain 提供的正规做法是partial_variables:
from langchain_core.prompts import PromptTemplate prompt = PromptTemplate( template="""你是一名工单信息抽取助手。请从下面的工单原文中抽取结构化信息。 抽取要求: 1. 只输出 JSON,不要输出任何解释性文字 2. 不要使用 Markdown 代码块包裹结果 3. 原文中没有提到的信息,按字段说明里的默认值处理 4. 字段名必须与 schema 完全一致,不要改写、不要翻译 {format_instructions} 工单原文: {raw_text} """, input_variables=["raw_text"], partial_variables={"format_instructions": parser.get_format_instructions()}, )这里有个重点:input_variables里只写了raw_text,format_instructions放到partial_variables里。这样做的意义是,调用时你只需要传raw_text,说明书是自动填充的,不用每次都记得带上。批量跑的时候这个差别很明显——少一个必传参数,就少一处出错的机会。
模板的顺序也值得说。我把format_instructions放在原文之前,紧跟在抽取要求之后。理由是大模型的注意力对靠后的内容更敏感,原文放在最后,模型对原文的“在场感”最强;而 schema 作为规则放在中段,不至于被原文淹没。这个顺序我对比过几种排法,在我的场景里效果最稳。当然不同模型的最优顺序可能不一样,你可以做个小规模的 A/B 测试,各跑五十条看字段完整率。
模板里那四条要求看着琐碎,但每一条都对应一类实际故障。“只输出 JSON”防解释性前缀,“不要用代码块包裹”防围栏污染,“没有的按默认值”防字段缺失,“字段名必须一致”防 key 漂移。这四条是通用配置,不管你做什么抽取任务都可以直接带上。
3.4 完整链路跑通与结果使用
现在把三段拼起来。用 LCEL 的管道写法最清爽:
from langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import PromptTemplate def build_chain(model): parser = PydanticOutputParser(pydantic_object=TicketItem) prompt = PromptTemplate( template="""你是一名工单信息抽取助手。请从下面的工单原文中抽取结构化信息。 抽取要求: 1. 只输出 JSON,不要输出任何解释性文字 2. 不要使用 Markdown 代码块包裹结果 3. 原文中没有提到的信息,按字段说明里的默认值处理 4. 字段名必须与 schema 完全一致,不要改写、不要翻译 {format_instructions} 工单原文: {raw_text} """, input_variables=["raw_text"], partial_variables={"format_instructions": parser.get_format_instructions()}, ) return prompt | model | parser chain = build_chain(build_chat_model()) raw = """ 单号 TK20240517001,客户反馈设备在保修期内出现异响,比较着急, 涉及维修报价 1,200.50 元,工单创建于 2024-05-17 09:30。 """ result = chain.invoke({"raw_text": raw}) print(type(result)) # <class '__main__.TicketItem'> print(result.ticket_id) # TK20240517001 print(result.priority) # Priority.urgent 或 Priority.high print(result.amount * 1.1) # 数字,可以直接算 print(result.tags) # ['保修', '异响']注意最后几行的输出。result不是字典,是TicketItem实例,你可以直接点属性拿值,IDE 会给补全,类型检查也能过。amount是float,可以直接参与运算。priority是枚举成员,比较时用result.priority == Priority.urgent就行。这些细节看起来只是“方便一点”,但在几百行的业务函数里传递时,能省掉大量的dict.get()和类型断言。
如果确实需要字典,Pydantic v2 用result.model_dump(),v1 用result.dict()。要序列化成 JSON 字符串用model_dump_json(),枚举会被自动转成值。这两个方法名不一样,写的时候留意版本,dict()在 v2 里虽然还能用但会告警。
再补充一句关于chain.invoke的异常处理。解析失败时抛出的异常类型是OutputParserException,它区别于模型调用本身的网络异常。这两类异常要分开捕获,因为处理策略完全不同:网络异常该重试重试,解析异常要考虑的是格式修复或者重新生成。混在一个except Exception里,等于放弃了针对性处理的机会。
有一种情况需要提前打招呼:如果你的模型不支持在单次请求里返回较长的结构化输出,或者max_tokens设得偏小,输出会被截断,表现为 JSON 末尾缺失,parse报“Expecting value”之类的错误。这个问题的根因不在解析器,在参数配置。粗略估算方式是:输出 token 数 ≈ schema 字段数 × 每个字段的平均值长度(token 计) × 1.5 的冗余系数。一个十个字段的模型,输出通常在 200 到 400 token 之间,max_tokens给到 1024 比较稳妥。设成 256 就属于自找麻烦。
4. 实战踩坑:解析失败的原因排查与兜底
4.1 常见失败模式速查表
跑得多了,失败原因其实高度集中在有限的几类。我把它们整理成一张表,出问题时对着看,基本能定位到八九成:
| 报错现象 | 根因 | 处理办法 |
|---|---|---|
Invalid json output且内容带 ```json | 模型用了代码块包裹 | 提示词明确禁止,或在解析前做一次围栏剥离 |
Expecting value: line 1 column 1 | 输出为空或被截断 | 检查max_tokens,检查是否有内容过滤 |
字段缺失,报Field required | 该字段是可选的,模型省略了 | 改成必填并给空值默认,或在 description 里强调必须出现 |
类型错误,如Input should be a valid number | 模型给了带单位的字符串 | 在 description 里明确纯数字,或加field_validator做清洗 |
枚举越界,如Input should be 'low'... | 模型自己造了一个值 | 枚举值数量少时,在提示词里再列一遍可选取值 |
UnicodeDecodeError或中文乱码 | 输出编码问题 | 检查客户端编码设置,确保按 UTF-8 处理 |
| 偶发成功偶发失败 | 温度太高 | 抽取任务把 temperature 设成 0 或 0.1 |
关于温度这一项,我吃过亏。早期觉得抽取任务温度设 0.3 也无所谓,跑起来发现同一批数据重跑两次,字段完整率能差 8 个百分点。抽取是确定性任务,不需要创造力,temperature=0是标配,有条件的话再把top_p也压一压。这一项改完,很多“偶发失败”就消失了。
编码问题在国内业务里出现的频率比想象中高。模型返回的中文如果客户端那边走了错误的编码,你在日志里会看到一堆问号或者奇怪的符号,而 Pydantic 校验时字段长度对不上,报错信息还特别难懂。排查办法是打印原始的response.content,看看是不是真的乱码,如果是,问题在客户端配置,不在解析器。
4.2 用修复解析器兜住那些格式小瑕疵
即便提示词写得很到位,仍然会有小比例的输出在格式上出岔子,典型的是多了个逗号、用了单引号、多了个解释性的前缀。这种“内容对但格式错”的情况,最经济的处理方式不是重新生成(那要花一次完整的调用成本),而是让模型只做格式修复。
LangChain 提供了两条路。第一条是OutputFixingParser,它包住原解析器,失败时把错误信息和原始输出一起发给模型,让它修成合法格式再重新解析:
from langchain.output_parsers import OutputFixingParser fixing_parser = OutputFixingParser.from_llm( parser=parser, llm=build_chat_model(), ) # 用它替换原来的 parser 即可 fixed = fixing_parser.parse(bad_output)第二条是RetryOutputParser,它比前者多知道一层上下文——它需要原始 prompt 的内容,因为修复时会把完整 prompt 一起带进去,让模型基于“原本要问什么”来修正:
from langchain.output_parsers import RetryOutputParser retry_parser = RetryOutputParser.from_llm( parser=parser, llm=build_chat_model(), max_retries=2, ) prompt_value = prompt.format_prompt(raw_text=raw) result = retry_parser.parse_with_prompt( completion=bad_output, prompt_value=prompt_value, )两者的区别在成本上。OutputFixingParser只发错误信息和脏输出,token 少、快;RetryOutputParser带完整上下文,准确率更高但贵。我的经验是:如果失败原因是纯语法问题(尾逗号、引号),用前者;如果是字段缺失或者语义理解跑偏,用后者,因为光看一段格式错误的 JSON,模型也不知道缺的那个字段该填什么。
还有一条必须强调:修复解析器要设上限。max_retries别给太大,2 到 3 次足够。我见过有人设成 10 次,结果模型在某条特别离谱的输入上反复修,一次任务跑了几十秒还没结束。失败率高的场景,本质是提示词或者 schema 有问题,靠重试是治标不治本。
4.3 从源头降低失败率的几个 Prompt 技巧
兜底方案永远是最后一道防线,真正该花力气的是把失败率压下去。下面这几招是我实测有效的,按投入产出比排序。
第一招:给一个完整的输出示例。这招的收益最大。不要只依赖format_instructions里的 schema,那毕竟是抽象描述。你在提示词里直接贴一段符合要求的 JSON 样例,模型会照着模板填空,格式错误的概率断崖式下降。样例不用长,两三个字段的代表值就够,注意样例里的字段顺序和最外层结构要和你期望的完全一致,模型很容易连顺序都模仿。
第二招:把 schema 里的字段说明写具体。前面讲过了,“工单编号”这种 description 等于没说,写成“字母TK加8位数字,例如 TK20240517”就有约束力了。这一步纯粹是写作功夫,多花十分钟,后面省下几百次重试。
第三招:在提示词里复述关键约束。枚举值、数值格式、时间格式这三类是最容易出错的,在模板的要求列表里再强调一遍,哪怕和 schema 里的信息重复也没关系。重复不花钱,重试才花钱。
第四招:给模型一个“不知道就填默认”的出口。很多字段缺失是因为原文里确实没提,模型不知道该填什么就干脆省略了。你明确告诉它“没提到就填 0 / 空字符串 / 空数组”,它就有了行为指引,输出会稳定很多。
第五招:控制单次抽取的字段数量。字段一多,模型注意力分散,漏字段的概率上升。我做过一个粗略对比,同样难度的抽取任务,8 个字段的时候完整率在 95% 左右,一口气给到 20 个字段就掉到 85% 以下。如果业务上确实需要几十个字段,拆成两次调用,分别抽不同的字段组,总成本可能还更低,因为重试少了。
注意:不要在提示词里写“尽可能准确”这种模糊要求,对模型没有任何指导作用,反而可能让它变得保守,开始给你加解释性文字。所有要求都应该是可判定的,比如“不超过30字”“只能是这四个值之一”。
5. 进阶:把解析器放进真实业务链路
5.1 批量抽取的并发与限速
真实场景很少是一条一条来。几百上千条文本要抽,串行跑太慢,必须并发。这里的关键是并发度不是越高越好,取决于你的模型服务端限流策略。我的经验值是从 4 到 8 起步,观察两件事:一是错误率有没有明显上升(限流会导致超时和空返回),二是总耗时有没有随并发线性下降(如果没有,说明瓶颈在服务端)。
批量跑的时候还有一个容易忽略的点:单条失败不能中断整批。用chain.batch()的话,默认行为是遇到异常直接抛出,整批全挂。更稳的做法是自己控制循环,逐条捕获OutputParserException和网络异常,把失败的条目单独收集起来,跑完之后统一做一次修复或者重试。这个模式我称为“两轮制”:第一轮全量跑,第二轮只处理失败的那些。实测下来,第二轮的条目数通常只占全量的 3% 到 8%,成本可控。
收集失败条目时记得把原始输入、模型的原始输出、异常信息三个都存下来。只存异常信息是没用的,你回看的时候根本不知道当时模型吐了什么。这三个字段存成一条记录,后面分析失败模式的时候价值极大,我就是靠这批数据发现“围栏污染”占比接近一半的。
还要提一下幂等性。批量任务难免要重跑,如果你的下游是直接写库,重跑会产生重复数据。给每条记录带一个由原文算出来的唯一标识(比如哈希值),写库时做更新而不是插入,重跑就安全了。这个和解析器没关系,但批量场景下必须想到。
5.2 和 LCEL 组合时的几个细节
用管道符|组合链的时候,解析器就是一个普通的 Runnable,它接收上游字符串、输出 Pydantic 对象。这个特性带来一个好处:你可以把它嵌在更大的链路中间。比如“抽取 → 校验 → 补充查询 → 生成摘要”这样的四段式,前三段都能用管道串起来,最后一段再接一个纯文本输出。
有个细节值得说:如果你想让链路支持流式输出,PydanticOutputParser是不支持的,因为它必须拿到完整字符串才能解析。这在预期之内,结构化输出天然和流式冲突。如果你的产品需要“边生成边显示”,那就得分两步走,先流式生成展示给用户,再另起一次调用做结构化抽取,或者干脆用轻量的 JSON 解析器配合前端做增量解析。
另一个细节是with_retry。LangChain 的 Runnable 提供了通用的重试封装,但它重试的是整个链,包括模型调用,成本高。解析失败的重试我建议用前面讲的修复解析器,只在解析这一步重试,不重新调用模型生成。这两个重试的粒度不同,别搞混了。
最后提一下调试。链路一长,出问题不知道是哪一环。管道写法里可以插一个“透传打印”的环节,把中间结果打到日志里,不改变数据流。这个技巧在排查“到底是模型没生成对,还是解析器没解析对”时特别有用,两边看一眼就清楚了。
5.3 结构化输出的几条路线对比
Pydantic 解析器不是唯一解。随着各家模型服务开始提供原生的结构化输出能力,路线的选择变多了。我把常见的几条路列出来对比,你可以按自己的条件选:
| 路线 | 原理 | 优点 | 局限 |
|---|---|---|---|
| Pydantic 解析器 + 提示词 | schema 写进提示词,模型自由生成,事后校验 | 通用,任何模型都能用,逻辑完全可控 | 依赖模型遵循指令,有小概率格式错误 |
| 模型原生 JSON 模式 | 服务端约束输出必须是合法 JSON | 语法层面几乎不会错 | 只保证是 JSON,不保证字段符合你的 schema |
| 原生结构化输出(带 schema) | 服务端按 schema 约束解码 | 字段和类型双重保证,成功率最高 | 依赖服务端支持,schema 复杂度有上限 |
| 函数调用式抽取 | 把 schema 当成一个函数的参数定义 | 复用已有的调用协议,兼容性好 | 需要服务端支持,调试信息不如前者直观 |
我的实际选择是:服务端支持原生结构化输出就用原生的,不支持就用 Pydantic 解析器兜底。两者在代码层面的接口可以做得一致——都是给一个 Pydantic 模型,返回一个实例,这样上层业务代码不用改。这个抽象值得做,因为模型服务的能力是在变的,今天不支持明天可能就支持了,把差异收敛在一处,迁移成本就低了。
需要注意的是,原生结构化输出对 schema 的复杂度有限制,嵌套层级深、字段特别多的模型可能会被服务端拒绝。这时候还是得退回提示词方案,把结构拆简单一点。所以 Pydantic 解析器这套东西,即便在有原生支持的今天,依然是必须掌握的底牌。
6. 几个不那么显眼但很实用的经验
写到这里,主体流程已经完整了。最后补几条细碎但确实救过我命的经验,都是文档里不太会写的。
关于 logger 的位置。千万别在解析器的parse外面套 try 然后直接pass。解析失败静默跳过,等于把问题藏起来,跑一周之后你发现数据缺了一大块,还不知道从哪天开始的。我的做法是失败时至少打三行日志:原始输入的前 200 字符、模型原始输出的前 500 字符、完整的异常堆栈。这三行足够定位绝大多数问题。
关于字段顺序。Pydantic 模型的字段定义顺序会影响format_instructions里 schema 的顺序,而模型是倾向于按顺序生成的。把最重要的字段放前面,把容易出错、需要仔细思考的字段放后面,让模型在“热身”之后再处理难的部分。这个影响不大但确实存在,尤其是字段多的时候。
关于测试。上线前一定要准备一个几十条的“刁钻样本集”,专门放那些边界情况:原文超短只有五个字、原文超长几千字、原文里混了表格和乱码、原文是纯英文、原文里数值和单位混在一起。这批样本跑通了,线上的稳定性才有底。我一般要求这批样本的解析成功率在 98% 以上才敢上线,低于这个数就先改提示词,别指望线上重试能救回来。
关于成本核算。结构化抽取的 token 消耗分三块:提示词(含 schema,每次都要发)、原文、输出。schema 那部分是固定开销,字段越多越贵。如果一个字段你其实用不上,果断删掉,format_instructions会跟着变短。我优化过一个模型,从 24 个字段精简到 15 个,单次调用的输入 token 降了将近三成,成功率反而升了,因为模型不用再纠结那些它根本抽不出来的字段。
关于错误信息的可读性。Pydantic v2 的校验错误信息比 v1 详细得多,会精确指出字段路径和期望类型。如果你的报错还是一句笼统的“解析失败”,那大概率是你在外层把它包起来了。把原始异常直接暴露在日志里,别自己拼一句模糊的提示,原始信息才是最有用的。
关于迭代节奏。我见过太多人一上来就想设计一个覆盖所有场景的大模型,二十几个字段、四五层嵌套,结果调了两天调不通。正确的做法是从最小可用开始,先跑通三五个核心字段,稳定之后再逐步加。每加一批字段就重跑一次样本集,看成功率掉没掉。这个节奏慢,但返工少。我个人的体会是,结构化抽取这件事,前期在 schema 设计上多花的那两个小时,永远比后期在线上排查脏数据省下来的时间多得多。