ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

LangChain+Pydantic结构化输出问答器实战:从自由文本到类型安全

LangChain+Pydantic结构化输出问答器实战:从自由文本到类型安全 1. 为什么我要做结构化输出问答器做Agent开发的朋友大概率都遇到过这个场景你辛辛苦苦搭好了一条链路模型也能正常返回内容但返回的东西是一坨自由文本。你想把它塞进下游系统比如写数据库、调接口、渲染前端组件结果发现根本没法直接用。你得写一堆正则去抠字段模型稍微换个说法正则就崩了。这种痛做过三个以上Agent项目的人应该都懂。我这次要聊的结构化输出问答器就是专门解决这个问题的。它的核心思路很简单让大模型在回答问题的同时直接吐出一个符合预定义Schema的结构化对象而不是一段需要二次解析的自然语言。技术栈上我用的是LangChain做编排Pydantic做Schema定义和校验这也是目前Agent开发里最主流、最稳的组合之一。这个问答器能做什么举个实际例子你问它帮我查一下北京今天天气怎么样适合穿什么衣服它返回的不是一段话而是一个对象里面有city、weather、temperature、suggestion这些字段每个字段类型明确、值可直接用。下游拿到这个对象想存库就存库想渲染就渲染完全不用再解析。适合谁来参考如果你已经写过基础的LangChain调用知道什么是PromptTemplate、什么是LLMChain但还没系统搞过结构化输出那这篇就是给你准备的。如果你是完全的新手也没关系我会把Pydantic的基础用法和LangChain的接入方式都讲清楚跟着抄作业就能跑起来。我踩过的坑先提前说一个很多人以为结构化输出就是加一句请用JSON格式返回然后json.loads一下完事。实测下来这种做法在简单场景能凑合一旦字段多了、嵌套深了、模型开始自由发挥翻车率极高。真正稳的做法是把Schema约束交给框架层去强制而不是靠Prompt祈祷模型听话。2. 整体设计思路与方案选型拆解2.1 为什么是LangChain加Pydantic这个组合先说选型逻辑。做结构化输出市面上大概有三条路一是纯Prompt约束加手动解析二是用模型厂商原生的Function Calling或Tool Calling三是用框架封装好的结构化输出能力。第一条路最原始也最脆弱前面已经吐槽过了。第二条路很稳但问题是绑定厂商你换个模型就得重写一遍。第三条路是我最终选的LangChain的with_structured_output方法底层会自动适配不同模型的能力能用原生工具调用的就用原生不支持的会退化成Prompt加解析器对上层代码是透明的。Pydantic在这里的角色是契约定义者。你用Python类的方式把期望的输出结构写出来每个字段的类型、描述、是否必填都标清楚。LangChain会把这个Pydantic模型转成模型能理解的JSON Schema塞进调用参数里。模型返回后LangChain再用Pydantic做一次校验类型不对、字段缺失都会在这一层被拦住。这个组合最大的好处是类型安全。你在代码里拿到的是一个Pydantic对象IDE能给你补全字段名写错了直接报错不用等到运行时才发现。相比拿到一个dict然后result[nmae]这种拼写错误体验完全不是一个级别。2.2 问答器的核心链路设计整个问答器的链路我拆成四层。第一层是输入层接收用户的自然语言问题。第二层是Schema定义层用Pydantic定义好期望的输出结构。第三层是模型调用层LangChain负责把Schema和问题一起发给模型并处理返回。第四层是后处理层对拿到的结构化对象做业务逻辑处理比如存库、调接口、返回给前端。为什么要分这么细因为每一层的职责边界清晰后期维护和替换成本低。比如你想换个模型只动第三层的配置就行你想加字段只动第二层你想改业务逻辑只动第四层。我见过太多项目把这几层揉在一起改一个字段要翻遍整个文件那滋味不好受。还有一个设计决策是关于是否要支持多轮对话。我最初做的是单轮问答后来发现实际场景里用户经常追问。比如先问北京天气再问那上海呢。如果每次都是独立的模型就丢失了上下文。所以我在链路里加了一个简单的对话历史管理用LangChain的ChatPromptTemplate配合MessagesPlaceholder来承载历史消息。这里要注意结构化输出的Schema在多轮场景下依然要生效不能因为加了历史就退化成自由文本。2.3 结构化输出的两种模式选择LangChain的with_structured_output支持两种模式这个细节很多人没注意。一种是强制模式模型必须返回符合Schema的内容不符合就报错重试。另一种是宽松模式允许模型返回不符合Schema的内容由框架尽力解析。我实测下来的建议是生产环境用强制模式调试阶段可以用宽松模式。强制模式的好处是数据质量有保障坏处是如果Schema设计得太苛刻模型可能反复失败浪费token。宽松模式适合你在探索阶段先看看模型大概能返回什么再反过来调整Schema。具体怎么选看你的下游能不能容忍脏数据。如果这个结构化对象要直接写数据库那必须强制。如果只是用来做展示宽松一点也无妨。我在项目里默认用的是强制模式配合重试机制稳定性很好。3. 核心细节解析与实操要点3.1 Pydantic模型定义的几个关键技巧先看一个最基础的Schema定义from pydantic import BaseModel, Field from typing import List, Optional class WeatherAnswer(BaseModel): city: str Field(description城市名称) weather: str Field(description天气状况如晴、多云、雨) temperature: float Field(description温度摄氏度) suggestion: str Field(description穿衣建议)这段代码看着简单但有几个细节决定了成败。第一每个字段都要写description。这不是可选项是必选项。模型是靠description来理解字段含义的你写城市名称和写city name模型的表现可能就不一样。中文场景下我建议用中文描述和用户提问的语言保持一致模型理解更准。第二类型要尽量具体。temperature用float而不是str这样模型返回25的时候框架会自动转成25.0返回二十五就会报错逼着模型用规范格式。如果你用str模型可能返回25度、25℃、二十五摄氏度各种花样下游处理起来很痛苦。第三嵌套结构要控制深度。Pydantic支持嵌套模型比如一个WeatherAnswer里再套一个Location模型。但嵌套超过三层模型的准确率会明显下降。我的经验是嵌套不要超过两层字段总数控制在15个以内。超过这个量级考虑拆成多次调用。第四可选字段用Optional。有些字段不是每次都有值比如空气质量可能查不到。用Optional[str] None模型返回null也不会报错。但要注意Optional字段太多会让模型困惑不知道该填还是不该填所以能必填的尽量必填。3.2 LangChain接入结构化输出的完整写法Schema定义好之后接入LangChain的代码大概长这样from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm ChatOpenAI(modelgpt-4o-mini, temperature0) structured_llm llm.with_structured_output(WeatherAnswer) prompt ChatPromptTemplate.from_messages([ (system, 你是一个天气问答助手根据用户问题返回结构化信息。), (human, {question}) ]) chain prompt | structured_llm result chain.invoke({question: 北京今天天气怎么样适合穿什么}) print(result.city, result.temperature)这里有几个点值得展开。temperature设为0这是结构化输出的标配。温度越高模型越有创意字段值就越不可控。做结构化输出我们要的是稳定和确定不是创意。with_structured_output的调用位置是在llm对象上不是在chain上。这个顺序不能反。先拿到一个绑定了Schema的llm再和prompt组合成chain。chain的写法用了管道符这是LangChain Expression LanguageLCEL的语法。相比老式的LLMChainLCEL更简洁也更容易组合。如果你还在用老写法建议迁移过来社区支持更好。3.3 字段校验与错误处理机制结构化输出不是万能的模型偶尔还是会返回不符合Schema的内容。这时候Pydantic会抛ValidationError。我的处理策略是捕获异常加重试from pydantic import ValidationError from langchain_core.exceptions import OutputParserException def safe_invoke(chain, inputs, max_retries3): for i in range(max_retries): try: return chain.invoke(inputs) except (ValidationError, OutputParserException) as e: if i max_retries - 1: raise print(f第{i1}次失败重试中{e})重试的时候有个技巧把错误信息反馈给模型。LangChain的某些版本支持自动把校验错误塞回Prompt让模型知道哪里错了下次改正。这个机制叫self-correction实测能显著提升成功率。如果你的版本不支持可以手动实现把e的内容拼进下一轮的Prompt里。还有一个坑是超时和限流。结构化输出的调用通常比普通调用慢因为模型要思考字段怎么填。如果你的场景对延迟敏感要考虑加超时控制和降级方案。我的做法是设一个3秒的超时超时就返回一个默认的空对象保证主流程不阻塞。4. 实操过程与核心环节实现4.1 环境准备与依赖安装先把环境搭起来。Python版本建议3.10以上Pydantic 2.x对Python版本有要求。依赖装这几个pip install langchain langchain-openai pydantic python-dotenvlangchain-openai是LangChain拆出来的OpenAI集成包现在LangChain主包越来越轻各家模型的集成都是独立包。python-dotenv用来管理API Key别把Key硬编码在代码里这是基本的安全习惯。环境变量文件.env长这样OPENAI_API_KEY你的key OPENAI_BASE_URL你的接口地址如果你的接口地址是默认的OPENAI_BASE_URL可以不填。代码里用load_dotenv()加载然后ChatOpenAI会自动读取环境变量。4.2 从零搭建一个天气问答器我把完整流程走一遍。第一步定义Schema前面已经写过了。第二步初始化模型和chainimport os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate load_dotenv() llm ChatOpenAI( modelgpt-4o-mini, temperature0, timeout10, max_retries2 ) structured_llm llm.with_structured_output(WeatherAnswer) system_prompt 你是一个专业的天气问答助手。 根据用户的问题提取城市、天气、温度信息并给出穿衣建议。 如果用户没有明确城市默认使用北京。 温度如果用户没提根据天气状况合理估计。 prompt ChatPromptTemplate.from_messages([ (system, system_prompt), (human, {question}) ]) chain prompt | structured_llm第三步调用并处理结果def ask_weather(question: str): try: result chain.invoke({question: question}) return { success: True, data: result.model_dump() } except Exception as e: return { success: False, error: str(e) } print(ask_weather(上海明天冷不冷要穿羽绒服吗))跑出来的结果大概是这样{ success: True, data: { city: 上海, weather: 多云, temperature: 8.0, suggestion: 建议穿羽绒服或厚外套注意保暖 } }注意suggestion这个字段模型不是简单复述而是结合了温度和用户的具体问题要不要穿羽绒服给出的建议。这就是结构化输出配合好的Prompt的威力既有结构又有内容质量。4.3 多轮对话场景的改造单轮跑通之后加多轮支持。核心改动是引入MessagesPlaceholderfrom langchain_core.messages import HumanMessage, AIMessage prompt ChatPromptTemplate.from_messages([ (system, system_prompt), MessagesPlaceholder(variable_namehistory), (human, {question}) ]) chain prompt | structured_llm history [] def ask_with_history(question: str): result chain.invoke({ question: question, history: history }) history.append(HumanMessage(contentquestion)) history.append(AIMessage(contentresult.model_dump_json())) return result这里有个细节历史消息里存的是结构化对象的JSON字符串不是自然语言。为什么因为下一轮模型看到历史时能直接理解上一轮的结构化结果保持格式一致。如果存自然语言模型可能被带偏下一轮又返回自由文本了。实测下来多轮场景下结构化输出的成功率会比单轮略低因为历史消息增加了上下文长度模型注意力被分散。我的应对是限制历史轮数只保留最近3轮超过就丢弃最早的。这样既保留了上下文又控制了复杂度。4.4 参数选择与性能调优几个关键参数的取值建议我整理成表格参数推荐值说明temperature0结构化输出必须为0保证稳定性timeout10秒结构化调用比普通调用慢留足时间max_retries2网络抖动时的自动重试历史轮数3超过会显著增加失败率字段总数≤15超过建议拆分调用嵌套深度≤2再深模型准确率断崖下降关于模型选择我实测过几个主流模型在结构化输出上的表现。小模型如gpt-4o-mini在字段少、结构简单的场景下完全够用速度快、成本低。字段一多、嵌套一深小模型就开始丢字段或者类型搞错这时候得上大模型。所以我的建议是先用小模型跑跑不通再升级别一上来就用最贵的。还有一个容易被忽略的点是Prompt里的字段说明要和Schema里的description呼应。Schema的description是给框架和模型看的Prompt里的说明是给模型看的两者一致能提升准确率。我见过有人Schema写温度摄氏度Prompt里写返回华氏度模型直接懵了。5. 常见问题与排查技巧实录5.1 模型返回字段缺失怎么办这是最高频的问题。表现是Pydantic报错说某个必填字段缺失。排查思路分三步。第一步看Schema里这个字段是不是真的必填如果业务上允许为空改成Optional。第二步看Prompt里有没有明确要求模型填这个字段没有就补上。第三步看字段的description是不是太模糊模型不知道该怎么填。我遇到过一个典型案例字段叫leveldescription写等级模型有时候返回高有时候返回3有时候干脆不返回。后来把description改成等级取值范围1-5的整数1最低5最高问题就解决了。description要具体到模型不用猜这是核心原则。如果三步都做了还是缺那就上重试机制。重试的时候把缺失的字段名明确告诉模型上次返回缺少level字段请补充。实测这个反馈机制很有效。5.2 类型不匹配的排查方法类型不匹配的表现是Pydantic报错说expected int, got str之类。常见原因有两个。一是模型返回了带单位的字符串比如温度返回25℃而不是25。二是模型返回了数字的字符串形式比如25而不是25。解决办法是在description里明确格式要求。比如temperature: float Field(description温度数值只返回数字不要带单位)。另外可以在Pydantic模型上加validator做预处理from pydantic import field_validator class WeatherAnswer(BaseModel): temperature: float Field(description温度数值不带单位) field_validator(temperature, modebefore) classmethod def clean_temperature(cls, v): if isinstance(v, str): v v.replace(℃, ).replace(度, ).strip() return float(v)这个validator会在Pydantic校验前先跑一遍把25℃清洗成25。modebefore是关键表示在类型转换之前执行。这个技巧能救回很多边缘情况。5.3 结构化输出失败的速查表我把常见问题和解决方案整理成表方便对照排查问题现象可能原因解决方案字段缺失description模糊补充具体说明和取值范围类型错误模型加了单位或格式加field_validator预处理返回自由文本模型不支持结构化换支持tool calling的模型嵌套字段丢失嵌套太深拆成多次调用或扁平化多轮后失效历史太长限制历史轮数到3轮偶发超时模型负载高加超时和重试机制中文乱码编码问题确保UTF-8编码5.4 几个我踩过的坑第一个坑是Schema改了就忘了改Prompt。有次我把字段从temp改成temperatureSchema改了Prompt里还写着temp模型就懵了返回的字段名对不上。后来我养成了习惯Schema和Prompt放在一起维护改一个就检查另一个。第二个坑是用中文类名。Pydantic支持中文类名但LangChain转JSON Schema的时候可能出问题而且IDE补全也不方便。老老实实用英文类名description用中文这是最稳的。第三个坑是忽略token消耗。结构化输出的token消耗比普通输出高因为Schema本身要占token模型还要思考字段。我做过对比同样的问答结构化输出的token消耗大概是普通输出的1.5到2倍。如果你的场景量大成本要提前算清楚。第四个坑是在流式输出场景下用结构化。结构化输出本质上是等模型全部生成完才能解析和流式输出天然冲突。如果你既要流式又要结构化得用特殊方案比如先流式输出自然语言给用户看同时后台跑一个结构化调用。这个复杂度不低非必要不建议。6. 结构化问答器的扩展玩法6.1 接入真实数据源做增强前面演示的是模型自己编天气实际项目里肯定要接真实数据。做法是在chain前面加一个工具调用环节先查天气API把结果塞进Prompt再让模型做结构化提取。这样模型不用猜只需要把API返回的数据整理成Schema格式准确率会高很多。具体实现可以用LangChain的Tool或者直接在手写函数里调API。我倾向于手写因为天气API的调用逻辑简单用Tool反而增加了一层抽象。流程是用户提问 - 提取城市 - 调天气API - 把API结果和用户问题一起给模型 - 模型返回结构化对象。6.2 批量处理与并发优化单条调用跑通后如果要做批量比如一次处理100个问题串行调用会非常慢。这时候用chain.batch()方法questions [{question: q} for q in question_list] results chain.batch(questions, config{max_concurrency: 5})max_concurrency控制并发数别设太高容易被限流。我一般设5兼顾速度和稳定性。batch方法内部会自动处理重试和错误比手写循环省心。6.3 把结构化结果存进数据库拿到Pydantic对象后存库很简单result.model_dump()转成dict直接塞进ORM。我用SQLAlchemy举例from sqlalchemy import create_engine, Column, String, Float from sqlalchemy.orm import declarative_base, sessionmaker Base declarative_base() class WeatherRecord(Base): __tablename__ weather id Column(String, primary_keyTrue) city Column(String) weather Column(String) temperature Column(Float) suggestion Column(String) def save(result: WeatherAnswer): record WeatherRecord(**result.model_dump()) session.add(record) session.commit()这里的好处是Pydantic的字段名和数据库列名能直接对应不用手动映射。如果字段名不一致用model_dump(by_aliasTrue)配合Pydantic的alias功能。6.4 后续可以怎么扩展这个问答器的骨架其实可以套用到很多场景。把WeatherAnswer换成别的Schema就是一个新的结构化问答器。比如换成CustomerServiceAnswer字段是intent、entities、response就是一个客服意图识别器。换成ResumeAnswer字段是name、skills、experience就是一个简历解析器。核心不变的是那套链路Pydantic定义契约LangChain负责调用和校验后处理做业务。掌握了这个模式你就能快速复制出各种结构化输出应用。我个人的体会是结构化输出是Agent从玩具走向生产的关键一步因为只有输出可控下游系统才敢用。自由文本的Agent再聪明接不进业务系统也是白搭。最后分享一个小技巧如果你的Schema经常变可以把它抽成配置文件用YAML或者JSON定义运行时动态生成Pydantic模型。这样改字段不用动代码非技术人员也能维护。这个方案我用在一个多租户项目里每个租户有自己的Schema效果很好。
返回列表