LangChain结构化输出实战：从文档解析到Agent调用

我最早做 RAG 项目的时候最头疼的还不是检索效果差而是模型回答完我怎么把答案塞进下游系统。你问这份合同里甲方是谁、违约金比例是多少模型可能给你一段带着 Markdown 标题的散文也可能给你一份换行混乱的伪 JSON。后来我转向 LangChain接触到结构化输出structured output这个概念才意识到问题的根源不是模型不够聪明而是我从来没有给模型的输出划定边界。这篇内容没有高深理论就是一套我反复验证过、可以直接抄作业的结构化输出方案。我会从为什么必须做结构化讲起接着对比几条技术路线的取舍然后给一个完整案例——把招标文件解析成带页码、章节、段落的层级 JSON最后再聊几个我实际踩过的坑。这篇适合谁正在做文档解析、RAG 数据入库、Agent 工具调用参数提取的开发者以及刚接触 LangChain 想知道with_structured_output到底怎么用的朋友。1. 为什么非要把 LangChain 的输出钉成结构先讲一个我自己的教训。之前做一个合同审查辅助工具第一步是让模型抽取合同的基本信息签署方、金额、有效期、违约责任条款。早期实现非常朴素就是把合同文本拼进 Prompt然后让模型用 JSON 格式回答。看起来没什么问题可真到了批量跑的时候同一个模型今天输出标准的 JSON明天可能在 JSON 外面包一层 Markdown 代码块字段名不稳定昨天叫party_a今天模型心情好叫甲方最致命的是嵌套层级模型自己发挥把amount和currency合并成amount: 100万人民币然后下游入库直接报类型错误。后来我只能写正则把 Markdown 代码块剥掉再写一堆防御逻辑去猜字段。那段时间我意识到一个道理**自由文本生成和结构化数据提取本质上是两件事。**前者追求语义丰富度后者追求确定性。你让模型在无限维度上自由发挥却想拿到一个固定 Schema 的数据这本就是一种概率赌博。1.1 自由文本的代价我踩过的正则清洗我见过很多团队在模型输出后面挂一段清洗脚本处理逻辑长到几百行什么json.loads失败就strip再失败就找{和}的位置截取。实话说在只有几十条样本时这种方案能跑但一旦数据量上到几千份文档你就会发现问题根本不可控模型返回的字符串里嵌套了多余的逗号JSON 解析失败字段值里有 HTML 标签清洗规则写不完日期格式时而是2024年3月时而是2024-03-05程序只能靠猜更麻烦的是这些错误往往是随机出现的你修好一个下一个文档又冒出来新的。正则清洗的本质是在给模型的不可控行为打补丁。而结构化输出的做法完全不同它不等到模型生成完再收拾而是在生成之前就用 Prompt 和调用协议把所有约束传达给模型。这样你拿到的就是一份可以直接解析的数据而不是一篇需要二次加工的散文。1.2 结构化输出到底解决了什么问题简单说结构化输出解决的是模型输出与程序输入之间的对齐问题。它做了三件事约束结构通过 Pydantic 模型或 JSON Schema 预先定义字段名、字段类型、嵌套关系模型必须在这个框架内填空。约束语义每个字段都可以写 description告诉模型这个字段里到底该填什么避免金额和违约金比例被塞进同一个字段。约束校验输出最后会被 Pydantic 解析和校验类型不匹配、缺字段都能在程序层面暴露出来而不是吞进日志里。这套思路最典型的落地场景除了合同、招标文件解析还有文档解析后入库把实施方案、合同、招标文件解析成包含页码、章节、段落的层级结构方便后续做结构化检索数据抽取从测试报告里批量提取指标直接对接报表系统Agent 工具调用让模型决策后的输出直接变为工具入参不经过自然语言中转RAG 前置处理把非结构化文本先转成结构化字段再做向量化和过滤检索。2. 三条结构化输出技术路线以及我为什么首选 with_structured_outputLangChain 生态里做结构化输出的方式看似多核心其实就三种Pydantic 模型、TypedDict 类型注解、直接给 JSON Schema。我日常 90% 的情况用 Pydantic剩下 10% 用 TypedDict 处理快速原型。下面拆开讲。2.1 第一种Pydantic 模型 with_structured_output这是我最推荐的方式。先用 Pydantic 定义结构然后通过llm.with_structured_output(schema)拿到一个绑定了结构的调用对象。代码大概是这样的from typing import List, Optional from langchain_core.pydantic_v1 import BaseModel, Field class ContractParty(BaseModel): 合同一方的主体信息 name: str Field(description公司全称或自然人姓名) role: str Field(description甲方或乙方) unified_social_code: Optional[str] Field( description统一社会信用代码没有则填空字符串 ) class Contract(BaseModel): 合同概要信息 contract_no: str Field(description合同编号) title: str Field(description合同标题) parties: List[ContractParty] Field(description合同签署方列表) total_amount: Optional[float] Field(description合同总金额单位万元无法识别则填 0) effective_date: Optional[str] Field(description合同生效日期格式 YYYY-MM-DD) key_clauses: List[str] Field(description关键条款摘要每条约 50 字以内)定义好之后对接模型from langchain_openai import ChatOpenAI from langchain_core.output_parsers import PydanticOutputParser llm ChatOpenAI(modelgpt-4o, temperature0) structured_llm llm.with_structured_output(Contract) resp structured_llm.invoke( 请从以下合同中提取结构化信息\n 合同编号 HT-2024-001甲方是北京某某科技有限公司乙方是上海某运营服务有限公司 合同总金额为 230 万元有效期为 2024 年 4 月 1 日至 2025 年 3 月 31 日…… ) print(resp.model_dump())注意我用的是langchain_core.pydantic_v1。不是因为我怀旧而是当时项目里很多依赖还停留在 Pydantic v1 API。你现在新开项目用langchain_core.pydantic_v2也没问题关键是统一版本不要混用这个坑后面细说。with_structured_output这个方法是最值得花时间研究的入口。LangChain 官方在 0.1.x 版本之后把原来散落的各种OutputParser收敛到了这个方法上它会根据你使用的模型能力和传入的 Schema 类型自动选择底层调用策略。2.2 第二种TypedDict / JSON Schema dict如果你只是临时想抽两三个字段不想为一件事专门建一个 Pydantic 类可以用 TypedDict。它写起来更轻但校验能力弱很多from typing import TypedDict, List class ClauseItem(TypedDict): clause_no: str content: str class DocOutline(TypedDict): title: str clauses: List[ClauseItem] structured_llm llm.with_structured_output(DocOutline)这背后的原理LangChain 会把 TypedDict 转成 JSON Schema 传给模型。这也引出了第三种方式。2.3 底层三种模式怎么选with_structured_output只是一个门面真正的执行策略取决于模型供应商。我用表格总结一下调用模式适用模型原理优点缺点Function/Tool CallingOpenAI、Claude、GLM-4、Qwen 等支持工具调用的模型模型先输出工具调用参数再通过解析拿到结构化数据结构严谨字段约束力最强支持嵌套较深的结构模型不支持工具调用时直接报错JSON Mode部分云厂商专用 API模型被强制输出合法 JSON但结构仍靠 Prompt 约束兼容性比工具调用好一些字段名、类型约束相对弱偶尔会多字段或少字段JSON Schema Prompting任意通用对话模型在 Prompt 中嵌入 JSON Schema让模型照着生成任何模型都能跑成功率随模型能力波动本地小模型经常翻车我的选型建议非常简单如果模型支持 Function Calling永远优先走这条如果不支持就退化为 JSON Mode如果模型再弱一些结构化输出成功率会肉眼可见地下降这时你要考虑换模型而不是继续优化 Prompt。另外不要试图在同一个 Schema 里放超过 20 个字段。字段越多模型出错的概率指数上升。我倾向于把大任务拆成多次抽取例如先抽文档层面的标题、编号、日期再针对每个章节做细粒度抽取。3. 实战案例把招标文件解析成带页码、章节、段落的层级结构这部分是全文最核心的实操内容。很多团队做文档解析最终目标是让模型输出实施方案、合同、招标文件的结构化文本包含页码、章节、段落这个需求本质上就是层级结构抽取。我会用一个简化但完整的例子走通全流程。3.1 目标结构与数据分析招标文件通常是几百页 PDF里面有封面、目录、正文、附件。要做结构化我们关心的不是每一个字符而是第几条、第几页、什么内容。所以我设计了四层结构DocumentMeta文件名、总页数、招标编号、招标人Part一级章节如第一章 招标公告Section二级章节记录章节号、标题、起始页码Paragraph正文段落带页码、段落序号、文本内容。在动手写代码之前先抽样分析 5 份文档搞清楚常见结构。这一步很容易被跳过但它决定了你的 Pydantic 模型长什么样。我见过有人不理解数据就先写字段结果抽出来的字段一半是空的。分析完数据结构就可以定义模型了。from typing import List, Optional from langchain_core.pydantic_v1 import BaseModel, Field class Paragraph(BaseModel): paragraph_no: int Field(description段落在章节内的序号从 1 开始) page_number: int Field(description段落所在页的页码) content: str Field(description段落正文内容保留原始表述) class Section(BaseModel): section_no: str Field(description章节编号如 2.1) title: str Field(description章节标题) start_page: int Field(description章节起始页码) paragraphs: List[Paragraph] Field(description章节下的所有段落) class Part(BaseModel): part_no: int Field(description一级章节序号) title: str Field(description一级章节标题) start_page: int Field(description一级章节起始页码) sections: List[Section] Field(description该章节下的二级章节列表) class DocumentStruct(BaseModel): file_name: str Field(descriptionPDF 文件名) total_pages: int Field(descriptionPDF 总页数) bidding_no: Optional[str] Field(description招标编号从封面页提取) tenderer: Optional[str] Field(description招标人名称) parts: List[Part] Field(description文档的章节结构)这里有几个设计细节值得说明paragraph_no和page_number分开存。前者表示逻辑顺序后者用于复制引用检索时需要精确定位到页码。所有字段都给了 description这不是废话而是给模型的填表说明书。模型是靠 description 理解该往哪里填的字段名只是符号。3.2 核心实现代码读取 PDF 我推荐pypdf或pdfplumber这一步我们不做解析原文字数限制直接把文本按页拆开缩成若干块喂给模型。注意长文本直接全量塞进模型上下文结构化输出大概率会翻车因为模型的注意力会被长内容稀释。我的做法是先用简单规则做粗切分按章节编号切块每块调用一次结构化输出。import pdfplumber def extract_pages(pdf_path: str): pages [] with pdfplumber.open(pdf_path) as pdf: for i, page in enumerate(pdf.pages, start1): try: text page.extract_text() or except Exception: text pages.append({page_number: i, text: text}) return pages def chunk_document(pages, max_chars1600): chunks [] current current_pages [] for page in pages: if len(current) len(page[text]) max_chars and current: chunks.append({text: current, pages: current_pages}) current current_pages [] current page[text] \n current_pages.append(page[page_number]) if current: chunks.append({text: current, pages: current_pages}) return chunks切片之后对每个 chunk 做结构化抽取。这里有个细节我是让模型返回这个 chunk 里的所有章节结构而不是整个文档结构。因为每个 chunk 只包含局部信息让模型一次生成全量结构就会产生幻觉尤其是页码这种需要精确定位的字段。llm ChatOpenAI(modelgpt-4o, temperature0) structured_llm llm.with_structured_output(DocumentStruct) def parse_doc(pdf_path: str): pages extract_pages(pdf_path) chunks chunk_document(pages) all_parts [] for c in chunks: prompt f 你是一个招标文件结构化解析引擎。下面是 PDF 文档的一部分内容 请提取其中的章节结构严格按照输出 Schema 填入。 注意 - 页码必须是该内容实际出现所在的页码 - 内容缺失的字段填空字符串或空列表不要编造 - 段落内容尽量保留原文 文档内容第 {c[pages][0]} 页到第 {c[pages][-1]} 页 {c[text]} try: result structured_llm.invoke(prompt) # 只取这个 chunk 内的 parts不覆盖全量 if result.parts: all_parts.extend(result.parts) except Exception as e: # 记录错误 chunk便于后续修复 print(fchunk 解析失败: {e}) return all_parts运行之后你会拿到一个列表里面是各个 chunk 抽出的Part对象。最后再做一次合并去重、按part_no排序输出成 JSON 文件落库。这样做到底有什么好处最直接的一点下游检索可以按章节过滤 按页码定位。比如用户问第三章里关于投标保证金的条款在哪一页你直接查Section.title包含投标保证金的记录返回start_page和paragraph.content答案带出处用户信任度完全不同。3.3 扫描件另说OCR 之后的处理上面例子中的 PDF 是文本型 PDF也就是文字可以直接提取的。但很多旧合同的扫描件根本提不出文字必须先过一遍 OCR。热搜里那个文档解析能输出实施方案、合同还有招标文件的结构化文本包含页码、章节、段落的需求落到线下就是先用 OCR 再走结构化抽取。OCR 引擎我常用 PaddleOCR 这一类它会把页面转成一行行的文本附带每个文本块的位置坐标。结构化抽取对 OCR 文本的依赖其实非常高OCR 出来的文字越碎、错字越多后续结构化输出的准确率下降就越明显。所以做扫描件时我通常分两层处理第一层用 OCR 把版面还原成段落页面的中间态这一步主要靠位置坐标合并相邻文本块第二层再交给结构化输出模型做层级提取。中间态的还原质量决定了上层抽取得成败这一步别偷懒。4. 落地过程最容易踩的五个坑以及完整的排查思路我能写满一页踩坑记录。这里挑五个典型的按我真实遇到的顺序来讲每个坑都附排查链路你遇到同样问题时可以照着一模一样地走一遍。4.1 坑一模型根本不支持函数调用现象调用with_structured_output后正常提示词能回复但一旦换成结构化输出就异常不是超时就是解析报错。排查链路打开 LangChain 的调试日志LANGCHAIN_TRACING_V2true或者直接在代码里import langchain; langchain.debug True看请求真正发出去的结构检查模型 API 返回的 raw response 里是否有tool_calls字段如果是通过 Ollama 跑本地模型先确认模型本身是否支持 tool calling不少本地模型是在最新版本才支持你用了一个月前的镜像就会踩中如果确认不支持立刻改方案换一个支持工具调用的模型或者降级为 JSON Mode。我当时用一台测试机器跑本地 Qwen 2.5 7BLangChain 调用时报Function calling is not supported by this model我还以为是代码写错了后来查文档才发现是模型能力边界。排查的诀窍就是**先把模型能力和框架能力分开验证。**用一个最简单的 Prompt 和带两个字段的 Pydantic 模型测缩小问题范围。4.2 坑二字段缺失和值为空现象调with_structured_output不报错但返回的 Pydantic 对象里Optional字段全是None该填的金额、日期全是空。排查链路优化 System Prompt把字段要求明确写出来例如总金额必须填数值识别不到就填 0检查输入文本是否被截断尤其是长文档分块时字段出现在 text 末尾模型把后半段忘了观察模型返回的中间结果看它是压根没输出还是输出了但被 Pydantic 拒绝了。这个坑的本质是模型的遵从性问题。增强遵从性最有效的不是把 Prompt 写得更长而是做字段级校验和补充抽取。我通常的做法是先进行一次全量抽取然后写一段校验代码把所有空字段收集起来再调用一次模型专门补抽这些字段。这样比你反复调 Prompt 效率高得多。4.3 坑三列表字段退化成空数组现象明明文档里有很多签署方返回的parties只有一条或者干脆是空列表。这里有一个反直觉的经验列表字段比单值字段更容易出错。为什么因为模型在生成结构化输出时倾向于回答简短、别遗漏字段而不是尽量多列几项。它一旦觉得上下文里信息不够就直接返回空列表而不是继续找。排查链路检查 Pydantic 字段的 description 里是否写了列表必须包含所有项如果没有加上在 Prompt 里给模型一个列表项数量的下限比如至少识别出 2 个签署方如果还是不行把列表拆出来做二次抽取单独让模型提取所有公司名称。给 list 字段加示例是最实用的做法。Pydantic 的 Field 支持直接传 exampleLangChain 会把 example 编进 Schema 传给模型这比纯文字描述更稳定。4.4 坑四流式输出的碎片化问题现象开了streamTrue之后拿到的输出是一段段残缺字符串根本没法组成完整 JSON。很多人在做对话型应用时想兼顾流式体验和结构化输出但这两个诉求在底层是矛盾的。流式输出面向的是文本 token 流结构化输出面向的是完整对象实例。我的做法是不要在同一个链路里同时追求两者。对话界面先用普通流式输出做展示给用户即时反馈后台悄悄调用structured_llm.invoke()生成正式结构化结果。或者反过来先流式输出完整 JSON 文本给前端展示前端自己JSON.parse后端拿到解析成功的对象。实测下来前者更稳因为超时控制更好做。4.5 坑五Pydantic v1/v2 的版本摩擦这是 LangChain 社区讨论度极高但文档里不显眼的问题。LangChain 在不同版本里对 Pydantic 的依赖经历过一次大切换0.2.x 之后很多模块默认用 Pydantic v2但你要是看到老的教程使用langchain_core.pydantic_v1又有新的依赖包强制要求 pydantic v2两者混用就会出现诡异问题字段定义正常但校验时莫名报model_dump不存在或__fields_set__异常。排查链路pip list | grep pydantic python -c import pydantic; print(pydantic.VERSION)统一策略新项目全部走pydantic.v2.BaseModel或者统一走langchain_core.pydantic_v1但不要在一个进程里混用两套。如果你在一个旧项目里迁移优先把强制依赖 pydantic v2 的包独立成微服务否则版本冲突会消耗你大量时间。还有一个隐藏点LangChain 底层做 output parsing 时对 Pydantic v2 的解析依赖model_validate对 v1 依赖parse_obj。你的自定义输出解析器如果自己写了兼容逻辑两套都要支持否则一侧跑通一侧报错。5. 进阶玩法批量抽取、容错重试以及和 LangGraph 的配合当你把单个文档的结构化输出跑通接下来就是工程化的挑战。这部分我只讲两件事如何让批量任务更稳以及为什么在 LangGraph 的 Agent 编排里结构化输出是不可或缺的一环。5.1 批量并发抽取一个招标文件按 chunk 切分可以产生 50~100 个块。如果串行调用 GPT-4o一次要几分钟。批量并发是必须做的。LangChain 的Runnable天然支持.batch()tasks [ {prompt: build_prompt_for_chunk(chunk)} for chunk in chunks ] results structured_llm.batch(tasks, config{max_concurrency: 5})并发数不要贪多。你调的是云 API并发太高容易触发限流而且一旦超时会连带报错。我一般根据模型供应商的限制来定云厂商的 API 我控制在 5~10Ollama 本地部署我控制在 16 以内先跑一次小批观察平均响应时间和失败率再调整。5.2 容错重试机制批量任务跑起来后一定会出现个别 chunk 失败。不处理失败就入库坏数据会污染下游检索。我的重试策略分三层即时重试单个调用异常时延时 2 秒重试最多 3 次字段补抽调用成功但 Pydantic 校验发现空字段单独构造补抽任务人工落盘前两层都失败时把原始 chunk 和报错信息写入待处理队列而不是直接丢弃。import time from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max8)) def safe_structured_invoke(llm, prompt): result llm.invoke(prompt) # 业务校验 if not hasattr(result, parts) or result.parts is None: raise ValueError(missing parts field) return result用tenacity库做重试是比较省心的方案把重试逻辑和业务代码隔离避免到处写 try-except。5.3 与 LangGraph 的配合如果你已经在用 LangGraph 编排多步骤 Agent结构化输出会出现在两个关键位置。第一个位置是状态传递。LangGraph 的节点之间通过状态字典传递数据如果状态里装的是字符串下一步节点的代码就得先做一把逐字解析如果某个节点输出的是 Pydantic 对象下一个节点直接拿属性用就行。我在 Graph 定义里会明显把LLM 调用的原始文本输出和结构化对象分成两种状态槽位防止字符串满天飞。第二个位置是工具参数生成。Agent 决定调用某个工具时LangChain 底层通过 tool calling 把模型输出绑定到工具入参。这和结构化输出的原理是同一条链路模型被约束在工具定义的参数 Schema 内生成 JSON。所以你在写 Agent 工具时参数定义一定要写得严谨description 越具体模型选参数时的准确率越高。我见过不少团队把工具参数定义成自由字符串然后又在工具内部做正则解析这相当于是把结构化输出的责任从模型手里又拿回了自己手里。如果你还在对比 LangGraph 和 LangChain 的关系可以这样理解LangChain 提供的是模型能力封装LangGraph 提供的是流程编排能力。结构化输出属于前者但它的价值在后者里会被放大。一个节点负责抽结构化结果下一个节点基于这个结果做判断最后再路由到不同分支整个链路比纯文本传递要可控得多。最后分享一点个人体会我实际用下来的感受是结构化输出这件事最容易出成绩的不是模型选择也不是 Prompt 技巧而是 Schema 设计。设计得好的 Schema字段数量克制、类型简单、description 像说明书一样清晰模型填起来几乎不会错。设计得不好的 Schema字段叠到三四十个嵌套四五层什么模型来了都蒙。所以我现在的习惯是拿到一个新需求先不写代码先在纸上画一个 JSON 样例想象模型输出的结果应该长什么样然后再转成 Pydantic 模型再写 Prompt。整个过程十分钟但能省掉后面两小时的调试时间。你也不妨试试这个顺序结构化输出的体验会完全不同。