资讯动态

LangChain+Pydantic结构化输出问答器:Agent开发实战

发布时间:2026/10/8 20:45:29 来源:尧图企业网站定制
1. 为什么我要做这个结构化输出问答器做Agent开发的朋友大概率都经历过这样一个阶段一开始用大模型做问答直接让它输出一段自然语言看着挺像回事但一旦要把结果接到下游系统里麻烦就来了。比如你想让模型从一段用户描述里提取“姓名、电话、意向产品”三个字段它可能给你返回“好的这位客户叫张三电话是138xxxx对A产品比较感兴趣”你还得再写正则去抠。抠得准不准先不说字段一多、格式一变维护成本直接爆炸。这就是结构化输出要解决的问题。所谓结构化输出就是让大模型不再返回一段自由文本而是返回一个严格符合预定义Schema的数据对象比如JSON。你定义好字段名、类型、是否必填模型就必须按这个格式吐出来。这样一来下游代码可以直接反序列化成对象不用再做脆弱的字符串解析。我这次做的“Agent实践4-结构化输出问答器”核心目标很明确用LangChain做编排用Pydantic定义输出结构做一个能稳定返回结构化数据的问答器。它适合谁参考如果你正在学Agent开发已经会调用大模型API但还没搞明白怎么让模型输出可控的数据那这篇内容就是给你写的。如果你已经在做LangChain项目想找一个能直接抄作业的结构化输出方案也可以直接拿走。关键词里提到的Agent、LangChain、Pydantic、结构化输出、问答器这五个词基本就是整个项目的骨架。下面我会从设计思路、核心细节、实操过程、问题排查几个角度把这个问答器拆开讲清楚。2. 整体设计与技术选型拆解2.1 为什么选LangChain而不是裸调API裸调大模型API当然也能做结构化输出。你可以在Prompt里写“请返回JSON格式”然后在代码里用json.loads解析。但这种方式有几个坑第一模型不一定听话有时候会加一句“好的以下是结果”导致JSON解析失败第二字段类型没法约束你让它返回数字它可能返回字符串“25”第三字段缺失时没有统一处理机制。LangChain在这件事上的价值在于它提供了with_structured_output这类封装把“让模型按Schema输出”这件事标准化了。你只需要传入一个Pydantic模型LangChain会自动处理Prompt构造、输出解析、类型校验。底层它可能用Function Calling也可能用JSON Mode具体取决于你用的模型但对你来说接口是一致的。我选LangChain还有一个现实原因生态成熟。你后面要加记忆、加工具调用、加多轮对话LangChain都有现成组件。如果一开始裸调API后面扩展时还得自己造轮子。2.2 Pydantic在这里扮演什么角色Pydantic是这个项目的“契约层”。你定义一个类声明字段名、类型、描述、默认值Pydantic就帮你做校验。比如from pydantic import BaseModel, Field class CustomerInfo(BaseModel): name: str Field(description客户姓名) phone: str Field(description联系电话) product: str Field(description意向产品) budget: float Field(description预算金额单位元)这个类一定义LangChain就知道要让模型输出什么结构。模型返回后Pydantic会自动校验name是不是字符串budget是不是浮点数缺字段会报错。你拿到的是一个CustomerInfo对象直接.name就能访问不用再result[name]。这里有个细节值得说Field里的description不是写给人看的是写给模型看的。模型会根据这个描述理解字段含义。所以描述要写清楚比如“预算金额单位元”就比“预算”好能减少模型把“五千”返回成字符串“五千”而不是数字5000的情况。2.3 问答器的整体架构整个问答器的数据流是这样的用户输入一段自然语言问题或描述。LangChain把输入和Pydantic Schema一起发给大模型。模型返回符合Schema的结构化数据。Pydantic做校验LangChain做解析。你拿到结构化对象可以存库、可以展示、可以传给下游。这个架构看起来简单但每一步都有细节。比如第2步Schema怎么传给模型不同模型支持程度不一样第3步模型返回的JSON可能有多余字段Pydantic默认会忽略还是报错需要配置第4步校验失败时怎么重试LangChain有没有内置机制。我选这个架构的理由是它把“模型能力”和“数据契约”解耦了。模型可以换Schema不用动Schema可以改模型调用代码不用大改。这种解耦在Agent开发里很重要因为模型迭代太快今天用这个明天用那个如果业务代码和模型绑死迁移成本会很高。2.4 和其他Agent框架的对比热词里提到了LangChain、Dify、CrewAI等框架。我简单说下我的看法Dify更适合低代码场景拖拽式编排结构化输出也能做但定制性不如代码框架CrewAI强在多Agent协作如果你只是做一个单Agent问答器用它有点重LangChain的优势是灵活底层可控适合想深入理解Agent运行机制的开发者。如果你只是想快速搭一个DemoDify可能更快但如果你想搞明白结构化输出到底怎么实现的LangChain是更好的学习路径。我这个项目选LangChain就是因为它能让我把每个环节都拆开看。3. 核心细节解析与实操要点3.1 Pydantic模型设计的五个关键点设计Pydantic模型不是随便写几个字段就完事有几个点直接影响输出稳定性。第一字段类型要尽量具体。str比Any好float比str好。如果你用Any模型可能返回嵌套字典你后面处理起来很麻烦。如果预算字段你用str模型可能返回“5000元”你还得清洗。用float模型就知道要返回数字。第二必填和可选要分清。Pydantic里字段没有默认值就是必填有默认值就是可选。必填字段如果模型没返回会直接报错。所以你要想清楚哪些字段是必须的哪些可以缺。比如“姓名”必填“备注”可以可选。第三描述要写清楚。前面说了description是给模型看的。描述里可以写格式要求比如“日期格式YYYY-MM-DD”可以写枚举值比如“产品类型只能是A、B、C之一”。描述越清楚模型输出越准。第四嵌套结构要控制深度。Pydantic支持嵌套模型比如CustomerInfo里再套一个Address。嵌套能表达复杂结构但太深了模型容易出错。我一般控制在两层以内。第五枚举类型用Literal或Enum。如果你希望某个字段只能是固定几个值用Literal[A, B, C]模型就不会返回“D”。这比在描述里写“只能是A、B、C”更可靠。3.2 LangChain结构化输出的两种模式LangChain的with_structured_output支持多种模式常见的有两种Function Calling和JSON Mode。Function Calling模式依赖模型本身的函数调用能力。你把Pydantic模型转成函数定义模型返回函数调用参数LangChain再解析成对象。这种模式的好处是模型对Schema的理解更准因为函数定义是模型训练时就见过的格式。缺点是有些模型不支持或者支持得不好。JSON Mode模式是让模型直接返回JSON字符串LangChain用Pydantic解析。这种模式兼容性更好但模型可能返回多余字段或格式错误。用JSON Mode时我建议在Prompt里明确写“只返回JSON不要有其他内容”并且设置strictTrue让Pydantic严格校验。我实测下来如果模型支持Function Calling优先用Function Calling稳定性明显更好。如果不支持再用JSON Mode但要加校验和重试。3.3 参数配置与温度设置结构化输出场景下温度参数很关键。温度高模型发挥空间大但格式容易飘温度低输出稳定但可能过于死板。我的经验是结构化输出任务温度设0到0.3之间。我一般用0因为我要的是稳定不是创意。还有一个参数是max_tokens。如果你Schema字段多输出JSON会比较长max_tokens设太小会导致JSON被截断解析失败。我一般设2000以上具体看字段数量。字段多的话可以设4000。另外如果你用的是OpenAI系列模型可以关注response_format参数。设置成{type: json_object}能强制模型返回JSON。但注意这个模式下你必须在Prompt里提到“JSON”这个词否则模型可能报错。3.4 校验失败的处理策略即使做了这么多准备模型还是可能返回不符合Schema的数据。这时候怎么办第一捕获异常。LangChain解析失败会抛OutputParserException你要用try-except包住。第二重试。可以设置重试次数比如失败后重新调用模型。LangChain有with_retry方法可以配置重试策略。第三降级。如果重试几次还是失败可以降级到自由文本输出或者返回一个默认对象避免整个流程崩溃。第四记录日志。失败案例要记下来分析是Schema设计问题还是模型问题。我一般会把失败的Prompt和输出都存下来定期复盘。注意重试不是万能的。如果模型本身能力不够重试十次也没用。这时候要考虑换模型或者简化Schema。4. 完整实操过程与核心环节实现4.1 环境准备与依赖安装先准备环境。我用的Python 3.10太老的版本可能不支持一些新特性。pip install langchain langchain-openai pydantic如果你用其他模型比如国内的一些模型需要装对应的LangChain集成包。比如用智谱的模型可以装langchain-community然后配置对应的API。环境变量里配好API Keyexport OPENAI_API_KEY你的key如果你用的是兼容OpenAI接口的其他服务还需要配OPENAI_BASE_URL。4.2 定义Pydantic输出模型我以一个“客户意向提取”场景为例。用户输入一段描述问答器提取结构化信息。from pydantic import BaseModel, Field from typing import Literal, Optional class CustomerIntent(BaseModel): 客户意向信息 name: str Field(description客户姓名) phone: str Field(description联系电话11位数字) product: Literal[A产品, B产品, C产品] Field( description意向产品只能是A产品、B产品、C产品之一 ) budget: float Field(description预算金额单位元只返回数字) urgency: Literal[高, 中, 低] Field( description紧急程度只能是高、中、低之一 ) remark: Optional[str] Field(defaultNone, description备注信息可选)这个模型里Literal限制了产品类型和紧急程度float约束了预算Optional让备注可选。描述里写清楚了格式要求。4.3 构建LangChain结构化输出链接下来构建链。我用的是ChatOpenAI如果你用其他模型替换对应的类即可。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm ChatOpenAI( modelgpt-4o-mini, temperature0, max_tokens2000 ) structured_llm llm.with_structured_output(CustomerIntent) prompt ChatPromptTemplate.from_messages([ (system, 你是一个信息提取助手。从用户描述中提取客户意向信息严格按照要求的格式输出。), (human, {input}) ]) chain prompt | structured_llm这里with_structured_output(CustomerIntent)是关键它把LLM包装成返回CustomerIntent对象的组件。prompt | structured_llm是LangChain的管道语法输入经过Prompt格式化后传给结构化LLM。4.4 运行与结果验证跑一个测试result chain.invoke({ input: 我叫张三电话13812345678想了解A产品预算大概5000块比较急。 }) print(type(result)) print(result.name) print(result.phone) print(result.product) print(result.budget) print(result.urgency)输出应该是class __main__.CustomerIntent 张三 13812345678 A产品 5000.0 高你可以看到result直接就是CustomerIntent对象不是字典不是JSON字符串。budget是浮点数5000.0不是字符串“5000块”。urgency是“高”不是“比较急”。这就是结构化输出的价值。4.5 批量处理与异步调用实际项目里往往要批量处理。LangChain支持batch方法inputs [ {input: 李四13900001111B产品预算1万不急}, {input: 王五13700002222C产品预算3000中等紧急} ] results chain.batch(inputs) for r in results: print(r.name, r.product, r.budget)如果要异步用abatchresults await chain.abatch(inputs)批量处理时注意并发限制别一次性发太多请求容易被限流。我一般控制在10到20个并发。4.6 接入FastAPI做成服务如果你想把它做成一个HTTP服务可以接FastAPIfrom fastapi import FastAPI from pydantic import BaseModel app FastAPI() class QueryRequest(BaseModel): text: str app.post(/extract) async def extract(req: QueryRequest): result await chain.ainvoke({input: req.text}) return result.model_dump()model_dump()把Pydantic对象转成字典FastAPI会自动序列化成JSON返回。这样前端或其他服务就能直接调用。提示生产环境要加超时和重试。模型调用可能超时FastAPI默认没有超时设置建议用asyncio.wait_for包一层。5. 常见问题与排查技巧实录5.1 模型返回字段缺失怎么办这是最常见的问题。模型可能漏掉某个字段尤其是可选字段和必填字段混在一起时。排查思路先看Prompt里有没有明确要求所有必填字段。如果Prompt没问题看Schema描述是否清楚。如果描述也清楚可能是模型能力问题。解决方法第一把必填字段在描述里标出来比如“姓名必填”。第二用Pydantic的Field(..., min_length1)加约束。第三如果还是缺考虑换模型或加Few-shot示例。我踩过的坑有一次备注字段设成可选模型就经常不返回。后来我在Prompt里写“即使没有备注也返回空字符串”就稳定了。5.2 JSON解析失败怎么排查JSON解析失败通常有几个原因模型返回了多余文本、JSON被截断、字段类型不对。排查步骤打印原始输出。LangChain解析失败时异常里通常有原始输出先看模型到底返回了什么。检查max_tokens。如果输出被截断JSON不完整解析必失败。调大max_tokens。检查Prompt。如果Prompt里没强调“只返回JSON”模型可能加解释性文字。检查Schema。如果Schema太复杂模型可能生成错误JSON。我一般会在Prompt里加一句“只返回JSON对象不要包含任何其他文字、标记或解释。”这句话能减少很多问题。5.3 类型不匹配怎么处理模型可能把数字返回成字符串把枚举值返回成近义词。比如预算返回“5000元”紧急程度返回“比较急”。处理方法第一用Literal限制枚举模型就不敢返回“比较急”。第二用Pydantic的field_validator做清洗比如把“5000元”转成5000.0。第三在描述里写清楚格式比如“只返回数字不要带单位”。from pydantic import field_validator class CustomerIntent(BaseModel): budget: float Field(description预算金额单位元只返回数字) field_validator(budget, modebefore) classmethod def parse_budget(cls, v): if isinstance(v, str): v v.replace(元, ).replace(块, ).strip() return float(v) return v这个validator在Pydantic校验前执行能把字符串清洗成浮点数。5.4 常见问题速查表问题现象可能原因解决方法字段缺失Prompt未强调必填在描述和Prompt中标注必填JSON解析失败输出被截断调大max_tokensJSON解析失败模型加了多余文字Prompt强调只返回JSON类型不匹配模型返回字符串用field_validator清洗枚举值错误未用Literal限制改用Literal或Enum输出不稳定温度过高温度设为0到0.3调用超时网络或模型负载加超时和重试机制5.5 独家避坑技巧第一个技巧Schema字段不要太多。我试过一个Schema有20多个字段模型输出错误率明显上升。后来拆成两个Schema分两次调用稳定性好很多。字段数量建议控制在10个以内。第二个技巧用Few-shot示例。如果某个字段模型总是理解错在Prompt里加一两个示例告诉它“输入是这样输出应该是这样”。这比改描述更有效。第三个技巧日志要记全。每次调用的输入、输出、耗时、是否成功都记下来。出问题时能快速定位。我用的是Python的logging模块输出到文件按天切分。第四个技巧版本锁定。LangChain和Pydantic更新都很快有时候新版本会改行为。生产环境一定要锁定版本比如langchain0.3.0避免自动升级导致意外。第五个技巧测试集要覆盖边界情况。我建了一个测试集包含正常输入、缺字段输入、格式混乱输入、超长输入。每次改Prompt或Schema都跑一遍测试集看通过率有没有下降。6. 性能优化与扩展方向6.1 减少Token消耗的几种做法结构化输出因为要传SchemaToken消耗比普通问答高。优化方法有几个第一精简Schema描述。描述写清楚就行不用写太长。我见过有人把描述写成一段话Token浪费严重。第二用更小的模型。如果任务简单gpt-4o-mini就够不用上gpt-4。我实测下来字段提取任务mini和4的准确率差距不大但成本差很多。第三缓存。如果同样的输入反复出现可以加缓存。LangChain有set_llm_cache可以配内存缓存或Redis缓存。第四批量处理。前面说的batch方法比循环单次调用效率高因为可以并行。6.2 加记忆和多轮对话现在的问答器是单轮的输入一段文本输出结构化数据。如果要支持多轮比如用户先说了姓名再说电话需要加记忆。LangChain有ConversationBufferMemory可以存对话历史。但结构化输出场景下记忆要小心处理因为历史消息可能干扰Schema解析。我的做法是只把历史输入拼接到当前输入里不把历史结构化输出放回去。from langchain_core.messages import HumanMessage, AIMessage history [] def chat(user_input): messages history [HumanMessage(contentuser_input)] result structured_llm.invoke(messages) history.append(HumanMessage(contentuser_input)) history.append(AIMessage(contentresult.model_dump_json())) return result这样模型能看到之前的对话但输出仍然是结构化的。6.3 接入工具调用如果问答器需要查数据库或调API可以加工具。LangChain的Agent可以绑定工具模型决定什么时候调用。但结构化输出和工具调用结合时要注意工具调用返回的结果可能不是结构化的需要再经过一次结构化输出。我的做法是分两步第一步让Agent调工具拿数据第二步用结构化LLM把数据整理成Schema。6.4 监控与告警生产环境要加监控。我监控几个指标调用成功率、平均耗时、Token消耗、解析失败率。成功率低于95%就告警解析失败率高于5%就排查。监控工具可以用Prometheus加Grafana简单点用日志加定时脚本也行。我一开始用日志后来数据多了换成Prometheus看趋势更方便。7. 我在这件事上的个人体会这个问答器我前后迭代了三个版本。第一版裸调API用正则解析字段一多就崩。第二版用LangChain加Pydantic稳定了很多但Schema设计不合理模型经常返回错误枚举值。第三版加了Literal限制和field_validator才算真正能用。最大的体会是结构化输出的稳定性三分靠模型七分靠Schema设计和Prompt。模型能力固然重要但你把Schema设计好、描述写清楚、边界情况处理好即使用小模型也能得到不错的效果。反过来Schema设计得乱七八糟用再好的模型也白搭。另一个体会是不要追求一次到位。我一开始想做一个大而全的Schema覆盖所有可能字段结果模型输出错误率很高。后来拆成多个小Schema按场景调用反而更稳定。这跟写代码一样单一职责原则在Schema设计上也适用。最后分享一个小技巧如果你不确定某个字段该用什么类型先用str跑一段时间看模型实际返回什么再根据实际情况调整类型。比如预算字段我一开始用str发现模型有时返回“5000”有时返回“5000元”后来改成float加validator就统一了。这个迭代过程比一开始就追求完美更实际。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑