☰
构建AI结构化输出监考系统:Agent+Pydantic+Prompt四层治理
2026/10/7 18:41:21 网站建设 项目流程

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,定义了这样的Model:

from 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(因为没设strict=True),但下游数据库字段是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 class="highlight">北京某某科技有限公司</span></p>
  • 另一个工具返回的[{"amount": "3200000.00", "currency": "CNY"}]
  • 还有一段PDF元数据{"CreationDate": "D:20240428153211+08'00'"}

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(..., description="OCR识别的原始文本,已去除控制字符") confidence: float = Field(..., ge=0.0, le=1.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_text=clean_text, confidence=0.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,它的工作流是:

  1. 提取:用正则从LLM原始输出中提取表格值(如| signing_party | 北京某某科技有限公司 |→北京某某科技有限公司);
  2. 映射:将提取的字符串,按字段名映射到Pydantic Model的对应field;
  3. 转换:对每个field调用其_validate方法(如amount字段的float()转换,effective_date的datetime.strptime(..., "%Y-%m-%d"));
  4. 兜底:若转换失败(如"2024年5月1日"转date失败),启动备用规则——查同义词表(“年/月/日”→“-”)、调用轻量级NLP库(dateparser)重试,最多3次;
  5. 熔断:3次都失败,则返回{"signing_party": "", "amount": 0.0, "effective_date": "1970-01-01"}并记录warn日志,保证服务不崩。

这个Parser不是“非黑即白”的校验,而是“尽力而为”的翻译。它把LLM的“自然语言输出”当作一种“方言”,用自己的规则把它翻译成标准“普通话”。上线后,格式错误率从19%降到1.2%,且99%的错误都能自动修复。

3.4 第四层:Agent执行层——用“重试策略”替代“单次赌博”

最后,我们在Agent调用层加了重试逻辑。不是简单地max_retries=3,而是按错误类型分级重试:

错误类型触发条件重试动作最大次数
格式解析失败OutputParser捕获ValidationError降低LLM temperature(0.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_model=QueryResponse) async def structured_query(request: QueryRequest): try: # 关键:这里不直接调LLM,而是调Agent agent_result = await contract_agent.ainvoke({ "input": request.input, "context": request.context }) return QueryResponse( result=agent_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_code=400, detail=str(e)) except Exception as e: # 兜底错误 raise HTTPException(status_code=500, 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_length=2, max_length=100, description="签约方全称,必须为有效中文或英文公司名,不含括号、备注、联系方式" ) amount: float = Field( ..., ge=0.01, le=1e12, 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( model="gpt-4-turbo", temperature=0.3, # 不是0!留一点“思考空间”,完全0会导致LLM僵化 max_tokens=512, timeout=30 ) # 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(f"StructuredQueryError: {e.detail} | path={request.url.path}") raise except Exception as e: # 记录未预期错误(如LLM超时、网络问题) logger.error(f"AgentRuntimeError: {str(e)} | path={request.url.path}", exc_info=True) raise

同时,在Prometheus里暴露了两个关键指标:

  • agent_output_format_error_total{type="json_parse", service="contract"}:JSON解析失败次数
  • agent_output_fix_success_total{field="effective_date", service="contract"}:日期字段自动修复成功次数

这些指标直接接入我们的告警系统。当format_error_total5分钟内突增超过10次,就自动触发告警,并推送错误样本到钉钉群——运维同学不用登录服务器,就能看到是哪个字段、哪类错误在爆发。

这套代码,不是demo,而是每天处理2.3万次请求的生产系统。它证明了一件事:结构化输出不是LLM的附属功能,而是需要独立设计、独立监控、独立演进的核心能力。

5. 那些没人告诉你的“监考”陷阱:来自生产环境的6条血泪经验

理论和代码都给了,但真正决定成败的,往往是那些文档里找不到、教程里不提、只有在凌晨三点排查线上故障时才会咬牙记下的细节。以下是我们在半年运维中总结的6条血泪经验,每一条都配了真实案例和解决方案。

5.1 经验1:Pydantic的strict=True不是银弹,它会让你的Agent“假死”

我们曾给amount字段加上strict=True,期望它拒绝一切非float输入。结果上线后,Agent在处理“¥3,200,000.00”时直接抛ValidationError,但奇怪的是,日志里没有任何错误堆栈,服务CPU飙到100%,请求全部超时。

排查发现:strict=True在model_validate时,如果输入是字符串,它不会尝试转换,而是直接报错。而我们的OutputParser在提取时,拿到的就是字符串"¥3,200,000.00"。Pydantic不转换,Parser又没做清洗,就卡死了。

解决方案:永远不要在Agent链路里用strict=True。改为用@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)

教训:strict=True适合数据入库前的最终校验,不适合LLM输出这种“半成品”场景。Agent链路里,宁可多写几行清洗代码,也不要依赖Pydantic的“严格”。

5.2 经验2:LLM的temperature和top_p必须动态调整,静态值是最大隐患

我们最初把temperature=0.3写死在配置里。某天下午,客户上传了一批扫描质量极差的合同(模糊、倾斜、有水印),LLM提取signing_party时开始胡编,比如把“北京某某科技”识别成“北京某市科技”,错误率从1.2%飙升到37%。

分析发现:低temperature让LLM过于“保守”,面对模糊文本,它宁愿编造也不愿输出空值。而top_p=0.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喂给LLM,LLM一看"error": "timeout",就输出{"signing_party": "timeout", "amount": 0.0}——整个结果被污染。

解决方案:所有工具函数,必须用Pydantic Model包装返回值,并在Agent配置里开启return_intermediate_steps=True,这样就能在中间步骤里捕获ValidationError。

class DbQueryResult(BaseModel): data: List[Dict[str, Any]] = Field(...) error: Optional[str] = Field(default=None) def db_tool(query: str) -> DbQueryResult: try: result = execute_query(query) return DbQueryResult(data=result) except Exception as e: return DbQueryResult(data=[], error=str(e))

5.4 经验4:model_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 经验5:Agent的max_iterations不是防死循环,而是防“逻辑坍塌”

LangChain Agent默认max_iterations=15。我们没改,结果遇到一个特殊合同: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{field="amount", type="precision"},统计“修复前后精度变化”的比例。当这个指标突增,就说明修复逻辑有问题,需要人工介入。

现在,我们的监控面板有三块核心区域:

  • 上游健康度:OCR置信度分布、工具调用失败率
  • 监考有效性:各字段的parse_success_rate、fix_success_rate
  • 下游兼容性:各字段写入数据库的truncate_count、type_mismatch_count

这六条经验,每一条都来自真实的火线。它们不性感,不炫技,但能让你的结构化问答器,在千万次请求中,稳如磐石。

我在实际使用中发现,最难的从来不是让AI输出结构化数据,而是让整个系统在数据质量波动、模型行为漂移、业务规则变更的多重压力下,依然保持输出的确定性。这需要的不是某个神奇的prompt,而是一套像工业流水线一样精密、可监控、可演进的监考机制。当你把Agent当作一个需要被管理的“员工”,而不是一个可以被信任的“专家”时,结构化输出才真正从理想照进现实。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询