ARTICLE DETAIL

资讯详情

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

LangChain结构化输出实战:用Pydantic让Agent输出可编程

LangChain结构化输出实战:用Pydantic让Agent输出可编程 1. 为什么我要做结构化输出问答器做Agent开发的人都有一个共同的痛点大模型返回的内容太“自由”了。你问它一个问题它给你洋洋洒洒写一大段看起来什么都说了但你想把它塞进下游系统里——比如存数据库、调API、做条件判断——就发现根本没法用。字符串解析正则匹配稍微换个模型或者换个问法格式就全乱了。我踩过这个坑。之前做一个客服工单自动分类的Agent需要模型返回“分类标签置信度摘要”三个字段。一开始用提示词硬约束让模型按JSON格式输出。测试的时候好好的一上生产就翻车有时候多一个逗号有时候字段名拼错有时候干脆给你包一层markdown代码块。后来我算了一笔账光是处理格式异常的兜底逻辑就写了三百多行维护成本高得离谱。这个项目就是来解决这个问题的。结构化输出问答器核心思路很简单用Pydantic定义好你期望的数据结构让LangChain的Agent在生成回答时直接按照这个Schema来输出拿到手的就是一个类型安全、字段完整的Python对象不需要再做任何字符串解析。它解决的是Agent从“能聊天”到“能干活”之间最关键的一环——让模型的输出可编程。适合谁来参考如果你正在用LangChain做Agent开发或者你手头有任何一个需要把大模型输出接入下游系统的场景这个方案可以直接抄作业。哪怕你刚入门LangChain跟着走一遍也能理解结构化输出的完整链路。2. 整体设计思路与方案选型2.1 为什么是Pydantic而不是JSON SchemaLangChain支持多种结构化输出的方式最原始的是在提示词里写“请以JSON格式返回”高级一点的是用response_format参数指定JSON Schema。但我最终选了Pydantic原因有三个。第一类型安全。Pydantic的BaseModel定义出来的字段是有类型的age: int就是整数name: str就是字符串。模型返回的数据经过Pydantic校验后你在代码里拿到的是真正的Python对象IDE能给你补全类型检查工具能帮你发现错误。JSON Schema只是一份文档它不提供运行时保障。第二校验能力强。Pydantic内置了丰富的校验器你可以限制字符串长度、数值范围、正则匹配甚至写自定义校验函数。比如我要求“置信度必须在0到1之间”直接在Field里写ge0, le1就行。模型如果返回了1.5Pydantic会直接抛异常我就能捕获并重试。第三和LangChain的集成度最高。LangChain的with_structured_output()方法原生支持Pydantic模型底层会自动把Pydantic的Schema转换成模型能理解的格式并且处理了不同模型提供商的差异。用JSON Schema的话你得自己处理这些兼容性问题。2.2 问答器的核心架构整个问答器的架构分三层。最底层是模型层负责调用大模型中间是结构化层用Pydantic定义输出格式并做校验最上层是业务层处理校验失败后的重试、降级和日志记录。为什么要把这三层分开因为实际生产中模型调用可能失败结构化校验也可能失败这两类问题的处理策略完全不同。模型调用失败通常是网络问题或限流重试就行结构化校验失败说明模型没理解格式要求需要调整提示词或者换模型。分层之后每层的职责清晰排查问题的时候一眼就能定位。2.3 工具选型的取舍LangChain的版本迭代很快我选的是0.3.x的稳定版。为什么不追最新版因为Agent相关的API在0.2到0.3之间有过一次大的重构很多网上的教程还是老版本的写法直接抄会报错。0.3.x的API已经稳定下来了with_structured_output的用法在后续版本中应该不会有破坏性变更。Pydantic我用的V2版本。V1和V2的API差异很大比如validator变成了field_validatorConfig类变成了model_config字典。网上很多LangChain的示例代码还是V1的写法如果你直接复制粘贴会遇到一堆弃用警告甚至报错。我的建议是直接用V2虽然学习成本高一点但性能更好而且LangChain新版本已经全面适配V2了。3. 核心细节解析与实操要点3.1 Pydantic模型的定义技巧定义Pydantic模型看起来简单但有几个细节直接决定了结构化输出的成功率。字段描述一定要写清楚。很多人定义模型的时候只写字段名和类型比如category: str然后指望模型能猜出来这个字段该填什么。模型不是人它需要明确的指令。正确的做法是用Field的description参数详细说明每个字段的含义、取值范围和示例。from pydantic import BaseModel, Field class QAResponse(BaseModel): 问答器的结构化输出模型 answer: str Field( description对用户问题的直接回答要求简洁准确不超过200字 ) confidence: float Field( description回答的置信度0到1之间的小数越高表示越确定, ge0.0, le1.0 ) sources: list[str] Field( description回答所依据的知识来源列表没有来源时返回空列表, default_factorylist ) category: str Field( description问题所属类别只能是以下之一技术、产品、售后、其他 )注意category字段的描述里我用了“只能是以下之一”这样的限定语。实测下来加上这句话之后模型返回预期类别之外值的概率从大概15%降到了3%以下。模型对枚举值的理解能力有限你给它一个明确的候选列表它才能准确命中。还有一个技巧是用Literal类型代替str来做枚举约束。Pydantic的Literal[技术, 产品, 售后, 其他]会在校验阶段直接拒绝不在列表中的值比在描述里写“只能是以下之一”更可靠。但缺点是如果模型返回了列表外的值你会得到一个校验异常需要处理重试。我的做法是两者结合用Literal做硬约束同时在描述里也写清楚候选值降低模型出错的概率。3.2 with_structured_output的底层机制LangChain的with_structured_output()方法做的事情比表面看起来要多。它接收一个Pydantic模型作为参数然后返回一个新的Runnable这个Runnable的输入输出和原来的模型一样但输出会被强制转换成你定义的Pydantic对象。底层实现上LangChain会根据你使用的模型提供商选择不同的策略。对于OpenAI系列的模型它会使用Function Calling或者JSON Mode对于Anthropic的模型它会使用Tool Use对于开源模型它可能会用提示词工程的方式。这些差异被LangChain封装起来了你不需要关心。但有一个坑需要注意不是所有模型都支持结构化输出。如果你用的模型不支持Function Callingwith_structured_output()会退化成基于提示词的方式成功率会大幅下降。我在测试的时候用过一个国产的开源模型结构化输出的成功率只有60%左右换成GPT-4o之后直接到了98%以上。所以选模型的时候一定要确认它支持结构化输出相关的能力。3.3 提示词与结构化输出的配合用了with_structured_output()之后提示词还需要写格式要求吗我的经验是要写但不用写得太细。LangChain会自动把Pydantic的Schema信息注入到提示词里你不需要再手动描述JSON格式。但你可以补充一些业务层面的约束。比如我会在系统提示词里写“你是一个专业的问答助手回答用户问题时请基于事实不确定的内容要降低置信度。如果问题涉及多个类别选择最相关的一个。”这些是Schema本身表达不了的业务规则需要靠提示词来传达。另一个技巧是在提示词里给一两个示例。Few-shot对结构化输出的帮助很大尤其是当你的Schema比较复杂的时候。示例不需要多一两个就够了但一定要覆盖边界情况。比如我会给一个“置信度很低”的示例让模型知道什么情况下应该输出低置信度。4. 完整实操流程与核心代码实现4.1 环境准备与依赖安装先把环境搭起来。我用的Python版本是3.11太老的版本可能不支持Pydantic V2的一些特性。pip install langchain0.3.7 pip install langchain-openai0.2.8 pip install pydantic2.9.2 pip install python-dotenv1.0.1langchain-openai是LangChain拆出来的OpenAI集成包0.3版本之后LangChain把各个模型提供商的集成都拆成了独立的包这样核心包更轻量。如果你用的是其他模型比如Anthropic或者国内的通义千问需要安装对应的集成包。API密钥用环境变量管理不要硬编码在代码里。我习惯用.env文件加python-dotenv的方式本地开发方便上线的时候换成环境变量注入就行。# .env 文件内容 OPENAI_API_KEY你的密钥 OPENAI_BASE_URL你的接口地址4.2 定义结构化输出模型根据问答器的业务需求我定义了四个字段回答内容、置信度、来源列表和问题类别。为什么是这四个回答内容是核心产出置信度用于下游做质量过滤来源列表用于追溯和审计问题类别用于路由和统计。这四个字段覆盖了问答场景的主要需求。from pydantic import BaseModel, Field from typing import Literal class QAResponse(BaseModel): 问答器的结构化输出模型 answer: str Field( description对用户问题的直接回答要求简洁准确不超过200字 ) confidence: float Field( description回答的置信度0到1之间的小数越高表示越确定, ge0.0, le1.0 ) sources: list[str] Field( description回答所依据的知识来源列表没有来源时返回空列表, default_factorylist ) category: Literal[技术, 产品, 售后, 其他] Field( description问题所属类别从技术、产品、售后、其他中选择一个 )这里category用了Literal类型Pydantic会在校验时强制检查值是否在允许的范围内。sources用了default_factorylist这样模型不返回这个字段时默认为空列表不会报错。4.3 构建问答器Agent问答器的核心逻辑是接收用户问题调用模型返回结构化结果。我用LangChain的表达式语言来串联各个组件。import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate load_dotenv() # 初始化模型 llm ChatOpenAI( modelgpt-4o, temperature0, api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL) ) # 绑定结构化输出 structured_llm llm.with_structured_output(QAResponse) # 定义提示词 system_prompt 你是一个专业的问答助手。回答用户问题时请遵循以下规则 1. 基于事实回答不确定的内容要降低置信度 2. 如果问题涉及多个类别选择最相关的一个 3. 回答要简洁控制在200字以内 4. 如果引用了知识来源在sources字段中列出 prompt ChatPromptTemplate.from_messages([ (system, system_prompt), (human, {question}) ]) # 组装链 qa_chain prompt | structured_llmtemperature0是为了让输出更稳定。结构化输出场景下创造性不是我们需要的一致性才是。实测下来temperature设为0时同一个问题多次调用的结构化输出成功率接近100%设为0.7时会降到90%左右。4.4 调用与结果处理调用链的时候返回的直接就是QAResponse对象不需要任何解析。def ask_question(question: str) - QAResponse: 向问答器提问返回结构化结果 try: result qa_chain.invoke({question: question}) return result except Exception as e: # 结构化输出失败时的降级处理 print(f结构化输出失败: {e}) return QAResponse( answer抱歉我暂时无法回答这个问题。, confidence0.0, sources[], category其他 ) # 使用示例 response ask_question(LangChain的with_structured_output怎么用) print(f回答: {response.answer}) print(f置信度: {response.confidence}) print(f类别: {response.category}) print(f来源: {response.sources})降级处理很重要。结构化输出不是100%成功的模型可能返回不符合Schema的内容Pydantic校验会抛异常。这时候不能直接把异常抛给用户要有一个兜底的返回值。我选择返回一个置信度为0的默认回答下游系统看到置信度为0就知道这次回答不可靠可以走人工审核流程。4.5 批量处理与并发控制实际生产中问答器往往需要批量处理问题。LangChain的Runnable支持batch方法可以一次传入多个输入。questions [ LangChain是什么, Pydantic V2有什么新特性, Agent和Chain有什么区别 ] results qa_chain.batch( [{question: q} for q in questions], config{max_concurrency: 5} ) for q, r in zip(questions, results): print(f问题: {q}) print(f回答: {r.answer}) print(f置信度: {r.confidence}) print(---)max_concurrency控制并发数。设太高会触发模型的速率限制设太低处理速度慢。我的经验值是5到10之间具体取决于你的API配额。如果用的是按量付费的API并发太高还可能产生额外的费用因为失败的请求也会计费。5. 常见问题与排查技巧实录5.1 结构化输出失败的典型原因结构化输出失败的原因可以归为三类我整理了一个速查表。问题现象可能原因排查方法解决方案校验异常字段缺失模型没理解Schema打印原始输出在提示词中补充字段说明校验异常类型错误模型返回了字符串形式的数字检查原始输出用Pydantic的coerce模式校验异常枚举值越界模型自创了类别检查category字段用Literal约束提示词限定调用超时网络问题或模型负载高检查网络和API状态增加重试机制和超时时间返回None模型不支持结构化输出查看模型文档换支持Function Calling的模型5.2 重试策略的设计结构化输出失败后要不要重试我的答案是要但要有策略。无脑重试只会浪费token和时间。我的做法是区分错误类型。如果是网络超时直接重试最多三次。如果是Pydantic校验失败先检查是不是提示词的问题如果是偶发的格式错误可以重试一次如果同一个问题连续两次校验失败说明模型确实理解不了这个Schema重试也没用直接走降级逻辑。from tenacity import retry, stop_after_attempt, wait_exponential from pydantic import ValidationError retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10), retrylambda e: not isinstance(e, ValidationError) ) def ask_with_retry(question: str) - QAResponse: return qa_chain.invoke({question: question})注意retry参数里的lambda只有非ValidationError的异常才重试。ValidationError说明是格式问题重试大概率还是失败不如直接降级。5.3 置信度的校准问题模型自己报的置信度准不准说实话不太准。我做过一个测试让模型回答100个问题并记录置信度然后人工标注实际正确率。结果发现模型在置信度0.8以上的回答中实际正确率只有70%左右。模型普遍存在过度自信的问题。怎么校准两个方法。一是后处理校准收集一批标注数据画一个置信度-准确率的校准曲线然后根据曲线调整阈值。比如你发现置信度0.8对应的实际准确率是0.7那下游系统就应该把阈值设到0.85以上才认为回答可靠。二是提示词校准在系统提示词里明确告诉模型“只有当你非常确定时才给高置信度不确定时要给低分”实测能稍微改善过度自信的问题。5.4 踩过的坑字段顺序影响输出这个坑很隐蔽。Pydantic模型的字段定义顺序会影响模型生成输出的顺序。如果你把confidence放在answer前面模型会先输出置信度再输出回答。这本身没问题但有些模型在生成置信度的时候还没想好答案导致置信度不准。我的做法是把answer放在第一个字段让模型先想答案再评估置信度。实测下来这样得到的置信度更合理。这个细节在官方文档里没有提是我对比了几百次调用结果之后发现的。5.5 踩过的坑默认值的陷阱Pydantic的default和default_factory用起来很方便但在结构化输出场景下要小心。如果你给某个字段设了默认值模型可能会偷懒不生成这个字段直接用默认值。比如category字段如果设了默认值“其他”模型可能把所有问题都归类为“其他”因为它觉得不填也没关系。我的原则是核心字段不设默认值辅助字段才设。answer和confidence是核心字段必须让模型生成sources是辅助字段可以设默认空列表。这样既保证了核心信息的完整性又避免了模型偷懒。6. 进阶优化与扩展方向6.1 多轮对话中的结构化输出单轮问答的结构化输出已经跑通了但多轮对话要复杂一些。主要问题是历史消息怎么处理。如果把所有历史消息都塞进提示词token消耗会很大如果只保留最近几条模型可能丢失上下文。我的方案是用LangChain的MessagesPlaceholder来管理对话历史同时给历史消息做一个摘要压缩。具体做法是保留最近三轮的完整对话更早的历史用一个小模型生成摘要把摘要作为系统消息的一部分注入。这样既控制了token消耗又保留了关键上下文。from langchain_core.prompts import MessagesPlaceholder prompt ChatPromptTemplate.from_messages([ (system, system_prompt), MessagesPlaceholder(variable_namehistory), (human, {question}) ])6.2 动态Schema的实现有些场景下输出的字段不是固定的。比如用户问技术问题需要返回代码示例问产品问题需要返回价格信息。这时候可以用动态Schema根据问题类型选择不同的Pydantic模型。实现思路是用一个路由链先判断问题类型然后根据类型选择对应的结构化模型。LangChain的RunnableBranch可以做这个事情。from langchain_core.runnables import RunnableBranch tech_llm llm.with_structured_output(TechResponse) product_llm llm.with_structured_output(ProductResponse) router RunnableBranch( (lambda x: 技术 in x[question], tech_llm), (lambda x: 产品 in x[question], product_llm), llm.with_structured_output(QAResponse) # 默认 )这个方案的好处是每个类型的Schema可以更精简模型的理解成本更低结构化输出的成功率更高。缺点是路由判断本身也可能出错需要加一层校验。6.3 结构化输出的监控与告警上线之后监控是必不可少的。我主要监控三个指标结构化输出成功率、平均置信度、各类别的分布。成功率低于95%就要告警说明模型或者提示词出了问题平均置信度突然下降可能意味着知识库需要更新类别分布异常可能说明路由逻辑有bug。监控数据我直接存在日志里用简单的脚本做聚合分析。不需要上很重的监控系统对于中小规模的Agent应用日志加定时脚本就够了。import json from datetime import datetime def log_response(question: str, response: QAResponse, success: bool): log_entry { timestamp: datetime.now().isoformat(), question: question, success: success, confidence: response.confidence if success else None, category: response.category if success else None } with open(qa_log.jsonl, a, encodingutf-8) as f: f.write(json.dumps(log_entry, ensure_asciiFalse) \n)这个日志格式可以直接用pandas读取做分析也可以导入到任何日志系统中。关键是字段要统一方便后续做聚合。6.4 结构化输出与RAG的结合问答器如果接了知识库结构化输出会更有价值。因为RAG场景下你不仅需要模型回答还需要知道它引用了哪些文档片段。sources字段可以直接存文档ID或者片段内容下游系统可以据此做引用展示或者溯源。我在RAG场景下的做法是检索阶段拿到文档片段后把片段ID和内容一起传给模型要求模型在sources字段中返回引用的片段ID。这样回答和来源就绑定在一起了用户点击来源可以直接跳转到原文。这个方案的关键是提示词要写清楚让模型只返回真正引用了的片段ID不要把所有检索到的片段都列上。实测下来模型在这方面的判断还是比较准的偶尔会多列一两个但不会漏掉关键来源。6.5 性能优化的几个实测数据最后分享几个性能相关的实测数据供参考。用GPT-4o做结构化输出单次调用的延迟在1.5到3秒之间取决于回答长度。加了with_structured_output之后延迟比普通调用增加约15%因为模型需要额外生成结构化格式的token。批量处理时并发数设为5的情况下100个问题的总处理时间大约是40秒。并发数提到10总时间降到25秒左右但失败率从1%升到了3%。所以并发不是越高越好要平衡速度和稳定性。Pydantic V2的校验速度比V1快很多对于我这种只有四个字段的模型校验耗时可以忽略不计。即使字段增加到二十个单次校验也在1毫秒以内不会成为性能瓶颈。这些数据是在特定环境和模型下测的你的实际情况可能不同但量级上应该差不多。做容量规划的时候可以参考。
返回列表