Pydantic与LLM结合:从非结构化文本到稳定结构化输出的工程实践
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邮箱地址格式通常是 xxxxxx.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_keyyour-api-key) def extract_event(text: str, model: str gpt-4o-mini) - EventInfo: schema EventInfo.model_json_schema() response client.chat.completions.create( modelmodel, 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, le1.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, modelgpt-4o-mini) print(event.model_dump())运行这个示例能直观看到几个关键点sentiment_score被约束在 -1 到 1 之间如果模型返回 1.5Pydantic 会直接校验失败我们就能立刻发现并处理。列表字段如果模型漏掉了也能得到一个空数组而不是 KeyError。这种“模型自由发挥但结构从来不乱”的体验是单纯用 prompt 解析 JSON 给不了的。4.2 重试与校验逻辑一次不行就再来一次模型第一次输出可能残缺不全这是常态别指望一次成功。我的做法是捕获 ValidationError把它变成一条提示语拼进新 prompt让模型自己看着办。这类似于人面对审稿意见时的修改循环。第二次不修改的话才让异常向业务层抛出。from tenacity import retry, stop_after_attempt, retry_if_exception_type class ExtractionWithRetry: def __init__(self, modelgpt-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、必要时自动重试。整个流程熟练之后基本上半天就能搭建一套可上线的结构化提取服务。