ARTICLE DETAIL

资讯详情

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

基于LangChain与Pydantic的Agent结构化输出问答器实战

基于LangChain与Pydantic的Agent结构化输出问答器实战 1. 为什么我要做这个结构化输出问答器做Agent开发的朋友大概率都经历过这样一个阶段一开始用大模型做问答直接让它输出一段自然语言看着挺流畅但一旦要把结果接到下游系统里麻烦就来了。比如你想让模型从一段用户描述里提取“姓名、电话、意向产品”三个字段它可能给你返回一段“好的根据您的描述这位客户叫张三电话是138xxxx对A产品比较感兴趣”——人看着没问题但代码要解析这段文字就得写一堆正则稍微换个措辞就崩了。这就是我做这个“结构化输出问答器”的直接动机。它本质上是一个基于Agent思路构建的问答系统核心目标不是让模型“说得好听”而是让模型稳定地吐出符合预定义Schema的结构化数据。你给它一个问题或者一段文本它返回的不是散文而是一个可以直接被程序消费的JSON对象字段类型、必填项、取值范围都在掌控之中。这个项目适合谁如果你正在学LangChain、刚接触Agent开发、或者被“模型输出格式不稳定”折磨过那这篇内容应该能帮你少走弯路。我会把整个设计思路、Pydantic模型怎么定义、LangChain怎么串起来、踩过哪些坑全部摊开讲。代码可以直接抄思路可以迁移到你自己的业务场景里。关键词先摆出来Agent、LangChain、Pydantic、结构化输出、问答器。这几个词贯穿全文后面每一节都会围绕它们展开。2. 整体设计思路与方案选型2.1 为什么是“结构化输出”而不是“自由文本”先说清楚一个概念。大模型天然擅长生成自由文本但自由文本对程序不友好。结构化输出要解决的核心矛盾就是让模型的输出从“给人看”变成“给机器读”。我举个生活化的类比。你去餐厅点菜跟服务员说“我想吃点清淡的不要太辣最好有蔬菜”这是自由文本。但如果餐厅用的是点菜机你得在屏幕上勾选“口味清淡”“辣度不辣”“品类蔬菜”这就是结构化输出。点菜机不会理解你的模糊表达但它能保证厨房拿到的是标准化的订单。Agent场景下结构化输出的价值更大。因为Agent往往要调用工具、写数据库、触发下游流程每一步都需要精确的参数。如果模型返回的是“我觉得这个用户可能想查询订单状态”而你的工具需要的是{action: query_order, order_id: 12345}那中间就断了。2.2 技术选型LangChain Pydantic 的组合逻辑市面上做结构化输出的方案不少我最终选了LangChain配Pydantic理由有三条。第一Pydantic的Schema定义能力足够强。它本身就是Python生态里做数据校验的主流库支持类型注解、字段约束、嵌套模型、自定义验证器。你定义一个BaseModel子类字段类型写清楚Pydantic会自动帮你校验。模型输出的JSON如果不符合SchemaPydantic会直接报错而不是悄悄放过去。第二LangChain对结构化输出的支持已经比较成熟。它提供了with_structured_output方法可以直接把Pydantic模型绑定到LLM上让模型按照Schema生成。底层它会根据不同的模型提供商选择function calling、JSON mode或者prompt-based的方式来实现。第三Agent编排需要LangChain的链路能力。单纯的问答器不需要Agent但如果你想让问答器具备“先判断问题类型再决定用哪个Schema”的能力就需要Agent的决策逻辑。LangChain的Agent框架可以把这个决策过程串起来。提示如果你用的是比较新的LangChain版本with_structured_output已经是标配。老版本可能需要用PydanticOutputParser配合prompt模板效果类似但代码更啰嗦。2.3 整体架构三层结构我把整个问答器拆成三层这样职责清晰方便调试。第一层是输入层。接收用户的自然语言问题或文本做基本的预处理比如去除多余空白、截断超长输入。第二层是推理层。这是核心LangChain的LLM链在这里工作。它接收输入结合Pydantic Schema的约束生成结构化输出。如果用了Agent模式这一层还会包含“选择哪个Schema”的决策。第三层是校验层。Pydantic模型对LLM的输出做最终校验。校验通过就返回给调用方校验失败就触发重试或降级逻辑。这三层的好处是每一层都可以单独测试。输入层的问题不会污染推理层推理层的格式问题不会绕过校验层。实际调试的时候你能快速定位是哪一层出了岔子。3. 核心细节解析与实操要点3.1 Pydantic模型怎么定义才合理这是整个项目的地基。Schema定义得好后面省一半事定义得烂模型天天给你返回意料之外的东西。先看一个我实际用的例子。假设我要做一个“客户意向提取器”从销售跟客户的聊天记录里提取关键信息from pydantic import BaseModel, Field, field_validator from typing import Optional, List from enum import Enum class ProductInterest(str, Enum): A_PRODUCT A产品 B_PRODUCT B产品 C_PRODUCT C产品 UNKNOWN 未知 class CustomerIntent(BaseModel): 客户意向结构化提取结果 name: str Field(description客户姓名如果未提及则填未知) phone: Optional[str] Field(defaultNone, description联系电话11位数字) product: ProductInterest Field(description意向产品类别) budget: Optional[float] Field(defaultNone, description预算金额单位元) urgency: int Field(ge1, le5, description紧急程度1最低5最高) notes: List[str] Field(default_factorylist, description其他备注要点) field_validator(phone) classmethod def validate_phone(cls, v): if v is None: return v digits .join(filter(str.isdigit, v)) if len(digits) ! 11: raise ValueError(电话号码必须是11位数字) return digits这段代码有几个关键设计点我逐个解释。用Enum约束枚举字段。ProductInterest继承str和Enum这样模型只能从预定义的几个值里选。如果你直接写product: str模型可能返回“A产品”“A类产品”“产品A”各种变体下游根本没法用。Enum把选择空间收窄模型输出的稳定性会大幅提升。Field的description不是装饰。LangChain在构造prompt的时候会把每个字段的description传给模型相当于给模型的填写说明。description写得越清楚模型填得越准。比如phone字段我写了“11位数字”模型就知道不要填成“138-xxxx-xxxx”这种带横杠的格式。Optional和default的配合。不是所有字段都必填。phone和budget用Optional标记默认None。这样模型如果没提取到信息可以留空而不是硬编一个假数据。自定义验证器兜底。validate_phone做了二次清洗把非数字字符去掉再校验长度。这是防御性编程因为模型有时候会“自作聪明”加格式。注意字段的description尽量用中文写因为你的输入大概率是中文模型在中文语境下对中文说明的理解更准确。我试过中英文混写效果不如纯中文。3.2 LangChain怎么绑定结构化输出Schema定义好之后下一步是把它绑到LLM上。LangChain提供了两种主要方式我都用过说说区别。方式一with_structured_outputfrom langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) structured_llm llm.with_structured_output(CustomerIntent) result structured_llm.invoke(客户张三电话13812345678对A产品感兴趣预算大概5万比较急) print(result) # CustomerIntent(name张三, phone13812345678, productProductInterest.A_PRODUCT: A产品, budget50000.0, urgency..., notes[...])这种方式最简洁LangChain会自动处理Schema到模型API的转换。底层它优先用function calling如果模型不支持就退回到JSON mode。方式二PydanticOutputParser Promptfrom langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import ChatPromptTemplate parser PydanticOutputParser(pydantic_objectCustomerIntent) prompt ChatPromptTemplate.from_messages([ (system, 从用户输入中提取客户意向信息。\n{format_instructions}), (human, {input}) ]) chain prompt | llm | parser这种方式更灵活你可以在prompt里加各种约束和示例。缺点是格式说明要自己拼而且模型不一定严格遵守。我实测下来如果模型支持function calling优先用方式一。稳定性明显更好代码也更干净。方式二适合模型能力较弱、或者你需要精细控制prompt的场景。3.3 温度参数和重试策略这两个参数看起来不起眼但对结构化输出的稳定性影响很大。温度temperature。做结构化输出温度一定要调低。我一般设0或者0.1。温度高的时候模型会更“有创造力”但结构化输出恰恰不需要创造力需要的是严格遵守Schema。我试过temperature0.7模型偶尔会把Enum字段填成Schema里没有的值或者把数字字段填成字符串。重试策略。即使温度调到0模型偶尔还是会输出不符合Schema的内容。这时候不能直接报错给用户要有重试机制。LangChain的with_retry可以配置重试次数structured_llm llm.with_structured_output(CustomerIntent).with_retry( stop_after_attempt3 )重试的时候最好把上一次的错误信息也传给模型让它知道哪里错了。LangChain的with_fallbacks可以配合使用如果主模型一直失败降级到备用模型或者返回一个默认结构。实操心得重试次数不要设太多3次足够。超过3次还失败说明要么Schema设计有问题要么输入太离谱再试也是浪费token。4. 实操过程与核心环节实现4.1 环境准备与依赖安装先把环境搭起来。我用的是Python 3.10以上依赖管理用pip或者poetry都行。pip install langchain langchain-openai pydantic python-dotenv如果你用的是其他模型提供商把langchain-openai换成对应的包比如langchain-anthropic、langchain-community等。API密钥通过环境变量管理不要硬编码在代码里# .env 文件 OPENAI_API_KEYyour_key_herefrom dotenv import load_dotenv load_dotenv()4.2 完整代码实现下面是一个可以直接跑的完整示例。我把它拆成几个函数方便你按需修改。import os from typing import Optional, List from enum import Enum from pydantic import BaseModel, Field, field_validator from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from dotenv import load_dotenv load_dotenv() # ---------- 1. 定义Schema ---------- class ProductInterest(str, Enum): A_PRODUCT A产品 B_PRODUCT B产品 C_PRODUCT C产品 UNKNOWN 未知 class CustomerIntent(BaseModel): name: str Field(description客户姓名未提及填未知) phone: Optional[str] Field(defaultNone, description11位手机号) product: ProductInterest Field(description意向产品) budget: Optional[float] Field(defaultNone, description预算单位元) urgency: int Field(ge1, le5, description紧急程度1-5) notes: List[str] Field(default_factorylist, description备注要点) field_validator(phone) classmethod def clean_phone(cls, v): if v is None: return v digits .join(filter(str.isdigit, v)) if len(digits) ! 11: raise ValueError(手机号必须11位) return digits # ---------- 2. 构建链 ---------- def build_chain(): llm ChatOpenAI(modelgpt-4o-mini, temperature0) structured_llm llm.with_structured_output(CustomerIntent).with_retry( stop_after_attempt3 ) return structured_llm # ---------- 3. 问答器主逻辑 ---------- class StructuredQnA: def __init__(self): self.chain build_chain() def ask(self, text: str) - CustomerIntent: 输入自然语言返回结构化结果 try: result self.chain.invoke(text) return result except Exception as e: print(f结构化提取失败: {e}) # 降级返回一个空结构 return CustomerIntent( name未知, productProductInterest.UNKNOWN, urgency1, notes[f解析失败: {str(e)}] ) # ---------- 4. 使用 ---------- if __name__ __main__: qna StructuredQnA() text 客户李四电话是139-8765-4321想了解B产品预算8万左右比较着急 result qna.ask(text) print(result.model_dump_json(indent2))跑出来的结果大概是这样{ name: 李四, phone: 13987654321, product: B产品, budget: 80000.0, urgency: 4, notes: [比较着急] }注意phone字段输入是“139-8765-4321”经过验证器清洗后变成了纯数字。urgency模型根据“比较着急”判断为4这个判断是模型做的不是硬编码的。4.3 参数选择与计算过程有几个参数需要根据实际情况调整我说说我的选择依据。模型选择。我用的gpt-4o-mini性价比高结构化输出能力够用。如果你对准确率要求极高可以上gpt-4o或者claude-3-5-sonnet。实测下来mini版本在简单Schema上的准确率能到95%以上复杂嵌套Schema会降到85%左右。temperature0。前面说过了结构化输出不需要创造力。设0之后同样的输入基本能得到同样的输出可复现性强。max_tokens。这个要根据Schema的复杂度来估。一个6字段的简单Schema输出JSON大概200-400 token。我一般设1024留足余量。如果Schema嵌套很深设2048。重试次数3次。第一次失败可能是偶发第二次失败可能是输入问题第三次还失败就是Schema或者模型的问题了。再重试边际收益很低。4.4 从单Schema到多Schema的Agent化单Schema的问答器已经能解决很多问题但实际业务里往往有多个Schema。比如一个客服系统可能要处理“订单查询”“退款申请”“产品咨询”三类问题每类对应不同的Schema。这时候就需要Agent的决策能力。我的做法是加一个路由层class IntentRouter(BaseModel): 判断用户意图属于哪一类 category: str Field(description订单查询/退款申请/产品咨询/其他) def route_and_extract(text: str): # 第一步路由 router_llm llm.with_structured_output(IntentRouter) intent router_llm.invoke(text) # 第二步根据路由结果选择Schema schema_map { 订单查询: OrderQuery, 退款申请: RefundRequest, 产品咨询: ProductInquiry, } target_schema schema_map.get(intent.category, GeneralInquiry) # 第三步用对应Schema提取 extractor llm.with_structured_output(target_schema) return extractor.invoke(text)这个模式的好处是每个Schema可以独立优化互不干扰。路由层用最简单的Schema只做分类准确率高。提取层用复杂Schema专注字段填充。实操心得路由层的分类不要超过5类。类别太多模型容易混淆。如果业务确实复杂可以做两级路由先分大类再分小类。5. 常见问题与排查技巧实录5.1 模型返回的字段类型不对怎么办这是最常见的问题。比如Schema里定义budget: float模型返回五万或者50000元。排查思路先看Pydantic有没有报错。如果报错信息是Input should be a valid number说明模型返回了字符串。这时候有两个解决方向。一是在description里写清楚格式。把description预算单位元改成description预算金额纯数字单位元例如50000。给模型一个示例它更容易理解。二是加验证器做转换。写一个field_validator把“五万”这种中文数字转成50000。不过中文数字转换比较麻烦更稳妥的做法是让模型自己转验证器只做兜底。5.2 Enum字段返回了Schema外的值比如ProductInterest只定义了A、B、C、未知四个值模型返回了“D产品”。原因模型可能从输入里看到了“D产品”这个词但你的Enum里没有它就硬填了。解决在Enum里加一个OTHER 其他作为兜底。同时在description里强调“只能从以下选项中选择”。如果模型还是乱填说明输入里确实有Schema覆盖不到的信息这时候应该考虑扩展Schema而不是怪模型。5.3 嵌套Schema的准确率下降单层Schema准确率95%嵌套两层可能就降到80%了。原因嵌套结构对模型来说更复杂它需要同时维护多个层级的上下文。解决尽量扁平化Schema。如果业务允许把嵌套结构拆成多个独立的Schema分步提取。比如先提取订单信息再提取客户信息最后合并。虽然多了一次调用但准确率会明显提升。5.4 常见问题速查表问题现象可能原因排查方向解决方案Pydantic校验报错模型输出类型不符看报错字段和实际值改description加示例或加验证器Enum值超出范围Schema覆盖不全检查输入是否含未定义类别加OTHER兜底或扩展Enum必填字段为空模型没提取到检查输入是否真的包含该信息改Optional或给默认值重试多次仍失败Schema太复杂看失败集中在哪个字段拆分Schema分步提取输出不稳定temperature太高检查temperature设置调到0或0.1中文乱码编码问题检查输入输出编码统一用UTF-85.5 独家避坑技巧技巧一先用小模型试Schema。Schema设计好之后先用mini模型跑一批测试数据。如果mini能跑通大模型肯定没问题。如果mini跑不通先别急着换大模型大概率是Schema本身有问题。技巧二给模型看例子。在description里加一两个示例比写一堆约束管用。比如description紧急程度1-51是不急5是非常急模型一看就懂。技巧三日志要记全。每次调用都把输入、输出、耗时、是否重试记下来。出问题的时候这些日志是排查的唯一依据。我一般用JSON Lines格式记日志方便后续分析。技巧四Schema版本管理。Schema改了之后老数据可能不兼容。我习惯在Schema里加一个version字段或者用文件名区分版本。这样回溯的时候不会乱。技巧五不要追求100%准确。结构化输出做到95%以上就很好了剩下的5%用降级逻辑兜底。追求100%的代价是指数级上升的不划算。6. 这个问答器还能怎么扩展基础版本跑通之后我试过几个扩展方向效果不错分享给你。扩展一批量处理。把单条输入改成列表用batch方法批量调用。LangChain的batch会自动并发速度比循环快很多。注意控制并发数别把API限流了。扩展二缓存层。同样的输入没必要重复调用模型。我用functools.lru_cache做了个简单的内存缓存命中率大概30%省了不少token。扩展三人工审核队列。对于校验失败或者置信度低的结果不直接返回而是推到审核队列。人工确认后再入库。这个在金融、医疗等对准确率要求高的场景很有用。扩展四Schema自动生成。如果你的数据源是数据库表可以用代码自动生成Pydantic模型。这样表结构变了Schema自动跟着变不用手写。扩展五多语言支持。把description改成英文输入输出都走英文可以支持多语言场景。不过中文场景下中文description的效果确实更好这个我对比过。我个人在实际操作中的体会是结构化输出问答器的核心价值不在于模型多强而在于Schema设计得合不合理。Schema是人和模型之间的契约契约写得清楚模型就守规矩契约写得模糊模型就自由发挥。把Schema当成产品需求文档来写每个字段都问自己“这个字段下游怎么用”想清楚了再动手后面能省大量调试时间。最后再分享一个小技巧如果你不确定某个字段该不该设成必填先设成Optional跑一段时间看看模型实际填充率。填充率高的字段再改成必填填充率低的就保持Optional。用数据说话比拍脑袋靠谱。
返回列表