ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

Instructor实战:让大模型输出乖乖符合JSON Schema

Instructor实战:让大模型输出乖乖符合JSON Schema

调大模型的人应该都有过这种经历:Prompt 里明明白白写了"只返回 JSON",结果它还是给你夹带一段解释文本,或者字段名拼写跟 schema 不一致,又或者金额字段给你输出一个带逗号的"12,000元"。你拿正则去抠、拿 json.loads 去兜,搞完后还不敢保证下次一定干净。

这就是所谓"大模型强制结构化输出"要解决的问题。我在实际项目里把一个合同信息提取系统全部改成 Instructor 作为解析内核,接在 LiteLLM 和 LangChain/LangGraph 的外壳里,效果比之前靠 Prompt 硬抠稳定太多。这篇文章就把这套方案完整拆开讲一遍,包括原理、集成方式、实测踩坑和排查思路,适合已经在做 Agent 或信息抽取类应用、想彻底解决"模型输出不听话"问题的开发者参考。

1. 为什么要强制结构化:先回答"JSON Mode 不够用吗"

很多人第一反应是"直接用平台提供的 JSON Mode 不就行了"。确实,现在主流模型多少都支持 JSON 约束,甚至有些开源模型的指令遵循能力已经很不错。但这套方案在实际业务里有一个核心短板:它只约束了"输出格式是 JSON 这个大语法",并不保证"JSON 的内容符合我想要的 schema"。

打个比方,JSON Mode 相当于你让一个人"用中文回答",但没规定必须按表格来。他可能用一段叙述性的中文讲得天花乱坠,也可能把字段名换成同义词,比如你要求contract_no,它写成了serial_number,语义没问题,但下游代码data["contract_no"]直接 KeyError。更麻烦的是嵌套数组里面很容易中途漏字段。比如抽取合同条款列表,总共 12 条条款,它输出到第 8 条之后突然开始合并字段,第 9 到 12 条全部缺了amount。这种错误在纯 JSON Mode 下不会报错,只会在你数据后处理时静默出错。

还有一个隐蔽问题:平台内置的 JSON Mode 对返回值长度有限制。我实测过在长文本抽取场景下,输出超过一定长度后模型可能直接截断,留下半个右括号。拿到这段残缺 JSON 你没有任何机制让模型自我修复,除非在你的业务代码里再发一次请求把错误信息塞回去重新生成。而 Instructor 整个设计就是围绕"生成 -> 校验 -> 失败 -> 反馈重试"来做的,等于把"让模型自己纠正"这个能力变成了框架内置行为。

所以你会看到 Instructor 这类库解决的不只是"格式稳定",它事实上把 LLM 输出当成一个"可校验的程序返回值"来对待。你定义一个 Pydantic 模型等于定义了最终交付接口的契约,模型再强也得按这个契约走,不符合就反复重来,直到符合为止。这是工程上真正需要的"强制",而不是软绵绵的"建议"。

2. Instructor 的工作原理:一场从"自由发挥"到"照着表格填"的转变

先理解 Instructor 最核心的一个行为:它会在底层把 Pydantic 模型的字段约束转换成模型平台能理解的工具调用 schema,然后通过"工具调用"这个通道让模型以带参调用的方式输出结果。

这个设计很聪明。因为工具调用本身天然是结构化的,平台的 API 专门给这种返回做了训练和约束,所以比单纯文本生成模式下输出的 JSON 稳定得多。我们项目里对比过,同一份合同文本,普通生成模式的字段提取准确率大约在 82% 左右,切到工具调用通道后能到 94% 以上。差别主要在长字段和必填数字字段上。

它的核心流程可以总结为四步:

  1. 你声明一个继承自 Pydantic BaseModel 的响应模型,里面定义字段类型、默认值、校验逻辑甚至嵌套模型。
  2. Instructor 把响应模型转换成一个 JSON Schema,并通过 patch 机制包装你的模型客户端。
  3. 你调用create方法传入响应模型,模型根据 Schema 做工具调用返回原始结果。
  4. Instructor 拿到原始返回后用 Pydantic 做严格校验,校验不过就把异常信息作为新的消息追加到上下文里,让模型重新生成,直到成功或达到max_retries。

这第四步是多数人没用过的重试机制。举个例子,如果你的 Pydantic 模型里写了类似field_validator("date")要求日期不能晚于今天,模型返回了一个未来日期,Pydantic 校验会抛ValidationError,Instructor 自动把这个异常文本发给模型:"你刚才的输出的日期字段校验失败,错误是 X,请重新输出"。模型看到错误提示后通常会修正。这种"把校验器当作对话参与者"的做法,是它比 LangChain 内置的with_structured_output更"较真"的地方。

它还支持一个很实用的高级用法:llm_validator。普通 Pydantic validator 是你写死规则,比如正则以断言电话号码格式。但有些业务规则没法用正则表达,比如"合同方名称不能与招标代理机构名称相同"。这种场景可以定义一个llm_validator,让模型自己判断字段值是否满足语义约束,不满足就返回修正建议。Instructor 把这个 validator 也当成结构化输出的一部分,执行完字段校验后再用 LLM 做语义校验,实测在细粒度规则上能拦住不少模型"看着没问题其实语义错位"的返回。

最后补充一点:Instructor 也有异步版本,接口是AsyncInstructor,用法跟同步版一一对应。如果你的业务是在 FastAPI 异步路由里跑抽取任务,建议直接用异步版,避免阻塞事件循环。

3. 环境准备与一个最小可运行示例

理论讲完先上手跑通一个最小示例。安装很简单,核心依赖是 instructor 和 pydantic。我用的是 Python 3.11 + instructor 1.x 版本系列。

pip install instructor pydantic

然后写一个最基础的合同信息提取脚本。注意我这里直接把客户端对象your_client当作外部注入的变量,后面第四节会讲怎么接 LiteLLM,让它支持多种模型来源。

import instructor from pydantic import BaseModel, Field # 这里 your_client 是从你实际使用的模型平台获取的、实现了标准 Completions/工具调用接口的客户端对象 client = instructor.patch(your_client) class ContractInfo(BaseModel): contract_no: str = Field(description="合同编号") contract_date: str = Field(description="合同签订日期,格式 YYYY-MM-DD") party_a: str = Field(description="甲方全称") party_b: str = Field(description="乙方全称") total_amount: float = Field(description="合同总金额,单位元,只填数字") clauses: list[str] = Field(description="合同关键条款的要点列表") text = """ 采购合同 合同编号:HT-2024-0715 签订日期:2024年7月15日 甲方:某市信息中心 乙方:某软件有限公司 合同总金额:1,236,000.00元 第一条 乙方应当在合同生效后30日内完成系统部署... """ result = client.chat.completions.create( model="my-llm", response_model=ContractInfo, messages=[{"role": "user", "content": text}], max_retries=3, ) print(result.model_dump())

这里几个关键点要展开说。

response_model必须是 Pydantic 模型,字段描述尽量写详细。模型是看字段名和描述决定怎么往里面填的,描述越具体越不容易出现字段错配。比如我写total_amount字段描述时特意加了"单位元,只填数字",模型就会自动把文本里的"1,236,000.00元"清洗成浮点数,不依赖你额外写 transform。

max_retries是校验失败后的重试次数。我个人建议设 2 到 3,不要太高。重试次数太多在低水平模型上容易开始绕圈,表现为每次修正都会改掉其他本来正确的字段,最后陷入同一个错误来回改的循环。我后续会专门讲这个坑。

strict参数默认是 False。如果平台侧对工具调用支持严格的 schema 约束,你想完全禁用额外字段就可以设成 True。注意它在某些来源的模型上只是软约束,最终落地还是要靠 Pydantic 校验兜底。

跑通这个例子之后,你会发现返回的result不再是一个裸 JSON 字符串,而是一个已经校验过的 Pydantic 对象。后续拿result.contract_no取字段,类型都是安全的。这种体验才是工程上想要的:模型返回的"不可信字符串"边界到此为止,之后都是强类型对象。

4. 接入 LiteLLM:用一套代码轻松切换多家模型

实际开发里我很少绑定单一模型平台,原因有两个:一是不同任务需要不同能力级别的模型,抽合同用强模型、抽标签用弱模型能省不少成本;二是模型平台有时候接口不稳定,得有个备胎可以快速切换。LiteLLM 做的就是统一这些差异:它把几十种模型的 API 都包装成同一套 OpenAI 兼容接口,你只需要改一个model参数就能切换。

Instructor 跟 LiteLLM 配合有两种姿势,我先说最顺手的一种。前提是你已经把 LiteLLM 配好了,能够通过litellm.completion()拿到结果。Instructor 官方给出的方式是直接用instructor.from_litellm。

import Instructor from litellm import completion import instructor client = instructor.from_litellm(completion)

之后调用方式跟普通 create 基本一致,只需把model换成一个具体的 provider 模型标识。比如你在配置里配了本地方 ollama 模型,就可以这样写:

resp = client.chat.completions.create( model="ollama/qwen2.5:7b", response_model=ContractInfo, messages=[{"role": "user", "content": text}], )

这种方式的收益是:你的业务层代码完全感知不到模型来源差异,抽实体、抽关键词、抽表格,全部能用同一个response_model玩法。

第二种方式是把 LiteLLM 当纯路由,不借助from_litellm特殊封装,而是自己写一层适配。这样做的场景一般是:你只依赖 LiteLLM 做负载均衡和成本记录,但希望 Instructor 维持它原生对工具调用的强约束逻辑。其实from_litellm内部实现就是包装了 completion 的一个 adapter,少数情况下如果你对返回格式做了自定义,直接自己包一层也没问题。我试过在一个并发场景里把litellm.completion包成异步函数,再用instructor.AsyncInstructor去驱动,性能上没有问题。

这个环节最容易踩的坑有两个。第一个是部分模型源返回的原始内容里没有tool_calls字段。如果from_litellm包装后调用报错说拿不到 structure,多半就是这个问题。解决办法是先在 LiteLLM 那一层用mock_response或者先跑一个最普通的completion看返回字段是否完整。第二个坑是模型源如果不支持工具调用,那 Instructor 就会退化到"文本生成后解析 JSON",此时strict和重试机制虽然还在,但稳定性会明显下降。遇到这种情况,优先换一个支持工具调用的模型源,而不是在代码里跟它死磕。

LiteLLM 还自带一个很有用的能力叫做模型重试和回退。你可以配置成"主模型失败一定次数后自动切换备模型"。配合 Instructor 的max_retries时要小心,两者的重试是不同维度的:Instructor 是"校验失败后重发",LiteLLM 是"接口异常后换路"。建议把 LiteLLM 层的重试次数设低一点,比如 1 次即可,因为它在接口层面重试意义不大;把真正有业务价值的重试放在 Instructor 层。

5. 接入 LangChain:在 Agent 和工具链里保持结构化

很多项目里 LangChain 是被当作"编排层"来用的,比如大量 Agent、工具链、记忆机制都建立在它之上。这里要讨论的是:既然 LangChain 自己也有with_structured_output,为什么还要叠一层 Instructor?

我个人的判断是:with_structured_output解决的是"让模型输出符合某个 Pydantic 模型"这个需求,但它在校验失败时不会主动把错误反馈给模型让它自己修正,而是等你拿到结果后自行判断再手动重试。这意味着你仍然需要自己写失败处理逻辑。Instructor 的优势在于那个"反馈 + 重试"闭环,以及更丰富的llm_validator、Iterable、Partial 这类高级能力。如果模块内部复杂任务多、失败概率高,用 Instructor 做解析内核更省心。

在 LangChain 里用 Instructor 最常见的做法是把 LangChain 的 chat 模型对象直接 patch。LangChain 的ChatOpenAI等类内部大多是封装了平台的 SDK client,所以可以拿到底层对象再交给instructor.patch:

from langchain_openai import ChatOpenAI import instructor base_client = ChatOpenAI(api_key="...", base_url="...") patched = instructor.patch(base_client.openai_api_client)

然后你既可以在 LangChain 里用patched.chat.completions.create做直接调用,也可以在自定义 Tool 里把"调用模型并返回结构化对象"封装成一个小工具供 Agent 使用。我更倾向于后者:保持 LangChain 的 Agent 逻辑不变,只是在工具内部接入 Instructor。

举个例子,我做过一个"投标文件核查"Agent。它有一个工具是"提取标书里的技术要求条目",我用response_model=RequirementList做解析。Agent 决定调用这个工具时,返回的是一个包含结构化列表的 Pydantic 对象,而不是一份要二次解析的文本。这让后续的循环判断变得非常干净:Agent 可以直接len(result.items)来决定是继续追问还是进入下一环节,不用写正则。

如果你真的想把 Instructor 和 LangChain 的 chain 机制深度绑定,可以参考一个思路:把 Instructor 的 create 调用封装成一个@tool,然后在 LangGraph 或 LangChain 的序列里作为一环。Tool 的输入是文本,输出是已验证的 dict。这样编排逻辑全看 LangChain,数据校验全看 Instructor,职责非常清晰。

不过我要提醒一个反模式:不要让 LangChain 的output_parser和 Instructor 同时做一层 JSON 解析。有些人会把StrOutputParser()链在后面,结果模型输出被 LangChain 格式化后反而把原本 Instructor 校验过的对象又重新变成了字符串,体验瞬间倒退回"手工解析"时代。只要用了 Instructor,链条输出就直接给 Pydantic 对象,不要再加任何解析节点。

6. 接入 LangGraph:多步工作流里的结构化节点设计

LangGraph 比 LangChain 更进一步,它把工作流建模成有向图,节点之间通过共享的状态对象来传递数据。正因为状态是强类型的,结构化输出在这里价值更大:如果某个节点输出的是一段文本而不是结构化对象,下一个节点拿到的就是一团浆糊,条件分支根本没法写。

我建议的方式是:用 Instructor 处理每一个"需要跟模型交互并产出数据"的节点。节点输入一个状态 dict,调用client.create得到 Pydantic 对象,然后把对象model_dump()存入状态。这样每个节点之间的契约都是由 Pydantic 模型定义的,LangGraph 的StateSchema也能直接用这些模型做类型检查。

下面是一个简化示例,展示"合同审阅"两段式流程:第一步先提取合同关键字段,第二步根据字段判断风险等级。

from typing_extensions import TypedDict from langgraph.graph import StateGraph, END class ContractInfo(BaseModel): contract_no: str party_a: str party_b: str total_amount: float class RiskResult(BaseModel): risk_level: str = Field(description="low/mid/high") reasons: list[str] class ReviewState(TypedDict): raw_text: str contract: dict risk: dict patched_client = instructor.patch(your_client) def extract_node(state: ReviewState): contract = patched_client.chat.completions.create( model="my-llm", response_model=ContractInfo, messages=[{"role": "user", "content": state["raw_text"]}], ) return {"contract": contract.model_dump()} def risk_node(state: ReviewState): # 下个节点拿到的 state["contract"] 已经是干净的字典 risk = patched_client.chat.completions.create( model="my-llm", response_model=RiskResult, messages=[ {"role": "user", "content": state["raw_text"]}, {"role": "system", "content": f"已提取合同字段:{state['contract']}"} ], ) return {"risk": risk.model_dump()} graph = StateGraph(ReviewState) graph.add_node("extract", extract_node) graph.add_node("risk", risk_node) graph.add_edge("extract", "risk") graph.add_edge("risk", END) app = graph.compile() result = app.invoke({"raw_text": text})

这段代码里有几个工程细节值得展开。

首先是 State 里不要让 Pydantic 对象直接流通,而是用model_dump()转成普通 dict。原因是 LangGraph 的状态对象序列化逻辑可能对自定义类产生不必要的处理。你在节点里用 Pydantic 做全量校验,放入状态时转成 dict 完全够用,且后续节点读取时不会遇到类型问题。

其次是可以在节点之间传递"已提取的结构化信息"作为系统提示的一部分。上例中风险判断节点把前面抽取的合同字段作为上下文传入,模型回答风险结论时就不会偏离事实,这个"先结构化抽取、再基于结构化结果推理"的模式在多步 Agent 里至关重要。

有个实际场景能体现这个价值:我在某供应链项目里做了"物流单据 + 合同 + 验收报告"三份文件联合核验。先分别做三个提取节点,各自产出结构化字段,再进入一个比对节点。比对节点的 Prompt 只需引用三个 dict 里的关键字段做交叉校验,结果非常稳定。如果不用结构化输出,三个提取节点各返回一段文本,比对节点连字段对齐都做不到,更别提自动判断差异了。

LangGraph 还有条件边机制,也可以依赖结构化结果来做路由。比如风险节点返回的risk_level如果是 "high",就走向"人工复核"节点;否则走向"自动归档"节点。这种判断直接读state["risk"]["risk_level"]即可,完全不用解析文本。

7. 实战中踩过的坑与排查速查表

这部分是我最想写的。这套方案看起来顺滑,但在真实环境里跑起来总会遇到各种意外。我把一年多来在多个项目里遇到的典型问题都整理成一份速查表,方便你们对照排查。

现象可能原因排查与解法
返回的是普通 JSON 字符串而不是 Pydantic 对象patch 没生效,或调用的是未包装的客户端检查是否用了instructor.patch()或from_litellm()包装后的对象,不要再拿原生 client 调用
字段偶尔缺失,但校验不报错字段没有必填约束或没有默认值在 Pydantic 字段上加Field(..., description=...),必填字段不要给默认值;给全部字段写描述
重试次数耗尽,仍然校验失败模型能力不足以理解 validator 错误增大max_retries到 3~5,同时尝试把 validator 错误信息写得更具体;确认模型支持工具调用
数字字段出现"1,236,000.00"字符串模型的数值清洗能力不足字段描述里强调"只填数字",或用 PydanticBeforeValidator做清洗
同一模型同一输入两次结果不一致重试和版本选择造成偏差固定temperature=0,有条件的平台用seed固定随机数;校验逻辑里允许等价但不同表述
在 LangGraph 节点里调用 Instructor 报 Patch 错误重复 patch 了同一个 client把 patch 过的 client 放在模块顶部复用,不要在每个节点里重复包装
输出被截断未设置max_tokens或设置太小给 create 调用显式设置max_tokens,保守估计为期望 JSON 长度的 1.5 倍
字段名错配,模型把party_a填成partya模型对下划线命名不稳定用Field(alias="甲方全称")定义别名,或者在描述里写明字段含义
跟 LangChain 链式解析叠加后对象变了多了一层 output parser去掉多余的StrOutputParser或链上解析步骤

我再重点讲三个不常有人提的坑。

第一个是关于temperature的取向。有些人认为结构化输出纯提取任务应该直接把 temperature 设成 0,我同意大部分情况如此,但在创造性写作类字段提取上适度保留 0.2 到 0.3 会让摘要类字段更自然。关键是别让 temperature 太高,否则重试两次后模型有时候会跑偏。

第二个是llm_validator的排序问题。Instructor 的 validator 执行顺序依赖于字段定义顺序。如果有一个跨字段的一致性校验,比如"负责人不能是供应商联系人",必须定义为后置校验器model_validator(mode="after"),并且在校验失败信息里明确提示"请重新输出整个对象,不要只改这一个字段"。否则模型可能只修正一个副本字段,依然过不了另一边的校验。

第三个是 Ollama 或本地部署场景下的兼容性。本地模型对工具调用的 schema 遵循程度参差不齐。有些模型源返回的tool_calls结构跟 OpenAI 标准不一致,Instructor 解析时可能取不到内容。这种情况我不建议你在 Instrctor 层做太多 hack,更粗暴但有效的做法是:直接用instructor.from_litellm(completion, mode=instructor.Mode.JSON)强制走 JSON 模式。虽然稳定性略差,但至少能跑起来,后面再迭代找更适配的权重。

最后说一个关于性能的心得。Instructor 的每次重试都会把上一次的错误拼接进 messages 再发起一次完整请求,这意味着max_retries=3时最差要做 4 次调用。如果业务对延迟敏感,两道优化供参考:一是把max_retries降为 1,同时对字段描述做充分的防呆设计;二是用partial模式加流式解析,先在结果没完整时拿到一部分字段,减少整体等待。具体来说就是给create加一个stream=True参数,等所有字段校验完再返回,体验类似流式打字,能显著缓解首字延迟。

我之前在一个客服工单分类项目里试过把重试从 3 降到 1,配合高质量字段描述,准确率只掉了 1 个百分点左右,但平均延迟几乎降了一半。从那以后我就形成了一个习惯:优先通过"把字段描述写到足够清楚"来减少模型犯错的概率,而不是完全依赖重试机制兜底。毕竟每次重试都不只是时间成本,还有 token 成本,在数据量大的场景里这是一笔可观的资源消耗。

回到开头那个合同提取系统,我们最终把 Instructor 通过 LiteLLM 接到不同来源的模型上,同时在 LangGraph 里编排了"提取 -> 核验 -> 风险判断 -> 路由"四段流程。跑了大半年,最直观的感受是以前需要我反复维护的正则和 JSON 修复函数全删干净了,出错日志里"字段缺失"这类问题也基本绝迹。这个方向并不是什么魔法,它只是把"输出结构化数据"从模型的可选项变成了必选项,而工程上我们对必选项的要求本来就不该是碰运气式地碰对一次,而是每一次都必须符合约定。

分享一个小技巧作为结尾:写 Pydantic 字段描述时,不要只写"合同编号",而是写"合同编号,通常以 HT 开头,格式如 HT-2024-001"。描述给的信息越像"你手把手教一个新人怎么填表",模型返回的字段就越精准。这套系统后续还可以扩展的方向,是基于llm_validator做业务规则引擎,把用户自定义的校验规则也换成 Prompt 驱动,不过那是另一个话题了。你们如果在自己的 Agent 项目里反复被模型输出格式折腾,不妨从一个小响应模型开始,把这条链路先搭起来,收益会非常直接。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进