LangChain+Pydantic结构化输出实战:打造稳定可靠的问答器

📅 发布时间:2026/10/8 20:45:56
LangChain+Pydantic结构化输出实战:打造稳定可靠的问答器
1. 为什么我要单独做一个结构化输出问答器做Agent开发的人迟早会撞上同一堵墙模型能说会道但一让它吐结构化数据就开始飘。你让它返回JSON它给你包一层json代码块你让它按schema填字段它自作主张加个note字段你让它字段名用snake_case它偏要写camelCase。更崩溃的是同样的prompt跑十次有三次格式是对的七次各有各的错法。我在做Agent实践系列的时候前面几个demo都是能跑就行直到要接下游系统——下游要的是干净的、可校验的、字段固定的对象不是一段自然语言。这时候结构化输出从一个可选项变成了硬需求。这个问答器的定位很明确输入一个自然语言问题输出一个严格符合Pydantic模型定义的结构化对象中间不允许有任何格式漂移。它解决的核心问题有三个。第一是格式确定性下游拿到的东西必须能直接model_validate通过不需要写一堆正则去清洗。第二是语义可校验不只是字段类型对字段内容也要符合业务约束比如置信度必须在0到1之间分类只能是预设的几个枚举值。第三是可观测当模型输出不合规时我要知道它错在哪、错了几次、重试有没有救回来而不是两眼一抹黑。适合谁看如果你已经跑通过最基础的LangChain调用知道ChatOpenAI怎么实例化但一到让模型稳定输出JSON就头疼这篇就是给你写的。如果你还没接触过Agent建议先把LangChain的入门链路走一遍再回来不然中间一些概念会有点跳。关键词里出现了LangChain和Pydantic这两个是本文的技术底座。LangChain负责编排和模型调用Pydantic负责schema定义和校验。至于热词里那一堆agent框架对比、agent架构、agent学习路线我不打算在这里展开——那些是选型话题本文只聚焦一件事怎么把结构化输出这件事做扎实。选型可以后面再聊但结构化输出是无论你用哪个框架都绕不开的基本功。2. 结构化输出的三条技术路线与我的选型逻辑在动手写代码之前得先想清楚用哪条路。市面上做结构化输出大致有三条路线各有各的脾气选错了后面会一直难受。2.1 路线一纯Prompt约束加事后解析最朴素的做法在prompt里写清楚你必须返回JSON字段如下……然后拿到文本自己json.loads。优点是零依赖、任何模型都能用。缺点是完全靠模型自觉模型心情好就给你对的心情不好就给你加解释文字。我早期图省事用过这条路结果在一个批量任务里1000条数据有将近200条解析失败全是模型在JSON前后加了好的以下是结果这种废话。事后解析要写一堆容错逻辑维护成本极高。2.2 路线二模型原生Function Calling / Tool Calling主流模型都支持function calling你把schema传进去模型返回的tool_calls里就是结构化参数。这条路的好处是模型侧做了约束格式稳定性比纯prompt高一个档次。但它有两个坑一是不同厂商对schema的支持程度不一样有些复杂的嵌套结构、联合类型会直接报错或者被静默忽略二是tool_calls的返回结构本身还要再解析一层而且模型可能不调用工具而是直接回一段文本你得处理这种fallback。2.3 路线三LangChain的with_structured_output这是LangChain封装好的一层底层其实还是走function calling或者JSON mode但对外暴露的接口极其干净你给它一个Pydantic模型它直接返回一个该模型的实例。我最终选的就是这条理由很实在schema即代码。Pydantic模型本身就是校验器字段类型、约束、默认值、描述全写在一个地方不用维护两份定义。失败可捕获。校验失败会抛ValidationError我能精确知道是哪个字段、什么原因方便做重试。切换模型成本低。换模型时只要新模型支持结构化输出业务代码基本不用动。提示with_structured_output在不同LangChain版本里行为有差异早期版本对嵌套模型支持不完善建议用较新的稳定版并且在正式用之前先跑一个最小验证用例。选型这件事我的经验是别一上来就追求最花哨的方案先保证失败时你能定位问题。纯prompt方案失败了你只能猜function calling失败了你至少能看到原始返回with_structured_output失败了你还能拿到ValidationError的具体字段。可观测性比一时的成功率更重要因为成功率可以靠重试补可观测性补不了。3. 用Pydantic把问答结果这件事定义清楚结构化输出的灵魂不在代码在schema设计。schema设计得好模型填起来顺校验也顺schema设计得烂模型天天给你填错你还以为是模型不行。这一节我把自己踩过的坑和最终定下来的模型结构讲清楚。3.1 从业务需求倒推字段先问自己这个问答器到底要产出什么我的场景是给定一个问题返回答案、答案的置信度、答案涉及的知识类别、以及支撑答案的要点列表。于是字段就出来了from pydantic import BaseModel, Field from typing import List from enum import Enum class Category(str, Enum): TECH tech LIFE life FINANCE finance OTHER other class QAResult(BaseModel): answer: str Field(description对问题的直接回答控制在200字以内) confidence: float Field(ge0.0, le1.0, description对答案的置信度0到1之间) category: Category Field(description问题所属类别) key_points: List[str] Field(description支撑答案的要点2到5条)这里有几个设计决策值得说。answer用str而不是更复杂的结构是因为下游要直接展示越简单越好。confidence用float加ge/le约束是因为模型很容易给出1.5或者-0.2这种越界值用Pydantic的约束直接卡死。category用Enum而不是str这是关键——枚举是约束模型输出空间最有效的手段模型只能在几个值里选不会给你编出technology这种同义词。key_points用List[str]是因为要点数量不固定用列表最自然。3.2 Field的description是给模型看的prompt很多人写Pydantic模型只写类型不写description然后抱怨模型填得不准。description不是文档是prompt的一部分。with_structured_output会把整个schema包括description序列化后传给模型模型就是靠这些description来理解每个字段该填什么。我对比过answer: str和answer: str Field(description对问题的直接回答控制在200字以内)后者的输出质量明显更稳尤其是长度控制。模型看到200字以内就会自觉收敛不写description它可能给你写800字。3.3 嵌套模型要慎用Pydantic支持嵌套模型比如key_points: List[KeyPoint]每个KeyPoint又有自己的字段。功能上没问题但实测下来嵌套层级越深模型填错的概率越高。我试过三层嵌套失败率直接翻倍。所以我的原则是能用扁平结构就别嵌套能一层解决就别两层。List[str]够用就绝不上List[SomeModel]。如果业务真的需要嵌套建议把嵌套部分拆成独立的调用先拿到主结构再针对每个子项单独调用一次。虽然多花点token但稳定性提升明显而且失败时定位更容易。3.4 默认值和Optional的取舍Optional字段和带默认值的字段模型可以选择不填。这在某些场景下是好事比如备注这种非必填项但在核心字段上绝对是灾难——模型会偷懒不填然后你拿到一个None还得处理。我的做法是核心字段一律必填不给默认值。非核心字段才用Optional并且明确在description里写如果没有相关信息填null。这样模型知道这个字段可以空但空的时候要显式填null而不是直接省略。4. 把问答器跑起来从模型调用到重试闭环schema定好了接下来是把它接上模型跑通。这一节讲完整的调用链路包括我踩过的几个坑。4.1 最小可运行版本先上一个能跑的最小版本把链路打通from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm ChatOpenAI(modelgpt-4o-mini, temperature0) structured_llm llm.with_structured_output(QAResult) prompt ChatPromptTemplate.from_messages([ (system, 你是一个严谨的问答助手请根据用户问题给出结构化回答。), (human, {question}) ]) chain prompt | structured_llm result chain.invoke({question: Pydantic的Field约束有哪些常用类型}) print(result.answer) print(result.confidence) print(result.category)这里temperature0很重要。结构化输出场景下温度越低越稳。温度高了模型会发挥创意字段值就开始飘。我一般直接设0如果发现输出太死板再往上调一点点但绝不超过0.3。with_structured_output(QAResult)这一步是关键它把Pydantic模型转成模型能理解的schema并在返回时自动做校验和反序列化。你拿到的result直接就是QAResult实例可以.answer这样访问。4.2 第一次踩坑模型返回了不合规的枚举值跑通之后我拿一批真实问题去测发现category字段偶尔会返回technology而不是tech。按理说Enum应该卡死但模型在生成时确实可能吐出枚举外的值然后Pydantic校验失败抛异常。这时候有两个选择一是让异常往上抛调用方处理二是加一层重试。我选的是重试因为大部分格式错误是随机的重试一次就能救回来。重试的逻辑不是简单重跑而是把错误信息喂回去from pydantic import ValidationError def invoke_with_retry(chain, question, max_retries3): last_error None for attempt in range(max_retries): try: return chain.invoke({question: question}) except ValidationError as e: last_error e # 把校验错误拼进问题里让模型知道错在哪 question ( f{question}\n\n f上一次输出校验失败错误信息{e.errors()}。 f请严格按照schema重新输出注意枚举字段只能取预设值。 ) raise last_error这个重试逻辑的精髓在于把错误反馈给模型。直接重跑模型大概率还犯同样的错把ValidationError的具体内容拼进prompt模型看到category字段值technology不在允许的枚举里下一次就会改。实测下来第一次失败的问题重试一次的成功率能到90%以上。4.3 第二次踩坑重试把上下文撑爆了上面那个重试逻辑有个隐患每次重试都把错误信息往question里拼如果连续失败三次question会变得很长token消耗飙升而且模型可能被一堆错误信息搞晕。我的改进是只保留最近一次的错误而不是累积。另外给重试加个上限超过就放弃并记录不要无限重试。生产环境里我还见过因为重试逻辑写错导致死循环的token账单直接爆炸这个坑一定要避开。4.4 第三次踩坑并发下的限流问答器上线后要支持批量问题我一开始用chain.batch()并发跑结果触发模型侧的限流一堆请求失败。后来加了并发控制import asyncio async def batch_invoke(chain, questions, concurrency5): semaphore asyncio.Semaphore(concurrency) async def one(q): async with semaphore: return await chain.ainvoke({question: q}) return await asyncio.gather(*[one(q) for q in questions])concurrency5是我实测下来比较稳的值具体多少要看你的账号等级和模型。别一上来就开几十并发限流一触发重试反而更慢。5. 让输出真正可用校验、兜底与可观测跑通只是第一步能稳定用在生产里才算数。这一节讲三个让结构化输出从能跑到可靠的关键动作。5.1 业务级校验不能只靠PydanticPydantic管的是类型和基础约束但业务规则它管不了。比如key_points至少2条answer不能为空字符串confidence低于0.3时category不能是other——这些都得自己写。我的做法是在Pydantic模型上加model_validatorfrom pydantic import model_validator class QAResult(BaseModel): # ... 字段定义同上 ... model_validator(modeafter) def check_business_rules(self): if len(self.key_points) 2: raise ValueError(key_points至少需要2条) if not self.answer.strip(): raise ValueError(answer不能为空) return self这样业务规则和schema定义在一起校验失败同样抛ValidationError能被上面的重试逻辑捕获。把校验前移别等到下游用的时候才发现数据有问题。5.2 兜底策略失败时返回什么重试三次还是失败怎么办直接抛异常会让整个批处理挂掉。我的兜底是返回一个降级结果def safe_invoke(chain, question): try: return invoke_with_retry(chain, question) except Exception as e: return QAResult( answer抱歉暂时无法给出结构化回答。, confidence0.0, categoryCategory.OTHER, key_points[系统未能生成合规输出, f错误类型{type(e).__name__}] )降级结果本身也是合规的QAResult下游不用做特殊处理。同时把失败记录下来方便后续分析是哪些问题容易失败。5.3 可观测记录每一次失败结构化输出最怕的是静默失败——失败了但没人知道。我加了一个简单的日志记录把每次校验失败的问题、错误字段、重试次数都记下来记录项用途question定位是哪类问题容易失败error_field看是哪个字段最常出错retry_count评估重试策略是否有效final_status成功/降级/彻底失败跑一段时间后我发现category字段的失败率最高其次是confidence越界。针对性地在prompt里加强这两个字段的说明后失败率明显下降。没有观测就没有优化这一步千万别省。6. 几个让稳定性再上一个台阶的实战技巧前面讲的都是主干这一节补充几个我实际用下来很有效的细节技巧都是文档里不太会写的东西。6.1 在system prompt里预告schema虽然with_structured_output会自动传schema但我在system prompt里还是会用自然语言再描述一遍关键约束。比如category只能是tech/life/finance/other之一confidence是0到1的小数。这看起来冗余但实测能降低首次失败率。模型同时看到schema和自然语言描述理解更到位。6.2 给字段加示例值Pydantic的Field支持examples参数把示例值填进去模型会照着示例的格式来。比如confidence给个examples[0.85]模型就不太会填85或者high这种。示例比描述更直观尤其是对格式敏感的字段。6.3 控制输出长度answer字段如果不加长度约束模型可能给你写一大段。除了在description里写200字以内我还会在prompt里强调简洁直接。如果下游对长度有硬要求可以在校验里加max_length超了就当失败重试。6.4 别在schema里放太多字段字段越多模型填错的概率越高。我见过有人一个模型塞了二十几个字段结果失败率高得离谱。只放真正需要的字段能拆成多次调用的就拆。结构化输出的目标是稳定拿到需要的几个字段不是一次拿全所有信息。6.5 版本升级要回归测试LangChain和底层模型的更新很频繁with_structured_output的行为可能随版本变化。我每次升级依赖后都会拿一批固定的测试问题跑一遍对比成功率和输出质量。别裸升级结构化输出这种强依赖底层行为的功能升级翻车的概率不低。7. 关于这个问答器我最后想说的这个问答器本身不复杂代码量不大但它把Agent开发里一个很核心的问题讲透了怎么让模型的输出从大概对变成确定对。我见过太多项目卡在这一步模型demo跑得飞起一接真实系统就各种格式问题最后不得不写一堆脏兮兮的解析代码。我的体会是结构化输出这件事七分靠schema设计两分靠重试兜底一分靠运气。schema设计得好模型填起来顺失败率天然就低schema设计得烂再怎么重试也是治标不治本。所以如果你要动手做类似的东西先把Pydantic模型反复打磨几轮比急着写调用代码重要得多。另外一点别追求100%的成功率。模型输出本质上是概率性的总会有失败的时候。关键是失败时你能兜住、能观测、能定位而不是假装它不会失败。我现在的策略是接受一个可接受的失败率把重试和降级做扎实剩下的交给监控。这个问答器后面还能往几个方向扩展接上RAG让answer有依据、把key_points做成可点击的引用、根据confidence做人工复核的分流。但那些都是后话先把结构化输出这一层做稳后面的扩展才有地基。