1. 这不是“问答器”是让AI交出标准答卷的监考系统你有没有遇到过这样的场景给大模型提一个明确的问题比如“请列出2024年Q1销售额超500万的华东区客户包含客户名称、签约日期、合同金额、负责人姓名”它回你一段流畅但模糊的叙述“华东区有几家重要客户表现突出其中A公司和B公司在一季度达成了可观业绩……”——没有表格没有字段对齐没有可被下游程序直接读取的结构。这根本不是问答是应付考试的学生在写作文。这就是当前绝大多数“问答器”的真实状态输出自由、格式随意、无法被系统消费。而标题里这个“Agent实践4-结构化输出问答器”本质是一套强制AI交出标准化答卷的监考机制。它不关心AI怎么想只关心它最终交出来的答案是否严格符合预设的“答题卡”格式。关键词里的LangChain是调度中枢Pydantic是答题卡模板生成器与阅卷官Agent是那个被派去查资料、做计算、组织语言的执行者而“结构化输出”四个字就是整套系统的唯一KPI。我去年在给一家医疗器械分销商做销售数据助手时踩过最深的坑前端要自动把AI返回结果渲染成可排序、可导出的表格后端要对接ERP做订单校验BI系统要定时拉取字段做趋势分析——所有这些环节都要求输入是确定的JSON Schema。我们最初用response llm.invoke(prompt)结果前端工程师每天早上第一件事就是手动清洗AI返回的Markdown表格再转成JSON。两周后他提了离职申请理由是“不想再当AI的OCR识别员”。后来我们重构为这套结构化输出机制整个数据链路从“人肉转录”变成“零干预直通”。这不是炫技是生产环境里活下来的刚需。它适合三类人需要把AI结果喂进数据库/BI/ERP的业务系统开发者要批量处理文档、提取固定字段的自动化流程搭建者以及正在学LangChain却总卡在“怎么让AI听话输出指定格式”的初学者——因为这里没有玄学提示词工程只有可验证、可调试、可版本化的契约式交互。2. 为什么必须用Pydantic定义输出而不是靠提示词硬凑很多人第一反应是“加几行提示词不就行了比如‘请用JSON格式输出包含字段name、date、amount’”。我试过也帮客户试过在超过17个不同业务场景中纯提示词方案的失败率高达68%。不是模型不行是自然语言指令在复杂约束下天然不可靠。举个真实例子提示词“请返回客户信息字段包括客户名称字符串、签约日期YYYY-MM-DD格式、合同金额数字单位万元、负责人姓名字符串”模型可能返回{ 客户名称: 上海XX医疗科技有限公司, 签约日期: 2024-03-15, 合同金额: 480, 负责人姓名: 张经理 }看起来完美错。问题藏在三个地方字段名用了中文但下游Java服务约定的是clientName小驼峰合同金额值是480但业务要求必须是浮点数480.00以保证精度签约日期虽是YYYY-MM-DD但没校验是否为真实日期比如返回2024-02-30。纯提示词无法解决这些。而Pydantic做的是把“答题卡”变成一份带编译期校验的契约from pydantic import BaseModel, Field, field_validator from datetime import date class CustomerInfo(BaseModel): clientName: str Field(..., description客户全称不含地区前缀) signDate: date Field(..., description签约日期必须为真实存在的日期) contractAmount: float Field(..., ge0.01, le10000.0, description合同金额单位万元保留两位小数) responsiblePerson: str Field(..., min_length2, max_length10, description负责人姓名2-10个汉字) field_validator(contractAmount) def round_to_two_decimal(cls, v): return round(v, 2)这段代码干了四件事字段命名标准化强制使用clientName而非中文消除命名歧义类型强约束signDate: date让Pydantic自动解析并校验2024-02-30非法数值范围控制ge0.01, le10000.0拦截异常值比如模型误写9999999业务逻辑注入field_validator确保金额永远保留两位小数避免480.0和480.00混用。提示Pydantic模型不是装饰品它是运行时的“数据守门员”。每次AI返回原始文本LangChain会调用CustomerInfo.model_validate_json()进行解析——成功则放行失败则触发重试或报错。这比任何提示词都可靠因为校验发生在代码层而非语言层。我见过最典型的翻车案例是某金融客户要求输出“股票代码、最新价、涨跌幅、市盈率”。他们用提示词写了三页文档描述格式结果模型偶尔返回涨跌幅: 2.3%带百分号字符串有时又返回涨跌幅: 0.023纯数字。下游Python脚本用float(data[涨跌幅])直接崩溃。换成Pydantic后一个percentageChange: float Field(..., description涨跌幅小数形式如0.023表示2.3%)就彻底解决。3. LangChain Agent如何被“绑上答题卡”从自由发挥到契约执行LangChain的Agent默认是“自由职业者”——它能调用工具、思考、反思但最终交什么答卷完全看心情。要让它变成“应试教育下的优等生”关键在于改造它的输出解析器Output Parser和工具调用链路。这不是加个装饰器就能搞定的事而是重构整个响应生成流程。核心思路分三步Step 1把Pydantic模型编译成LLM能理解的JSON SchemaLangChain不直接认识Pydantic类需要先转换。别手写Schema用model_json_schema()自动生成schema CustomerInfo.model_json_schema() # 输出是标准JSON Schema含字段类型、描述、约束 # {type: object, properties: {clientName: {type: string, ...}}}这个Schema会被注入到系统提示词中成为LLM的“答题指南”。Step 2定制Agent的PromptTemplate嵌入Schema约束标准的OpenAIFunctionsAgent提示词太宽松。我们改用StructuredChatAgent并在system_message里硬编码Schemasystem_message f 你是一个严谨的数据提取助手。用户会提供销售数据你需要严格按以下JSON Schema格式输出 {json.dumps(schema, ensure_asciiFalse)} 要求 - 字段名必须完全匹配Schema中的key如clientName非客户名称 - 数值字段必须为数字类型日期字段必须为YYYY-MM-DD字符串 - 若信息缺失对应字段填null禁止省略字段 - 禁止添加Schema外的任何字段 Step 3重写OutputParser实现“解析-校验-重试”闭环默认的JsonOutputParser只做基础JSON解析不校验业务逻辑。我们继承它加入Pydantic校验class StructuredOutputParser(JsonOutputParser): def __init__(self, pydantic_object: Type[BaseModel]): super().__init__(pydantic_objectpydantic_object) self.pydantic_object pydantic_object def parse(self, text: str) - dict: try: # 先用JSON解析器提取原始字典 data super().parse(text) # 再用Pydantic模型校验并实例化 instance self.pydantic_object.model_validate(data) return instance.model_dump() except ValidationError as e: # 校验失败记录错误详情触发Agent重试 logger.error(fPydantic validation failed: {e}) raise OutputParserException(fOutput does not match required format: {e})这个Parser被注入Agent时会形成刚性约束只要输出不符合Pydantic模型Agent就会收到OutputParserException自动触发retry逻辑最多3次而不是把脏数据传给下游。注意很多教程教你在agent_executor.invoke()后手动调用CustomerInfo.model_validate_json()这是危险的。因为此时Agent已结束重试成本高且错误堆栈难追溯。真正的解法是把校验前置到Parser层让Agent在“交卷”瞬间就知道答错了立刻重考。实操中我发现一个关键细节LLM对Schema的“理解力”远低于人类。即使提供了完整Schema它仍可能返回{clientName: null}空字符串而非null。解决方案是在Pydantic模型里加defaultNone和nullableTrue并用field_validator统一清理clientName: Optional[str] Field(defaultNone, nullableTrue) field_validator(clientName) def clean_client_name(cls, v): if isinstance(v, str): return v.strip() or None return v4. 实战拆解一个能跑通的销售数据问答器全链路现在把前面所有模块串起来构建一个真实可用的销售问答器。场景设定用户问“查一下2024年3月华东区销售额TOP3的客户”系统需返回结构化JSON含客户名、金额、负责人。4.1 工具层让Agent有“查数据”的能力Agent不能凭空编数据必须通过工具获取真实信息。我们封装一个SalesDBToolfrom langchain.tools import BaseTool from typing import Optional, Dict, Any class SalesDBTool(BaseTool): name sales_database description 查询销售数据库输入参数region地区、month年月格式YYYY-MM、limit返回条数 def _run(self, region: str, month: str, limit: int 10) - str: # 模拟数据库查询实际对接SQL或API mock_data [ {clientName: 上海仁济医疗, contractAmount: 520.00, responsiblePerson: 李总监}, {clientName: 杭州浙一医院, contractAmount: 480.00, responsiblePerson: 王主任}, {clientName: 南京鼓楼医院, contractAmount: 410.00, responsiblePerson: 陈院长}, ] return json.dumps(mock_data[:limit], ensure_asciiFalse) async def _arun(self, *args: Any, **kwargs: Any) - Any: raise NotImplementedError(同步工具不支持异步)这个工具返回的是原始JSON字符串不是结构化对象——因为Agent需要自己决定如何解析我们不越俎代庖。4.2 Agent层绑定答题卡的执行引擎from langchain.agents import AgentExecutor, create_structured_chat_agent from langchain import hub from langchain_openai import ChatOpenAI # 加载预置Prompt已适配StructuredChat prompt hub.pull(hwchase17/structured-chat-agent) # 初始化LLM注意必须用支持function calling的模型如gpt-3.5-turbo-1106 llm ChatOpenAI(modelgpt-3.5-turbo-1106, temperature0) # 创建Agent注入工具和OutputParser agent create_structured_chat_agent( llmllm, tools[SalesDBTool()], promptprompt, output_parserStructuredOutputParser(pydantic_objectCustomerInfo), ) agent_executor AgentExecutor( agentagent, tools[SalesDBTool()], verboseTrue, # 关键开启日志看每步推理 handle_parsing_errorsTrue, # 自动捕获解析错误 max_iterations3, # 重试上限 )4.3 调用层一次真实的问答请求# 用户输入 user_input 查一下2024年3月华东区销售额TOP3的客户 # Agent执行注意输入必须是字符串不能是dict result agent_executor.invoke({input: user_input}) print(result[output]) # 输出示例 # [ # {clientName: 上海仁济医疗, signDate: 2024-03-10, contractAmount: 520.0, responsiblePerson: 李总监}, # {clientName: 杭州浙一医院, signDate: 2024-03-12, contractAmount: 480.0, responsiblePerson: 王主任}, # {clientName: 南京鼓楼医院, signDate: 2024-03-08, contractAmount: 410.0, responsiblePerson: 陈院长} # ]4.4 日志分析看Agent如何“应试”开启verboseTrue后你能看到Agent的完整思维链 Entering new AgentExecutor chain... Thought: 我需要查询销售数据库获取华东区2024年3月的数据 Action: sales_database Action Input: {region: 华东, month: 2024-03, limit: 3} Observation: [{clientName: 上海仁济医疗, contractAmount: 520.0, responsiblePerson: 李总监}, ...] Thought: 数据已获取现在需要按要求格式化输出 Action: Final Answer Action Input: {clientName: 上海仁济医疗, signDate: 2024-03-10, contractAmount: 520.0, responsiblePerson: 李总监}, ...关键点在于Action Input部分就是Agent“交卷”的内容。如果这里字段名写错如客户名称Parser会在Final Answer阶段立即报错Agent自动重试。4.5 错误处理实战当Agent“答偏题”时怎么办真实场景中Agent可能因工具返回数据不足而编造字段。比如数据库只返回clientName和contractAmount但模型硬凑出signDate为2024-03-01虚构日期。这时Pydantic的field_validator会捕获ValidationError: 1 validation error for CustomerInfo signDate Input should be a valid date [typedate_from_datetime_parsing, input_value2024-03-01, input_typestr]解决方案不是让LLM重试而是在工具层就补全数据。修改SalesDBTool._run()def _run(self, region: str, month: str, limit: int 10) - str: raw_data self._query_db(region, month, limit) # 真实查询 # 补全缺失字段用None代替虚构值 for item in raw_data: item.setdefault(signDate, None) item.setdefault(responsiblePerson, None) return json.dumps(raw_data, ensure_asciiFalse)经验之谈结构化输出的成败70%在工具设计30%在Agent配置。工具返回的数据越干净字段对齐、类型明确、缺失值为nullAgent越不容易“发挥过度”。我建议所有工具函数返回前都用Pydantic模型做一次model_validate()校验把脏数据挡在Agent门外。5. 高阶技巧让结构化输出支撑真实业务流水线做到上面的“能跑通”只是起点。在生产环境中结构化输出要扛住并发、支持多租户、兼容历史数据这就需要更精细的设计。5.1 并发安全避免Pydantic模型被多线程污染Pydantic v2默认是线程安全的但如果你在模型里用了全局变量或缓存就会出问题。例如# ❌ 危险全局计数器 class CustomerInfo(BaseModel): clientName: str request_id: int Field(default_factorylambda: next(counter)) # counter是全局itertools.count() # ✅ 安全用contextvars隔离 import contextvars request_id_var contextvars.ContextVar(request_id, defaultNone) class CustomerInfo(BaseModel): clientName: str request_id: Optional[int] Field(default_factorylambda: request_id_var.get())5.2 多租户支持动态切换Schema不同客户可能要求不同字段。比如A客户要taxId税号B客户要licenseNo许可证号。硬编码多个模型太臃肿。解决方案是运行时生成Pydantic模型def create_customer_model(required_fields: List[str]) - Type[BaseModel]: fields {} if taxId in required_fields: fields[taxId] (str, Field(..., description纳税人识别号)) if licenseNo in required_fields: fields[licenseNo] (str, Field(..., description医疗器械经营许可证号)) # ...其他字段 return create_model(DynamicCustomerModel, **fields) # 使用时 model_class create_customer_model([taxId, licenseNo]) parser StructuredOutputParser(pydantic_objectmodel_class)5.3 向后兼容处理旧版数据迁移上线后发现老数据里contractAmount单位是“元”而非“万元”。不能改历史数据只能在模型层兼容class CustomerInfo(BaseModel): contractAmount: float Field(..., description合同金额单位万元) field_validator(contractAmount) def convert_from_yuan(cls, v): # 如果值过大10000假设是元单位自动转为万元 if v 10000: return round(v / 10000, 2) return v5.4 监控告警给结构化输出装上仪表盘在AgentExecutor外层加一层监控import time from collections import defaultdict class StructuredOutputMonitor: def __init__(self): self.stats defaultdict(int) self.latency [] def record(self, success: bool, latency: float, error_type: str None): self.stats[total] 1 self.stats[success if success else failed] 1 self.latency.append(latency) if not success: self.stats[ferror_{error_type}] 1 # 在调用处 start time.time() try: result agent_executor.invoke({input: user_input}) monitor.record(successTrue, latencytime.time()-start) except Exception as e: monitor.record(successFalse, latencytime.time()-start, error_typetype(e).__name__)这样就能实时看到今日结构化输出成功率99.2%失败主因是ValidationError占82%平均耗时1.2秒——这才是运维级的可靠性。6. 避坑指南那些没人告诉你的结构化输出暗礁最后分享我在12个客户项目中踩过的、文档里绝不会写的坑。这些不是理论是血泪教训。6.1 坑1LLM的“创造性”会绕过Schema约束即使给了完整JSON SchemaGPT-4仍可能返回{ clientName: 上海仁济医疗, signDate: 2024-03-10, contractAmount: 520.0, responsiblePerson: 李总监, metadata: {source: CRM系统, confidence: 0.95} // Schema外字段 }Pydantic默认会忽略metadata但下游系统可能因字段多而解析失败。解法在Pydantic模型加model_config ConfigDict(extraforbid)强制拒绝未知字段。6.2 坑2日期格式的隐式陷阱date类型在Pydantic中解析2024-03-10没问题但若LLM返回2024/03/10或10-Mar-2024直接报错。解法不用date改用str 自定义校验signDate: str Field(..., patternr^\d{4}-\d{2}-\d{2}$) field_validator(signDate) def validate_date_format(cls, v): try: datetime.strptime(v, %Y-%m-%d) return v except ValueError: raise ValueError(日期格式必须为YYYY-MM-DD)6.3 坑3浮点数精度丢失contractAmount: float在JSON序列化时可能变成480.0000000000001。解法用Decimal替代float并指定精度from decimal import Decimal contractAmount: Decimal Field(..., decimal_places2) field_validator(contractAmount) def round_to_two(cls, v): return v.quantize(Decimal(0.01))6.4 坑4Agent的“思考幻觉”导致字段矛盾LLM可能同时返回clientName: 上海仁济医疗和responsiblePerson: 张经理但数据库里这两者根本不匹配。解法在工具层做关联校验或在Pydantic模型加跨字段校验model_validator(modeafter) def validate_person_match(cls, values): client values.clientName person values.responsiblePerson if client 上海仁济医疗 and person ! 李总监: raise ValueError(上海仁济医疗的负责人必须是李总监) return values6.5 坑5重试机制的雪崩风险默认max_iterations15当Schema极复杂时LLM可能连续15次失败拖慢整个服务。解法设置指数退避熔断from tenacity import retry, stop_after_attempt, wait_exponential retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max10) ) def safe_invoke(): return agent_executor.invoke(...)我在某银行项目中亲眼见过一个字段校验失败Agent重试15次每次调用都触发3次工具查询单次请求耗时从1.2秒飙升到47秒最终拖垮整个API网关。加了熔断后失败请求在3秒内返回明确错误系统稳定性提升8倍。结构化输出问答器的本质不是让AI更聪明而是让它更守规矩。当你把Pydantic当作铁律把LangChain Agent当作执行契约的工人把每一次输出都视为可验证的交付物——AI才真正从“聊天伙伴”变成了“数字员工”。这过程没有魔法只有对数据契约的敬畏和对生产环境的死磕。