1. 这不是“问答器”是让AI交出标准答卷的监考系统你有没有遇到过这样的场景给大模型提一个问题它洋洋洒洒写了一大段逻辑看似通顺但关键字段漏了、日期格式乱了、JSON结构根本没法被下游程序解析——更糟的是你根本没法在代码里直接拿它当数据用。我去年在做客户合同智能提取项目时就栽在这上面前端传一个PDF后端调用LLM提取“签约方”“金额”“生效日期”三个字段结果模型返回的是一段带编号的自然语言描述比如“1. 签约方北京某某科技有限公司2. 金额人民币叁佰贰拾万元整3. 生效日期2024年5月1日”。这玩意儿连正则都难稳定匹配更别说塞进数据库字段里了。后来我们把整个流程重构成“结构化输出问答器”核心就一句话不让模型自由发挥而是让它按考卷答题规范来写答案。这不是加个prompt就能解决的事——它需要一套完整的约束机制从输入意图识别、到中间工具调用编排、再到最终输出格式的强制校验与自动修复。LangChain本身不提供这个能力Pydantic也不是万能钥匙真正起作用的是三者之间那条看不见的“监考链路”Prompt模板定义题干 → Agent决定解题路径 → Pydantic Schema定义标准答案格式 → 输出解析器强制校验重试兜底。这四个环节缺一不可任何一个松动AI就会开始“写作文”。这个实践系列第四篇我们就拆解这个“监考系统”的真实落地过程。它不讲抽象概念只讲你在FastAPI接口里怎么写那一行agent.invoke({input: 请提取合同中的签约方和金额})之后背后发生了什么不讲LangChain文档里的hello world只讲你上线后第二天凌晨三点收到告警说“输出格式错误率突增17%”时该怎么定位是schema定义问题、还是LLM温度值设太高、或是工具调用返回了脏数据。关键词Agent、LangChain、Pydantic、结构化输出、问答器每一个都不是孤立存在它们是在真实业务压力下咬合在一起的齿轮。2. 为什么“结构化输出”不能靠Prompt硬刚——从三次失败尝试说起很多人第一反应是“加个system prompt不就行了比如‘请严格按JSON格式输出包含key1、key2、key3’”。我试过而且不止一次。下面这三次失败是我们踩出来的典型坑也是理解整个架构设计逻辑的起点。2.1 第一次失败Prompt越长模型越叛逆我们最初用的prompt是这样的你是一个专业的合同信息提取助手。请严格按以下JSON格式输出不要任何额外文字、解释或markdown格式 { signing_party: 字符串签约方全称不含括号和备注, amount: 数字单位为元保留两位小数, effective_date: 字符串格式为YYYY-MM-DD } 请根据以下合同文本提取信息实测下来错误率高达42%。最典型的错误有三类模型在JSON外加了“好的这是您要的信息”这类引导语amount字段返回了“¥3,200,000.00”这种带符号和逗号的字符串而不是纯数字effective_date返回了“2024年5月1日”这种中文格式。提示大模型对“严格按JSON格式”的理解和程序员对JSON Schema的理解根本不在同一维度。它把“格式”当成视觉样式而不是数据契约。就像你告诉小学生“请用楷体写作文”他真会去挑一支楷体笔但不会管你心里想的是“必须用田字格本、每行12字、标点占一格”。2.2 第二次失败Pydantic Schema只是“事后诸葛亮”我们很快引入Pydantic定义了这样的Modelfrom pydantic import BaseModel, Field from datetime import date class ContractInfo(BaseModel): signing_party: str Field(..., description签约方全称不含括号和备注) amount: float Field(..., description金额单位为元保留两位小数) effective_date: date Field(..., description生效日期格式YYYY-MM-DD)然后用ContractInfo.model_validate_json(response)做校验。结果呢校验失败直接抛异常服务崩了。我们以为加个try-except就行但问题没解决——用户看到的是500错误而不是“我正在重试”。更麻烦的是有些错误Pydantic根本抓不住比如模型返回{signing_party: null}Pydantic允许None因为没设strictTrue但下游数据库字段是NOT NULL插入时照样失败。注意Pydantic的model_validate_json不是“转换器”而是“契约验证器”。它只负责说“这个JSON符不符合我的Schema”不负责说“这个JSON能不能修好”。指望它自动把“2024年5月1日”转成date对象就像指望交通摄像头自动把闯红灯的车拖回停车线——它只记录违规不执行修正。2.3 第三次失败Agent的“工具调用”成了格式污染源我们改用LangChain的ReAct Agent让它先调用OCR工具提取文本再调用LLM提取字段。问题来了OCR返回的文本里有大量换行符、空格、乱码比如“北京某某科技有限公司\n\n甲方”LLM在处理时把这些噪声直接带进了输出。更隐蔽的是Agent在调用多个工具后会把所有工具返回的原始内容拼接进提示词而这些内容往往自带HTML标签或Markdown语法LLM一读就懵输出格式彻底失控。我们抓包发现一次典型失败请求中Agent传给LLM的context里混着OCR返回的p签约方span classhighlight北京某某科技有限公司/span/p另一个工具返回的[{amount: 3200000.00, currency: CNY}]还有一段PDF元数据{CreationDate: D:202404281532110800}LLM面对这种“食材混搭”做的不是提取而是“烹饪创作”。这三次失败让我们彻底明白结构化输出不是单点优化而是一套端到端的流水线治理。它必须覆盖从输入清洗、中间态标准化、到输出强制合规的全链路。而LangChain的Agent框架恰恰提供了这条流水线的骨架——只是默认没装上“监考摄像头”和“自动修正臂”。3. 四层监考链路如何让Agent交出标准答卷真正的结构化输出问答器不是在LLM输出后加一层校验而是把校验规则前置、嵌入、贯穿整个推理链路。我们最终落地的方案是四层递进式监考机制每一层都对应一个明确的技术组件和设计意图。3.1 第一层Prompt工程——用“填空题”代替“论述题”我们彻底放弃了“请按JSON格式输出”这类开放式指令改用结构化填空模板。核心思想是把输出格式变成LLM无法绕开的“答题卡”。请根据以下合同文本填写下方表格。只填写表格内容不要任何额外说明、标题或格式符号。 | 字段名 | 值 | |--------|----| | signing_party | {{value}} | | amount | {{value}} | | effective_date | {{value}} | 合同文本 {{input}}这个模板的关键在于使用表格而非JSON规避LLM对JSON语法的随意发挥它更习惯处理表格对齐{{value}}是占位符LLM必须填满否则表格不完整这触发了它的“完整性本能”表格头明确限定字段名且与Pydantic Model的field name完全一致为后续解析打下基础。实测效果仅靠这一层格式错误率从42%降到19%。最妙的是即使LLM在表格后加了废话我们也能用正则精准截取| signing_party | (.?) |这段拿到原始值再交给Pydantic做类型转换——相当于把“格式校验”降级为“文本提取”难度大幅降低。3.2 第二层Agent工具链——所有中间数据必须“过筛子”我们重构了所有工具OCR、PDF解析、数据库查询等强制它们返回的数据必须符合预定义的Pydantic Model。例如OCR工具不再返回原始字符串而是class OcrResult(BaseModel): raw_text: str Field(..., descriptionOCR识别的原始文本已去除控制字符) confidence: float Field(..., ge0.0, le1.0, description识别置信度) # 工具函数返回 def ocr_tool(pdf_bytes: bytes) - OcrResult: # ... OCR逻辑 clean_text re.sub(r[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f], , raw_text) return OcrResult(raw_textclean_text, confidence0.92)Agent调用工具时LangChain会自动用OcrResult.model_validate()校验返回值。如果OCR返回了None或格式错乱Agent立刻报错终止而不是把脏数据喂给LLM。这层“数据入口过滤”把83%的格式污染挡在了LLM之前。经验工具返回Model的Field描述要和LLM的system prompt严格对齐。比如raw_text的description写“已去除控制字符”那么prompt里就必须强调“你只能基于已清洗的文本进行提取”否则LLM可能自己去“修复”那些被工具删掉的换行符又制造新污染。3.3 第三层Output Parser——不是校验器而是“格式翻译官”我们没用LangChain内置的JsonOutputParser而是自研了一个StructuredOutputParser它的工作流是提取用正则从LLM原始输出中提取表格值如| signing_party | 北京某某科技有限公司 |→北京某某科技有限公司映射将提取的字符串按字段名映射到Pydantic Model的对应field转换对每个field调用其_validate方法如amount字段的float()转换effective_date的datetime.strptime(..., %Y-%m-%d)兜底若转换失败如2024年5月1日转date失败启动备用规则——查同义词表“年/月/日”→“-”、调用轻量级NLP库dateparser重试最多3次熔断3次都失败则返回{signing_party: , amount: 0.0, effective_date: 1970-01-01}并记录warn日志保证服务不崩。这个Parser不是“非黑即白”的校验而是“尽力而为”的翻译。它把LLM的“自然语言输出”当作一种“方言”用自己的规则把它翻译成标准“普通话”。上线后格式错误率从19%降到1.2%且99%的错误都能自动修复。3.4 第四层Agent执行层——用“重试策略”替代“单次赌博”最后我们在Agent调用层加了重试逻辑。不是简单地max_retries3而是按错误类型分级重试错误类型触发条件重试动作最大次数格式解析失败OutputParser捕获ValidationError降低LLM temperature0.3→0.1重发相同prompt2工具调用失败Agent报ToolException切换备用工具如OCR失败则用PDFPlumber重试2语义不一致输出字段值与上下文明显矛盾如amount为负数生成针对性修正prompt“上文提到金额为正请重新确认”1这个策略的关键是每次重试都改变一个变量而不是盲目重放。比如temperature降低是为了减少LLM的“创造性发挥”切换工具是为了排除数据源问题针对性修正prompt则是给LLM一个“纠错锚点”。上线三个月因格式问题导致的重试占比从31%降到4.7%且平均重试耗时控制在800ms内。这四层链路环环相扣Prompt定题型、工具守入口、Parser做翻译、Agent控流程。它们共同构成了一个“让AI交出标准答卷”的监考系统。没有哪一层能单独解决问题但合起来就把结构化输出从概率事件变成了确定性交付。4. 实战代码拆解从FastAPI接口到Pydantic Schema的完整链路光讲原理不够下面给你看真实跑在生产环境里的代码。我们以FastAPI为入口LangChain Agent为引擎Pydantic为契约展示每一行代码背后的决策逻辑。所有代码都经过脱敏但结构、参数、错误处理完全真实。4.1 FastAPI接口暴露的是“能力”不是“模型”from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Dict, Any app FastAPI(title结构化问答API) class QueryRequest(BaseModel): input: str Field(..., description用户提问如提取合同中的签约方和金额) context: Dict[str, Any] Field(default{}, description可选上下文如PDF文件base64编码) class QueryResponse(BaseModel): result: Dict[str, Any] Field(..., description结构化输出结果) metadata: Dict[str, Any] Field(..., description执行元信息耗时、重试次数、使用的工具) app.post(/query, response_modelQueryResponse) async def structured_query(request: QueryRequest): try: # 关键这里不直接调LLM而是调Agent agent_result await contract_agent.ainvoke({ input: request.input, context: request.context }) return QueryResponse( resultagent_result[output], metadata{ elapsed_ms: agent_result[elapsed_ms], retry_count: agent_result[retry_count], used_tools: agent_result[used_tools] } ) except ValueError as e: # 捕获Pydantic ValidationError等业务错误 raise HTTPException(status_code400, detailstr(e)) except Exception as e: # 兜底错误 raise HTTPException(status_code500, detail服务内部错误)注意这个接口的QueryRequest和QueryResponse和后面Pydantic Model的字段名刻意不同。request.input是用户原始提问agent_result[output]才是结构化结果。这样设计是为了隔离“用户视角”和“系统视角”避免前端开发者误以为input字段也要符合结构化Schema。4.2 Pydantic Schema定义“契约”而非“容器”from pydantic import BaseModel, Field, validator from datetime import date import re class ContractOutput(BaseModel): signing_party: str Field( ..., min_length2, max_length100, description签约方全称必须为有效中文或英文公司名不含括号、备注、联系方式 ) amount: float Field( ..., ge0.01, le1e12, description合同金额单位为元必须为正数 ) effective_date: date Field( ..., description合同生效日期格式YYYY-MM-DD ) validator(signing_party) def validate_signing_party(cls, v): # 强制清洗去空格、去常见干扰符 v re.sub(r\s, , v.strip()) v re.sub(r[\(\)\[\]【】], , v) if not re.match(r^[A-Za-z\u4e00-\u9fa5·\s]$, v): raise ValueError(签约方包含非法字符) return v validator(amount) def validate_amount(cls, v): # 金额必须保留两位小数用于数据库存储一致性 return round(v, 2) validator(effective_date) def validate_effective_date(cls, v): # 禁止未来日期合同一般不签未来生效日 if v date.today(): raise ValueError(生效日期不能晚于今天) return v这个Schema的精妙之处在于validator不是装饰器而是业务规则引擎。它把“签约方不能含括号”这种业务规则固化在数据层ge/le限制不是为了防黑客而是防止OCR误识别把“100”读成“1000000000000”round(v, 2)确保所有金额入库前都是统一精度避免浮点误差。4.3 LangChain Agent用“RunnableSequence”组装监考流水线from langchain_core.runnables import RunnableSequence, RunnablePassthrough from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 1. 定义Prompt模板填空题版 prompt ChatPromptTemplate.from_messages([ (system, 你是一个合同信息提取专家。请严格按下方表格格式填写只填值不要任何额外文字。 | 字段名 | 值 | |--------|----| | signing_party | {{value}} | | amount | {{value}} | | effective_date | {{value}} | 请基于以下合同文本作答。注意签约方必须是公司全称金额单位为元日期格式为YYYY-MM-DD。), (human, {input}) ]) # 2. 初始化LLM关键参数 llm ChatOpenAI( modelgpt-4-turbo, temperature0.3, # 不是0留一点“思考空间”完全0会导致LLM僵化 max_tokens512, timeout30 ) # 3. 自研OutputParser核心 class StructuredOutputParser: def __init__(self, schema: type[BaseModel]): self.schema schema def parse(self, text: str) - dict: # 正则提取表格值省略具体实现核心是安全提取 extracted self._extract_from_table(text) # 调用schema的model_validate触发validator validated self.schema.model_validate(extracted) return validated.model_dump() # 4. 组装RunnableSequence这才是真正的“监考流水线” contract_agent RunnableSequence( # 输入预处理清洗context注入工具 {input: RunnablePassthrough(), context: RunnablePassthrough()} | preprocess_context, # 自定义清洗函数 # LLM调用 {input: prompt | llm} # 输出解析关键 | {output: StructuredOutputParser(ContractOutput).parse} # 后处理添加metadata | add_execution_metadata )这个RunnableSequence的写法是LangChain 0.1版本的最佳实践。它把整个流程声明式地串起来每一环节都是可插拔的。比如preprocess_context函数里我们会检查context是否含PDF base64如果是就调用OCR工具并缓存结果——这步就在Agent调用前完成了避免了前面说的“工具数据污染”。4.4 错误处理与监控让“监考”看得见最后我们加了一层全局错误处理器app.middleware(http) async def log_structured_errors(request: Request, call_next): start_time time.time() try: response await call_next(request) return response except HTTPException as e: # 记录业务错误如Pydantic ValidationError logger.warning(fStructuredQueryError: {e.detail} | path{request.url.path}) raise except Exception as e: # 记录未预期错误如LLM超时、网络问题 logger.error(fAgentRuntimeError: {str(e)} | path{request.url.path}, exc_infoTrue) raise同时在Prometheus里暴露了两个关键指标agent_output_format_error_total{typejson_parse, servicecontract}JSON解析失败次数agent_output_fix_success_total{fieldeffective_date, servicecontract}日期字段自动修复成功次数这些指标直接接入我们的告警系统。当format_error_total5分钟内突增超过10次就自动触发告警并推送错误样本到钉钉群——运维同学不用登录服务器就能看到是哪个字段、哪类错误在爆发。这套代码不是demo而是每天处理2.3万次请求的生产系统。它证明了一件事结构化输出不是LLM的附属功能而是需要独立设计、独立监控、独立演进的核心能力。5. 那些没人告诉你的“监考”陷阱来自生产环境的6条血泪经验理论和代码都给了但真正决定成败的往往是那些文档里找不到、教程里不提、只有在凌晨三点排查线上故障时才会咬牙记下的细节。以下是我们在半年运维中总结的6条血泪经验每一条都配了真实案例和解决方案。5.1 经验1Pydantic的strictTrue不是银弹它会让你的Agent“假死”我们曾给amount字段加上strictTrue期望它拒绝一切非float输入。结果上线后Agent在处理“¥3,200,000.00”时直接抛ValidationError但奇怪的是日志里没有任何错误堆栈服务CPU飙到100%请求全部超时。排查发现strictTrue在model_validate时如果输入是字符串它不会尝试转换而是直接报错。而我们的OutputParser在提取时拿到的就是字符串¥3,200,000.00。Pydantic不转换Parser又没做清洗就卡死了。解决方案永远不要在Agent链路里用strictTrue。改为用validator做显式转换validator(amount) def parse_amount(cls, v): if isinstance(v, str): # 移除货币符号和逗号 v re.sub(r[¥$€,], , v) try: return float(v) except ValueError: raise ValueError(f无法解析金额: {v}) return float(v)教训strictTrue适合数据入库前的最终校验不适合LLM输出这种“半成品”场景。Agent链路里宁可多写几行清洗代码也不要依赖Pydantic的“严格”。5.2 经验2LLM的temperature和top_p必须动态调整静态值是最大隐患我们最初把temperature0.3写死在配置里。某天下午客户上传了一批扫描质量极差的合同模糊、倾斜、有水印LLM提取signing_party时开始胡编比如把“北京某某科技”识别成“北京某市科技”错误率从1.2%飙升到37%。分析发现低temperature让LLM过于“保守”面对模糊文本它宁愿编造也不愿输出空值。而top_p0.9又限制了它的探索空间。解决方案根据OCR置信度动态调整def get_llm_params(ocr_confidence: float): if ocr_confidence 0.6: return {temperature: 0.7, top_p: 0.95} # 模糊时鼓励LLM多猜几个可能 elif ocr_confidence 0.8: return {temperature: 0.4, top_p: 0.9} else: return {temperature: 0.2, top_p: 0.8} # 清晰时追求精确现在Agent会先调OCR拿到confidence再动态设置LLM参数。上线后模糊文档的错误率回到2.1%。5.3 经验3别信“工具返回一定是干净的”所有工具输出都要过model_validate我们有个数据库查询工具返回{id: 123, name: 张三}。测试时一切正常。上线后某次数据库慢查询工具超时返回了{error: timeout, data: null}。Agent没做校验直接把这个dict喂给LLMLLM一看error: timeout就输出{signing_party: timeout, amount: 0.0}——整个结果被污染。解决方案所有工具函数必须用Pydantic Model包装返回值并在Agent配置里开启return_intermediate_stepsTrue这样就能在中间步骤里捕获ValidationError。class DbQueryResult(BaseModel): data: List[Dict[str, Any]] Field(...) error: Optional[str] Field(defaultNone) def db_tool(query: str) - DbQueryResult: try: result execute_query(query) return DbQueryResult(dataresult) except Exception as e: return DbQueryResult(data[], errorstr(e))5.4 经验4model_dump()和model_dump_json()选错会让前端崩溃我们曾用model_dump_json()返回给前端觉得“JSON字符串更标准”。结果前端JavaScript的JSON.parse()报错因为Pydantic默认序列化date字段为2024-05-01但某些老版本浏览器不认这种格式。解决方案永远用model_dump()返回dict让FastAPI的JSONResponse自动处理序列化。FastAPI内置的JSON encoder对date、datetime、Decimal等类型有完善支持。# ✅ 正确 return ContractOutput(...).model_dump() # ❌ 错误除非你明确需要字符串 return ContractOutput(...).model_dump_json()5.5 经验5Agent的max_iterations不是防死循环而是防“逻辑坍塌”LangChain Agent默认max_iterations15。我们没改结果遇到一个特殊合同OCR把“甲方北京某某科技有限公司”识别成“甲方北京某某科技有限公司”少了冒号。LLM在第一次尝试时把“甲方北京某某科技有限公司”当成了signing_party值。第二次迭代Agent又调OCR结果还是同样错误LLM又输出同样错误……15次后返回{signing_party: 甲方北京某某科技有限公司}这显然不对。解决方案把max_iterations设为3并在每次迭代后用规则引擎做“语义合理性检查”def semantic_check(output: dict) - bool: # 检查签约方是否含“甲方”“乙方”等前缀 if output.get(signing_party, ).startswith((甲方, 乙方)): return False # 检查金额是否为整数合同金额通常不带小数 if output.get(amount, 0) % 1 ! 0: return False return True # 在Agent循环里 for i in range(3): result agent.invoke(...) if semantic_check(result[output]): break # 否则生成修正prompt重试5.6 经验6监控不是看“成功率”而是看“修复率”我们最初的监控只看success_rate。当它从99.8%降到99.2%大家觉得没问题。直到某天运营同学反馈“客户说提取的金额总是少一位小数”。查日志才发现amount字段的自动修复逻辑把“3200000”修复成了“3200000.00”但数据库字段是DECIMAL(10,2)插入时被截断成“3200000.00”——看起来对其实是错的。解决方案新增监控指标fix_ratio{fieldamount, typeprecision}统计“修复前后精度变化”的比例。当这个指标突增就说明修复逻辑有问题需要人工介入。现在我们的监控面板有三块核心区域上游健康度OCR置信度分布、工具调用失败率监考有效性各字段的parse_success_rate、fix_success_rate下游兼容性各字段写入数据库的truncate_count、type_mismatch_count这六条经验每一条都来自真实的火线。它们不性感不炫技但能让你的结构化问答器在千万次请求中稳如磐石。我在实际使用中发现最难的从来不是让AI输出结构化数据而是让整个系统在数据质量波动、模型行为漂移、业务规则变更的多重压力下依然保持输出的确定性。这需要的不是某个神奇的prompt而是一套像工业流水线一样精密、可监控、可演进的监考机制。当你把Agent当作一个需要被管理的“员工”而不是一个可以被信任的“专家”时结构化输出才真正从理想照进现实。