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如果不符合Schema,Pydantic会直接报错,而不是悄悄放过去。
第二,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(default=None, description="联系电话,11位数字") product: ProductInterest = Field(description="意向产品类别") budget: Optional[float] = Field(default=None, description="预算金额,单位元") urgency: int = Field(ge=1, le=5, description="紧急程度,1最低5最高") notes: List[str] = Field(default_factory=list, 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_output
from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) structured_llm = llm.with_structured_output(CustomerIntent) result = structured_llm.invoke("客户张三,电话13812345678,对A产品感兴趣,预算大概5万,比较急") print(result) # CustomerIntent(name='张三', phone='13812345678', product=<ProductInterest.A_PRODUCT: 'A产品'>, budget=50000.0, urgency=..., notes=[...])这种方式最简洁,LangChain会自动处理Schema到模型API的转换。底层它优先用function calling,如果模型不支持就退回到JSON mode。
方式二:PydanticOutputParser + Prompt
from langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import ChatPromptTemplate parser = PydanticOutputParser(pydantic_object=CustomerIntent) 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。我试过temperature=0.7,模型偶尔会把Enum字段填成Schema里没有的值,或者把数字字段填成字符串。
重试策略。即使温度调到0,模型偶尔还是会输出不符合Schema的内容。这时候不能直接报错给用户,要有重试机制。LangChain的with_retry可以配置重试次数:
structured_llm = llm.with_structured_output(CustomerIntent).with_retry( stop_after_attempt=3 )重试的时候,最好把上一次的错误信息也传给模型,让它知道哪里错了。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_KEY=your_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(default=None, description="11位手机号") product: ProductInterest = Field(description="意向产品") budget: Optional[float] = Field(default=None, description="预算,单位元") urgency: int = Field(ge=1, le=5, description="紧急程度1-5") notes: List[str] = Field(default_factory=list, 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(model="gpt-4o-mini", temperature=0) structured_llm = llm.with_structured_output(CustomerIntent).with_retry( stop_after_attempt=3 ) 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="未知", product=ProductInterest.UNKNOWN, urgency=1, 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(indent=2))跑出来的结果大概是这样:
{ "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%左右。
temperature=0。前面说过了,结构化输出不需要创造力。设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-8 |
5.5 独家避坑技巧
技巧一:先用小模型试Schema。Schema设计好之后,先用mini模型跑一批测试数据。如果mini能跑通,大模型肯定没问题。如果mini跑不通,先别急着换大模型,大概率是Schema本身有问题。
技巧二:给模型看例子。在description里加一两个示例,比写一堆约束管用。比如description="紧急程度1-5,1是'不急',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。用数据说话,比拍脑袋靠谱。