Pydantic 搭配 LLM 做结构化提取,这个组合我在生产环境里已经折腾了小半年。先说结论:如果你还在用正则表达式和写满 if-else 的规则引擎来解析用户输入、合同文本、简历信息,这可能是你今年最值得迁移的一套技术方案。
这套方案做的事其实很直白:让大模型读懂非结构化文本,再用 Pydantic 把模型输出的内容“校准”成你预先定义好的数据结构。一个负责语义理解,一个负责数据校验,各管一段,边界非常清晰。无论你是在做客服工单的自动分类、电商评论的要素抽取,还是想把纸质单据的关键信息落地到数据库,这套组合都能直接接上。
这个方案最吸引人的地方在于,它不需要你训练任何模型,也不需要你维护一堆摇摇欲坠的匹配规则。你只需要把“想要什么样的数据”这件事用 Pydantic 模型描述出来,剩下的事情交给 LLM 的指令遵循能力。而且 Pydantic v2 的校验机制会像一道质检流水线,把模型偶尔抽风输出的残缺字段、越界枚举值、错误格式一层层拦下来。
写这篇文章之前我复盘了一下最近做过的几个真实项目,把踩过的坑和经验都理了一遍。下面从选型、代码实现、常见问题三个角度拆开讲,有些代码可以直接抄。
1. 我为什么把这套方案推到生产环境
最早接触这个方向,是因为一个客户的数据录入项目。客户说他们每天有几百份非固定格式的 Excel 文件要录入系统,里面既有日期、金额这种标准字段,又有发票号、合同编号这种需要按特定规则去匹配的内容。之前的方式是 Python + 正则 + 模糊匹配,每种格式写一套解析规则。一开始只有两三种模板还好,后来格式越滚越多,正则规则膨胀到维护不下去。
当时第一次尝试用 LLM 直接返回 JSON,效果确实惊艳——完全没见过的格式,它也能把关键字段找出来。但问题也明显:模型偶尔会返回格式错误的 JSON,偶尔把金额字段的类型搞成字符串,偶尔把“生效日期”和“签署日期”搞反。这些错误在 demo 里无所谓,落到生产系统就变成脏数据了。
后来把 Pydantic 加进来,问题的解决思路就清晰了。Pydantic 不只是一个数据类定义工具,它其实是一层契约校验。你告诉模型你要什么结构,同时 Pydantic 在后台为你拦截一切不符合结构的内容。两者的分工明确:LLM 负责从模糊文本中找到信息,Pydantic 负责无缝转换成类型安全的 Python 对象。如果校验不通过,我们可以把错误信息再喂回给模型,让它重新修正一次。这个闭环一旦跑通,输出的稳定性会提升一个量级。
如果你只在 Jupyter Notebook 里做过一两次实验,可能体会不到这套组合在生产环境里的分量。但当你面对的是生产数据库、下游报表系统、定时任务管道,你会发现结构校验这个步骤是在替你挡麻烦,省掉无数个凌晨三点的告警电话。
2. 技术选型:Pydantic 不是数据类,是契约层
2.1 为什么是 Pydantic 而不是 dataclass 或字典
直接说结论:字典和 dataclass 在 LLM 提取场景里撑不住生产环境的需求。字典的问题是没有约束,拿到的数据没有任何类型检查,模型输出什么你就收什么。dataclass 虽然带了类型声明,但没有自动的校验与转换能力,比如字符串转 datetime、枚举值的合法性检查、嵌套结构的递归校验,这些都得你自己写。
Pydantic v2 的模型定义有几个我高频使用的特性:
- 类型自动转换:比如字段声明了
datetime,传入字符串时它能自动解析。 - Field 描述与约束:每个字段都可以挂上 description,这就是给 LLM 看的最佳指令。
- 枚举校验:把字段限制在合法的枚举值集合内,模型跑偏了直接报错。
- 嵌套模型:Order 内嵌 Customer、Item 数组,天然对应复杂的业务结构。
在结构化提取场景中,Pydantic 模型就是你和 LLM 之间的“协议”。你把协议定义清楚了,模型的输出才能稳定可预期。
2.2 方案对比,直接抄作业
选型之前可以看看当下的主流方案,核心区别在结构化输出策略。我把常见的几种方案放在一起对比一下:
| 方案 | 可靠性 | 原理 | 适用场景 |
|---|---|---|---|
| Prompt 要求 + JSON 字符串解析 | 较差 | 告诉模型输出 JSON,用 json.loads 解析 | 快速原型、非关键场景 |
| Structured Output / Function Calling | 较好 | 模型原生返回 JSON Schema 对应的 JSON,配合 Pydantic 校验 | 生产环境首选 |
| 工具调用(Tools) | 好 | 定义工具参数为 JSON Schema,模型按 Schema 填参 | 需要同时调用多个功能的场景 |
| JSON Mode | 中等 | 强制模型生成合法 JSON,但结构不稳定 | 模型不支持函数调用时的备选 |
| 自建微调模型 | 视数据量而定 | 自己训练模型生成指定格式 | 领域特别特殊、成本不敏感 |
我的默认选择是Function Calling 配合 Pydantic 模型定义的 JSON Schema。因为它不需要靠 chat 去猜字段,模型的 API 层直接在内部照 JSON Schema 生成结构化内容,生成的 token 更少,解析的成功率也更高。
有些朋友可能对 Function Calling 有误解,觉得它是用来“调函数工具”的。其实它的底层机制就是“用 JSON Schema 约束输出”。你定义一个 dummy function,参数结构就是你的目标数据结构,模型输出的那个 argument 就是你要的稳定结果。本质上,你可以把它理解为“虚拟的 schema 映射工具”。
3. 核心实现:模型定义、提示词与解析
3.1 模型定义——契约先行
从业务角度出发,先把目标结构想清楚,再写代码。比如我要从一篇新闻报道里抽取“事件摘要”,我希望得到四类信息,分别是最基本的标题文本、事件涉及的组织、链路相关的时间和事件分类。
from pydantic import BaseModel, Field from typing import List, Literal class EventInfo(BaseModel): title: str = Field(description="事件的一句话摘要,不超过30字") organizations: List[str] = Field(description="事件中涉及的组织或机构名称列表") event_time: str = Field(description="事件发生的时间,若原文未提及则填'未知'") category: Literal["科技", "财经", "体育", "娱乐", "民生"] = Field( description="事件所属分类,从给定候选中选择" )几个细节想特别提醒一下:
- 字段名用下划线风格,别用中文名。一方面 Pydantic 对中文 key 支持一般,另一方面在发送给模型作为 JSON Schema 时,下划线风格兼容性最好。
- Field 的 description 写详细一些,这就是你在给模型下指令。实测下来,description 写得好不好,直接决定提取准确率 20% 以上的浮动空间。比如 description 只写“事件时间”和写“事件发生的时间,若原文未提及则填'未知'”,效果差异很大。
- Literal 类型在结构化提取里太好用了。它能限制分类输出的取值范围,让模型做选择题而不是填空题,犯错概率能降低一个量级。
如果是简单的任务,这样做已经足够。复杂任务里,模型还可能需要输出一个包含多级嵌套的完整结构,比如“订单包含客户信息和商品列表”。这时 Pydantic 的嵌套模型能力就很受用,你可以定义 OrderInfo -> Customer、OrderInfo -> Items,一个标记好类的模型树就是一份天然的数据模板。
3.2 提示词设计——字段描述才是金矿
很多写 LLM 提取的人把精力都放在 system prompt 上,反复堆砌“你是一个专业的提取助手”这类内容,对字段描述反而一笔带过。我的实践结论是:在结构化提取场景,system prompt 只需简洁交代目标和输入格式,真正高价值的信息全在 Pydantic 字段的 description 里。
举个例子,对于一个“简历信息提取”任务:
class ResumeInfo(BaseModel): name: str = Field(description="候选人姓名,通常在简历开头出现") phone: str = Field(description="手机号码,11位数字,可能包含空格或横线,保留原始格式即可") email: str = Field(description="邮箱地址,格式通常是 xxx@xxx.com") years_of_experience: int = Field(description="工作年限,按整数填写,如果原文描述为'5年+'',填5") skills: List[str] = Field(description="候选人技能列表,如'Python'、'项目管理'等")这里的每个 description 都是在告诉模型“你要找什么、遇到变体怎么办”。模型不是一个简单的字符串匹配器,它需要这种细粒度引导才能应对同义不同形的表达。把系统提示中最碎、情况最复杂的部分挪进字段描述,模型的准确率提升立竿见影。
另一个重要的提示词细节是设置边界条件。例如某个字段如果原文没有对应的内容,应该返回空字符串还是保留默认值,必须在说明中写清楚。否则模型就可能自行脑补,编造出一个根本不存在的信息。
3.3 解析层:从 raw 到结构化对象
不管底层用的是 Function Calling 还是 JSON Mode,我在业务代码里都会做一层保护兜底,避免直接依赖模型返回的原始字符串。解析层的作用是:把模型输出的字符串转成 Pydantic 对象,校验失败时提供可读的报错信息,并支持重试回调。
import json from openai import OpenAI from pydantic import ValidationError client = OpenAI(api_key="your-api-key") def extract_event(text: str, model: str = "gpt-4o-mini") -> EventInfo: schema = EventInfo.model_json_schema() response = client.chat.completions.create( model=model, messages=[ { "role": "system", "content": "你负责从用户输入中提取结构化信息,只输出JSON,不要输出其他内容。" }, { "role": "user", "content": f"文本内容:\n{text}" } ], functions=[ { "name": "extract_event_info", "description": "提取事件信息", "parameters": schema } ], function_call={"name": "extract_event_info"} ) arguments = response.choices[0].message.function_call.arguments data = json.loads(arguments) try: return EventInfo.model_validate(data) except ValidationError as e: print("提取结果校验失败:", e) raise这段代码里的关键点在model_json_schema()。Pydantic v2 可以把模型直接转换为 JSON Schema,而且嵌套模型和枚举约束也会被完整描述在 schema 里。把这个 schema 传给函数调用接口,模型就按图施工,输出可解析、可验证的内容。
在 Function Calling 模式下,function_call参数我通常会显式指定,而不是让模型自行选择。因为一旦让模型自己决定调用哪个函数,它可能偶尔会“聪明地”把结构化任务变成一个文本回复任务,这会让解析逻辑变得不可预期。无论底层的实现是什么样,解析层的兜底校验都建议保留。模型输出是一个概率事件,Pydantic 这一层验证不是为了限制模型,而是为了让错误尽早暴露、尽快修正,避免脏数据向上游系统扩散。
4. 实操流程:从 Demo 到可维护的提取管线
4.1 一次完整的事件分类 + 实体抽取
下面提供一个可直接跑的完整示例。我们抽取一批新闻标题,输出结构包括分类、公司实体、情绪指数三部分,情绪指数用 float 类型限制。这比单独的字符串提取更能展示 Pydantic 在类型校验上的优势。
from pydantic import BaseModel, Field from typing import List, Literal import json class NewsExtraction(BaseModel): category: Literal["科技", "财经", "体育", "娱乐"] = Field( description="新闻所属分类,只能从给定四类中选择" ) companies: List[str] = Field( description="新闻中提到的公司或品牌名称,没提到就返回空数组" ) sentiment_score: float = Field( ge=-1.0, le=1.0, description="新闻情绪分值,-1代表极负面,1代表极正面" ) @field_validator("sentiment_score") @classmethod def check_score(cls, v): return round(v, 2) news_list = [ "某科技公司发布新款手机,股价应声上涨 2%", "某快递公司被曝用户数据泄露,引发广泛担忧", "某餐饮连锁品牌宣布开放加盟计划" ] for news in news_list: event = extract_event(news, model="gpt-4o-mini") print(event.model_dump())运行这个示例,能直观看到几个关键点:sentiment_score被约束在 -1 到 1 之间,如果模型返回 1.5,Pydantic 会直接校验失败,我们就能立刻发现并处理。列表字段如果模型漏掉了,也能得到一个空数组而不是 KeyError。这种“模型自由发挥但结构从来不乱”的体验,是单纯用 prompt 解析 JSON 给不了的。
4.2 重试与校验逻辑:一次不行就再来一次
模型第一次输出可能残缺不全,这是常态,别指望一次成功。我的做法是:捕获 ValidationError,把它变成一条提示语拼进新 prompt,让模型自己看着办。这类似于人面对审稿意见时的修改循环。第二次不修改的话,才让异常向业务层抛出。
from tenacity import retry, stop_after_attempt, retry_if_exception_type class ExtractionWithRetry: def __init__(self, model="gpt-4o-mini"): self.model = model def extract_with_feedback(self, text: str, max_retries: int = 2) -> EventInfo: last_error = None messages = [ { "role": "system", "content": "提取结构化信息,只输出JSON,不添加多余解释。" }, {"role": "user", "content": f"文本内容:\n{text}"} ] for attempt in range(max_retries + 1): try: # 调用模型并解析,代码同上 return parsed_result except ValidationError as e: last_error = e messages.append({ "role": "assistant", "content": json.dumps(prev_arguments) }) messages.append({ "role": "user", "content": f"上次输出未通过校验,错误信息:\n{e}\n请修正并重新输出。" }) raise last_error这段逻辑看起来简单,但实际生产环境里非常管用。有一次模型提取金额字段时把“月收入”和“年收入”混淆了,多亏这个重试机制,反馈错误后模型自己就纠正过来了。成本方面确实会多一两次调用,但相比下游系统处理脏数据的隐性成本,这点损耗不值一提。
4.3 缓存与幂等:别让模型重复阅读同一份文本
如果你的输入文本重复率高(比如相同模板的合同、同一批发票的扫描结果),强烈建议引入缓存层。相同文本不必让模型解读两遍,既省 token 又省时间。做法很简单:
import hashlib def content_hash(text: str) -> str: return hashlib.md5(text.encode("utf-8")).hexdigest() class CacheableExtractor: def __init__(self): self.cache = {} def extract(self, text: str) -> EventInfo: h = content_hash(text) if h in self.cache: return self.cache[h] result = extract_event(text) self.cache[h] = result return result更进一步,可以用 Redis 做跨服务共享缓存。这个优化在数据去重清洗任务里收益明显,比如舆情系统里大量重复转载的文章,去重后的调用量能直接砍掉一半。
5. 常见问题与排查思路实录
5.1 问题现象对照表
| 现象 | 可能原因 | 排查方向 |
|---|---|---|
| 模型返回 JSON 解析失败 | 没用 JSON Mode / Function Calling,模型自由发挥了 | 切换输出约束方式,启用函数调用 |
| 必填字段偶尔缺失 | prompt 边界描述不够清楚 | 在 Field description 里写明缺失时的兜底行为 |
| 类型被转换错误(数值变字符串) | 模型对类型理解不准 | 在 description 中强调“数字类型,不要加引号” |
| 枚举值不在定义范围内 | Literal 提示不够强 | 在 description 中显式列举所有合法值 |
| 嵌套结构漏建 | 复杂结构中模型能力不足 | 拆分提取;或提供 few-shot 示例 |
| 校验失败导致重试反复 | 输入文本确实缺少信息 | 给字段设默认值,并把“未知”“无”设为合法兜底 |
| 响应速度慢 | 模型输出 token 太多 | 精简 schema,减少重复描述和长枚举列表 |
| 成本上涨 | 重试次数过高 | 截断超长文本,保留关键片段 |
5.2 深度案例:当模型“自作主张”填空该怎么办
一次真实场景:我在做采购合同信息提取时,需要识别“付款方式”。原文本里没有明确写“款到发货”还是“账期30天”,但模型却老老实实地返回了一个值,编了个“电汇”。这种幻觉不至于报校验错误,却会因为数据不真实污染下游。
排查后发现问题的根源在于 prompt 里“缺失时填未知”的约束不够硬。模型在找不到信息时倾向于“根据常识脑补”,而不是承认“不知道”。
之后我在所有容易缺失的字段上统一加了一条规则:
class ContractInfo(BaseModel): payment_term: str = Field( default="未知", description="付款方式。如果原文没有明确说明,必须填写'未知',严禁推断或补充。" )同时,把“未知”作为合法选项加入到枚举或默认值中。这样模型即使偷懒,也不可能产出虚假信息。边界情况必须给模型指路,否则它会自己另辟蹊径。
6. 进阶玩法与个人经验总结
上面讲的是基于 Function Calling 的单次调用,实际项目中还有几个衍生方向值得扩展。
第一是流式解析。对于超长文档,先对文本做切片或摘要,再做结构化提取,比直接塞全文更稳定、更便宜。把大任务拆成“摘要 -> 提取 -> 校验”三级管道,成功率能大幅提高。
第二是结合多步 Agent。当需要从对话中提取复杂结构化信息时,可以先用一次调用判断“文本里有没有目标信息”,再用一次调用提取具体字段。这样比让模型一次做完更不易出错。本质上就是拿两次清晰的小任务,替换一次混乱的大任务。
第三是数据版本管理。当你的抽取规则和字段定义变化后,之前批处理过的数据需要重新解析。建议把模型版本、schema 版本、源文本 hash 都记录到结果表里,方便回溯和重新对齐。我吃过一次亏,schema 从 v1 升到 v2 后直接把旧缓存全清了,后面对账对了一整天。从那以后,我在缓存 key 里直接拼上了 schema 版本号,干净利落。
最后分享一个个人体会:这一套方案的核心价值不只是“解析 JSON”,而是把程序和语言模型约束在同一个数据结构边界内共同工作。你给模型多少条清晰的边界,它就能回报给你多少份稳定的输出。与其在 prompt 层面反复试探,不如把要求写进 Pydantic 模型里,让校验机制替你兜底。把这个意识建立起来,我在很多项目里的提取脚本都会遵循同一个模式:定义 clear schema、约束 model 输出、catch ValidationError、必要时自动重试。整个流程熟练之后,基本上半天就能搭建一套可上线的结构化提取服务。