lang

1. 本章目标 #

前几章拿到的多是自由文本（AIMessage.content）。第 3 章末尾预览过 with_structured_output；第 5 章的模板管「怎么问」。业务里却常要：

把客服投诉变成可入库的工单字段
从用户口述里抽出订单号、商品、数量
下游代码直接读 ticket.urgency，而不是用正则去「猜」一段话

聊天回复给人类看，自由文本够用；一旦要进数据库、走工单、触发路由，格式一乱就麻烦。本章要解决的就是：

用 Pydantic / Structured Output 把模型输出约束成可校验对象，并学会处理解析失败。

学完你应能：

用 Pydantic BaseModel + Field 定义输出 schema
说清 with_structured_output 底层做了什么，并能把发给模型的工具定义打出来排查
用 model.with_structured_output(...) 做抽取，包括嵌套结构与可选字段
把 Prompt 模板与结构化模型拼成抽取链
用 include_raw、校验约束、field_validator 与重试处理失败
避开三个静默坑：模型编造必填字段、可选字段被填成字符串 "null"、重试时撞上 400
了解 Agent 的 response_format（为第 9 章铺路）

本章实战：订单抽取器 + 工单抽取器。

参考文档：

2. 为什么需要结构化输出 #

模型擅长「把意思说清楚」，但不保证「字段名、类型、枚举值永远一致」。
下游要的是稳定契约，不是一段通顺的中文。

自由文本：
  「用户很着急，好像是物流问题，快递三天没动……」
        │
        ▼  手写正则 / 再问一次模型 / 容易碎
下游业务（入库、路由、报表）很难稳定消费

结构化对象：
  Ticket(category="物流", urgency="高", summary="快递三天未更新")
        │
        ▼  直接 .category / model_dump() / 校验失败即重试

对比	自由文本	结构化输出
下游使用	要再解析	字段可直接用
格式稳定性	模型爱加废话	schema 约束字段与类型
校验	靠人工约定	Pydantic 运行时校验
典型场景	聊天回复、摘要给人类看	抽取、分类、填表、进库

可以记成这样的分工：

消费者	更合适的输出
人（客服、运营、用户）	自然语言
程序（DB、API、规则引擎）	结构化对象

一句话：

给人类看用自然语言；给程序用结构化输出。

3. 用 Pydantic 定义 Schema #

结构化输出的第一步不是调模型，而是先想清楚「希望拿到什么结构的数据」。

Pydantic 的 BaseModel 就是这份「结构说明书」。LangChain 做三件事：

把 schema 转成 JSON Schema，作为工具定义发给模型
强制模型调用这个「工具」，于是模型必须按该结构填参数
把模型填的参数交给 Pydantic 校验，得到实例（失败时抛错 / 带回 parsing_error）

所以：schema 写得好，成功率明显高于「只在 prompt 里口头约定 JSON」。

3.1. 先看清楚模型到底收到了什么 #

这一节值得花两分钟：后面许多建议都建立在「模型到底看到了什么」上——看不清就只能靠猜。

with_structured_output 不是魔法，返回的是一条链：

import json
from typing import Literal

from langchain.chat_models import init_chat_model
from pydantic import BaseModel, Field
from dotenv import load_dotenv
from rich import print

load_dotenv(override=True)


class Ticket(BaseModel):
    """客服工单摘要。"""

    category: Literal["退款", "物流", "咨询", "其他"] = Field(description="问题类别")
    urgency: Literal["低", "中", "高"] = Field(
        description="紧急程度；用户表达着急、投诉升级时偏高"
    )
    summary: str = Field(description="一句话摘要，不超过 30 字")


# 模型初始化的细节见 §4.1，这里先用起来
model = init_chat_model(
    "deepseek:deepseek-v4-flash",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)

extractor = model.with_structured_output(Ticket)
# 注意：返回的不是模型，而是一条链
print(type(extractor).__name__)  # RunnableSequence
for step in extractor.steps:
    print(step)
# with_structured_output 内部就是 bind_tools + 解析器，工具定义藏在第一步的 kwargs 里
print(json.dumps(extractor.steps[0].kwargs["tools"], ensure_ascii=False, indent=2))

extractor 已不是单纯的聊天模型，而是一条 RunnableSequence：先绑定工具，再解析输出。extractor.steps 就是这条链里的步骤列表：

steps[0]：绑定工具
模型经 bind_tools() 包装后的对象（多为 RunnableBinding）。with_structured_output(Ticket) 会把 Pydantic 的 Ticket 转成 Tool 定义，强制模型按该 JSON Schema 填参数。打印 kwargs["tools"]，就是这份完整工具定义——能直接看到类字段如何变成模型认识的 parameters（含 category、urgency、summary 的类型、描述与必填项）。
steps[1]：输出解析
通常是 PydanticToolsParser。模型返回的是 JSON 形态的 tool 参数；这一步把它解析成 Ticket 实例，后续即可直接用 ticket.category。

这段不会真正调用模型（只打印绑好的工具定义），所以不花钱、也不依赖网络。

第二个 print 的输出，就是模型真正收到的约束：

[
  {
    "type": "function",
    "function": {
      "name": "Ticket",
      "description": "客服工单摘要。",
      "parameters": {
        "properties": {
          "category": {
            "description": "问题类别",
            "enum": ["退款", "物流", "咨询", "其他"],
            "type": "string"
          },
          "urgency": {
            "description": "紧急程度；用户表达着急、投诉升级时偏高",
            "enum": ["低", "中", "高"],
            "type": "string"
          },
          "summary": {
            "description": "一句话摘要，不超过 30 字",
            "type": "string"
          }
        },
        "required": ["category", "urgency", "summary"],
        "type": "object"
      }
    }
  }
]

对照 Python 代码和这份 JSON，映射关系一目了然：

Python 里写的	变成了 JSON Schema 的
类名 `Ticket`	`function.name`
类 docstring `"""客服工单摘要。"""`	`function.description`
字段名 `category`	属性名
`Field(description="问题类别")`	该属性的 `description`
`Literal["退款", ...]`	`enum` 数组
`str` / `int` / `float`	`type`
`Field(ge=1, le=5)`	`minimum` / `maximum`
没有默认值的字段	出现在 `required` 里

这就是为什么 description 比字段名重要：模型看到的是这份 JSON，description 几乎是它唯一的说明。字段名叫 urgency 还是 u，影响远小于 description 写得清不清楚。

也可以不经 LangChain，直接看 Pydantic 生成的原始 schema：

print(json.dumps(Ticket.model_json_schema(), ensure_ascii=False, indent=2))

和上面的工具定义几乎一样，只多了 title（LangChain 转成工具定义时会去掉，对模型没用）。

调试技巧： 抽取不对时，先把工具定义打出来。枚举拼错、描述写错位置、可选字段误进 required，看一眼 JSON 往往就清楚，比反复改 prompt 快得多。

3.2. Schema 怎么写才好用 #

写法	作用
类型注解 `str` / `int` / `list[str]`	约束字段类型
`Literal["a", "b"]`	枚举，分类题首选
`Field(description="...")`	写给模型看的说明，比字段名更重要
`Field(ge=1, le=5)`	数值范围等校验
`xxx 或 None`	允许缺失（信息不足时别瞎编）
类 docstring	整体说明这个结构是干什么的

几个容易漏掉的点：

字段名给程序员看，description 给模型看。 urgency 不如写清「用户着急、投诉升级时偏高」——见 §3.1，模型看到的就是这句话。
分类题优先用 Literal。 会变成 enum，比让模型自由填「类别」稳得多。
不知道就别逼它编。 见下面 §3.3，这是本章最容易造成脏数据的地方。
范围约束能挡脏数据。 例如评分 ge=1, le=5 会变成 minimum/maximum 一并发给模型，多数时候模型会自己收敛（§7.2 有实测）。

原则：

字段描述写清楚「含义 + 取值规则」；不确定的字段声明成可选，并给默认值。

3.3. 可选字段：一个字符之差，语义完全不同 #

「这个字段可能没有」听起来简单，写法却有三种，生成的 schema 差别很大。把三种都打出来对比：

from typing import Optional

from pydantic import BaseModel, Field


class A(BaseModel):
    """带 default=None。"""
    order_id: str | None = Field(default=None, description="订单号")


class B(BaseModel):
    """只写 | None，不给 default。"""
    order_id: str | None = Field(description="订单号")


class C(BaseModel):
    """Optional 但给了 default。"""
    order_id: Optional[str] = Field(default=None, description="订单号")


for cls in (A, B, C):
    s = cls.model_json_schema()
    print(f"{cls.__name__}: required = {s.get('required', [])}")

输出：

A: required = []
B: required = ['order_id']
C: required = []

B 是个陷阱。 只写 str | None 却不给 default，Pydantic 会当成「字段必须出现，值可以为 null」——于是进了 required。模型见必填，就一定会填点什么。实测 B 的行为：

输入：买了个键盘，已经付款了。
模型返回参数: {'order_id': ''}
parsed: order_id=''

它填的是空字符串，不是 None。下游若写 if order.order_id is None 就会漏掉。

Optional[str] 与 str | None 等价（Optional 是旧写法），差别只在有没有 default。规则很简单：

可选字段一定要给 default=None。 类型上写 | None 只允许 null；有了 default=None 才真正可选。

3.4. 必填字段在信息不足时会被编造 #

上面说「该可选的要声明可选」，不是洁癖。看必填 schema 遇到无关输入时的表现：

from langchain.chat_models import init_chat_model
from pydantic import BaseModel, Field
from dotenv import load_dotenv

load_dotenv(override=True)

# 模型初始化的细节见下面 §4.1，这里先用起来
model = init_chat_model(
    "deepseek:deepseek-v4-flash",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)


class Strict(BaseModel):
    """严格订单。"""

    # 两个字段都没有默认值，所以都是必填
    order_id: str = Field(description="订单号，必填")
    amount: float = Field(description="金额，必填")


extractor = model.with_structured_output(Strict)
# 这段输入里既没有订单号也没有金额
print(extractor.invoke("我想问一下退货政策。"))

输出：

order_id='12345' amount=99.9

原文既没有订单号也没有金额，模型却凭空造出 12345 和 99.9。而且不报错——对 Pydantic 来说，这是一条完全合法的数据。

这是结构化输出最危险的失败模式：它不会说「我不知道」，而会给你一个格式完美的假答案。 格式校验能挡类型错误，挡不住编造。办法只有在 schema 设计阶段，把「可能没有」的字段声明成可选。

4. 最小闭环：工单抽取器 #

承接第 3 章的预览，先跑通最小例子：用户说一句投诉，程序拿到 Ticket 对象。

整条路径只有三步：

定义 Ticket schema
        │
        ▼
model.with_structured_output(Ticket)   → 得到 extractor
        │
        ▼
extractor.invoke(用户原话)             → Ticket 实例

本文默认 DeepSeek：结构化输出前通常要 关闭 thinking，并建议 temperature=0，否则容易报 400 或枚举乱跳。

4.1. 1.ticket_extractor.py #

# 从 typing 导入 Literal，用于把字段取值限制为若干固定标签
from typing import Literal

# 从 pydantic 导入 BaseModel 与 Field，用于定义带校验的数据结构
from pydantic import BaseModel, Field

# 从 langchain.chat_models 导入 init_chat_model，用于初始化聊天模型
from langchain.chat_models import init_chat_model

# 从 dotenv 导入 load_dotenv，用于加载环境变量
from dotenv import load_dotenv

# 加载 .env；override=True 表示覆盖已存在的同名变量
load_dotenv(override=True)


# 定义客服工单摘要的数据结构
class Ticket(BaseModel):
    """客服工单摘要。"""

    # 问题类别：只能是四个标签之一
    category: Literal["退款", "物流", "咨询", "其他"] = Field(description="问题类别")
    # 紧急程度：低 / 中 / 高
    urgency: Literal["低", "中", "高"] = Field(description="紧急程度；用户表达着急、投诉升级时偏高")
    # 一句话摘要，便于列表展示
    summary: str = Field(description="一句话摘要，不超过 30 字")


# deepseek-v4 默认开启 thinking；thinking 下强制 tool_choice 可能 400
# with_structured_output 默认常走 function_calling，因此这里关闭 thinking
model = init_chat_model(
    "deepseek:deepseek-v4-flash",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)

# 把模型包装成「按 Ticket schema 返回」的可调用对象
extractor = model.with_structured_output(Ticket)

# 传入用户原话，得到 Ticket 实例（不是字符串）
ticket = extractor.invoke("我的快递三天没更新了，很着急。")

# 打印整个对象
print(ticket)
# 按字段访问
print(ticket.category, ticket.urgency)
# 需要字典时（例如写 JSON / 入库）
print(ticket.model_dump())

实际输出：

category='物流' urgency='高' summary='快递三天未更新物流信息，用户很着急'
物流 高
{'category': '物流', 'urgency': '高', 'summary': '快递三天未更新物流信息，用户很着急'}

要点：

with_structured_output(Ticket) 传入的是类，不是实例
返回值是 Ticket 实例，可 .字段 / .model_dump()
temperature=0 有利于分类、抽取类任务更稳
若打印出来仍是一段话而不是对象，多半没走到 structured 路径，或看错了返回值类型

4.2. 这一步背后发生了什么 #

结果直接是对象，中间过程被藏起来了。用 include_raw=True 掀开看一眼（§7.1 细讲），后面许多现象会好解释得多：

# 承接上面的代码
extractor = model.with_structured_output(Ticket, include_raw=True)
result = extractor.invoke("我的快递三天没更新了，很着急。")

raw = result["raw"]
print("raw 类型:  ", type(raw).__name__)
print("raw.content:", repr(raw.content))
print("tool_calls: ", raw.tool_calls)
print("parsed:    ", result["parsed"])

输出：

raw 类型:   AIMessage
raw.content: ''
tool_calls:  [{'name': 'Ticket', 'args': {'category': '物流', 'urgency': '高', 'summary': '快递三天未更新物流信息，用户很着急'}, 'id': 'call_00_feq4YxpLJklU8mGsr5ZT7920', 'type': 'tool_call'}]
parsed:     category='物流' urgency='高' summary='快递三天未更新物流信息，用户很着急'

记住三件事：

content 是空的。 模型没有「说话」，是在调工具。第 4 章的 AIMessage.content 在这条路径上拿不到内容。
数据在 tool_calls[0]["args"] 里，是个普通 dict。
parsed 是把该 dict 交给 `Ticket(args)` 得到的实例。** 校验就发生在这一步。

弄清这个流程后，后面两个现象就顺理成章：漏字段 / 枚举填错，会在第 3 步被 Pydantic 拦住（§7）；编造一个格式合法的值（§3.4），第 3 步拦不住。

5. 实战：订单抽取器 #

工单是「扁」的三字段；真实业务常是「嵌套」：一张订单里有多行商品。

本节目标：从一段口语描述里抽出 Order（items 为 list[OrderItem]），并验证两件事：

结构化输出支持嵌套 / 列表
可选字段（订单号、单价）在原文缺失时，能否老实填 null，而不是编造

设计 schema 时建议这样分：

必填：没有就无法继续处理的字段（如 items 至少可以是空列表）
可选：原文可能没有的字段（order_id、unit_price）
枚举兜底：状态判断不清时用「未知」，比硬猜「已完成」安全

5.1. 2.order_extractor.py #

# 从 typing 导入 Literal，用于枚举订单状态等字段
from typing import Literal

# 从 pydantic 导入 BaseModel 与 Field
from pydantic import BaseModel, Field

# 从 langchain.chat_models 导入 init_chat_model
from langchain.chat_models import init_chat_model

# 从 dotenv 导入 load_dotenv
from dotenv import load_dotenv

load_dotenv(override=True)


# 订单行项目：一个订单可包含多件商品
class OrderItem(BaseModel):
    """订单中的单个商品行。"""

    product_name: str = Field(description="商品名称")
    quantity: int = Field(description="购买数量", ge=1)
    unit_price: float | None = Field(
        default=None,
        description="单价（元）；原文未提及时填 null，不要编造",
    )


# 完整订单结构（可嵌套 list[OrderItem]）
class Order(BaseModel):
    """从用户描述中抽取的订单信息。"""

    order_id: str | None = Field(
        default=None,
        description="订单号；未提及时填 null",
    )
    customer_name: str | None = Field(
        default=None,
        description="客户姓名；未提及时填 null",
    )
    status: Literal["待付款", "待发货", "运输中", "已完成", "未知"] = Field(
        description="订单状态；无法判断时用「未知」"
    )
    items: list[OrderItem] = Field(description="商品明细列表；没有商品时为空列表")
    notes: str = Field(description="需要人工关注的补充说明；没有则写空字符串")


model = init_chat_model(
    "deepseek:deepseek-v4-flash",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)
extractor = model.with_structured_output(Order)

text = (
    "帮我记一下：张三的订单 A20260328001，"
    "买了 2 件机械键盘，单价 399；还买了 1 个2块钱的鼠标垫。"
    "已经付款，等着发货。"
)

order = extractor.invoke(text)
print(order)
print(order.model_dump_json(indent=2))

实际输出：

{
  "order_id": "A20260328001",
  "customer_name": "张三",
  "status": "待发货",
  "items": [
    {
      "product_name": "机械键盘",
      "quantity": 2,
      "unit_price": 399.0
    },
    {
      "product_name": "鼠标垫",
      "quantity": 1,
      "unit_price": 2.0
    }
  ],
  "notes": ""
}

两行商品都抽对了，「2 块钱的鼠标垫」这类口语也正确解析成了 unit_price: 2.0。

5.2. 嵌套 schema 传给模型时是什么样 #

嵌套类不会被摊平，Pydantic 用 $defs + $ref 引用：

# 承接上面的代码
import json

print(json.dumps(Order.model_json_schema(), ensure_ascii=False, indent=2))

关键部分（已讲过的字段从略）：

{
  "$defs": {
    "OrderItem": {
      "description": "订单中的单个商品行。",
      "properties": {
        "product_name": { "description": "商品名称", "type": "string" },
        "quantity": { "description": "购买数量", "minimum": 1, "type": "integer" },
        "unit_price": {
          "anyOf": [{ "type": "number" }, { "type": "null" }],
          "default": null,
          "description": "单价（元）；原文未提及时填 null，不要编造"
        }
      },
      "required": ["product_name", "quantity"],
      "type": "object"
    }
  },
  "properties": {
    "items": {
      "description": "商品明细列表；没有商品时为空列表",
      "items": { "$ref": "#/$defs/OrderItem" },
      "type": "array"
    }
  },
  "required": ["status", "items", "notes"]
}

三个细节值得看：

items 的 $ref 指向 $defs/OrderItem：JSON Schema 的标准引用，主流模型都能理解
unit_price 类型是 anyOf: [number, null]，且不在 required 里——正是 §3.3 的正确写法
ge=1 变成了 minimum: 1：嵌套层的约束同样会传下去

5.3. 缺失字段的坑：模型可能填字符串 "null" #

这是本章最值得单独拎出的坑：不报错、不抛异常，却会污染数据。

喂一段没有订单号的文本试试：

# 承接 §5.1 的代码
extractor = model.with_structured_output(Order, include_raw=True)
r = extractor.invoke("买了三个杯子，还没付钱")

args = r["raw"].tool_calls[0]["args"]
print("模型返回的参数:", args)
print("parsed.order_id:", repr(r["parsed"].order_id))
print("是 None 吗:", r["parsed"].order_id is None)

输出：

模型返回的参数: {'order_id': 'null', 'customer_name': 'null', 'status': '待付款', 'items': [{'product_name': '杯子', 'quantity': 3}], 'notes': ''}
parsed.order_id: 'null'
是 None 吗: False

模型写的是字符串 "null"，不是 JSON 的 null。字段类型是 str | None，字符串 "null" 完全合法，Pydantic 照收。于是下游所有 if order.order_id is None 都会失效，库里会存进一堆 "null" 字符串。

根因就在描述里那句「未提及时填 null」——模型很听话，把 null 四个字母填了进去。

行为跟输入有关。同一 schema 跑 4 条不同的缺失文本、各两轮：

第1轮 买了三个杯子，还没付钱    raw: order_id='null'      | parsed: 'null'   <-- 字符串污染
第1轮 要退货，没别的信息      raw: order_id=None        | parsed: None
第1轮 我买了两包纸巾，已经收到   raw: order_id=<键不存在>    | parsed: None
第1轮 下单了一个鼠标，等发货    raw: order_id=None        | parsed: None
第2轮 买了三个杯子，还没付钱    raw: order_id='null'      | parsed: 'null'   <-- 字符串污染
第2轮 要退货，没别的信息      raw: order_id=None        | parsed: None
第2轮 我买了两包纸巾，已经收到   raw: order_id=<键不存在>    | parsed: None
第2轮 下单了一个鼠标，等发货    raw: order_id='null'      | parsed: 'null'   <-- 字符串污染

共 16 个可选字段，其中 6 个被填成了字符串而不是 None

三种行为同时存在：填字符串 'null'、填真正的 None、干脆省略该键（省略是安全的，Pydantic 会用 default=None 补上）。temperature=0 下同一输入结果稳定，换输入却可能变——偶发脏数据比稳定出错更难发现。

5.4. 修法：不要指望描述，加一道归一化 #

改描述有帮助，但不可靠（换输入、换模型就可能复发）。稳妥做法是加 Pydantic 校验器，把各种「假空值」统一成 None：

from typing import Literal

from pydantic import BaseModel, Field, field_validator

# 模型常用来表示「没有」的各种写法，按需扩充
BLANKS = {"", "null", "none", "n/a", "na", "未提及", "无", "未知", "不详", "-"}


class SafeOrder(BaseModel):
    """带空值归一的订单。"""

    order_id: str | None = Field(default=None, description="订单号；原文未提及时留空")
    customer_name: str | None = Field(default=None, description="客户姓名；原文未提及时留空")
    status: Literal["待付款", "待发货", "运输中", "已完成", "未知"] = Field(description="订单状态")

    # mode="before" 表示在类型校验之前先跑，这样才能拦住字符串 "null"
    @field_validator("order_id", "customer_name", mode="before")
    @classmethod
    def blank_to_none(cls, v):
        """模型可能把「没有」写成 "null" / "无" / ""，统一归一成 None。"""
        if isinstance(v, str) and v.strip().lower() in BLANKS:
            return None
        return v

直接喂各种假空值验证归一逻辑（不依赖模型是否碰巧出错）：

# 承接上面的代码
for fake in ["null", "NULL", "", "无", "未提及", "A20260328001"]:
    o = SafeOrder(order_id=fake, status="未知")
    print(f"输入 {fake!r:16s} -> order_id={o.order_id!r}")

输出：

输入 'null'           -> order_id=None
输入 'NULL'           -> order_id=None
输入 ''               -> order_id=None
输入 '无'              -> order_id=None
输入 '未提及'            -> order_id=None
输入 'A20260328001'   -> order_id='A20260328001'

mode="before" 是关键：默认校验器在类型转换之后跑，那时 "null" 已是合法 str；before 让它在类型校验前介入，才能改掉。

约定：所有 str | None 抽取字段，都值得配这样的归一化校验器。 比在 prompt 里反复叮嘱「请填 null」可靠——校验器是代码，提示词只是请求。

6. 和 Prompt 模板组合 #

只靠 schema，模型有时仍会「脑补」。更稳的做法是两层分工协作：

层	负责	典型内容
Prompt（怎么问）	角色、边界、缺失时怎么办	「不要编造订单号」「信息不足 urgency 用低」
Schema（按什么结构答）	字段、类型、枚举、校验	`Literal` / `Field` / `ge`/`le`

第 5 章的 ChatPromptTemplate 管前者；本章的 with_structured_output 管后者。
两者用管道 | 拼起来（第 7 章系统讲 LCEL，这里先建立用法）：

{"complaint": "..."}
        │
        ▼
 ChatPromptTemplate  →  messages
        │
        ▼
 with_structured_output(Ticket)  →  Ticket

6.1. 3.prompt_structured_chain.py #

from typing import Literal

from pydantic import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv

load_dotenv(override=True)


class Ticket(BaseModel):
    """客服工单摘要。"""

    category: Literal["退款", "物流", "咨询", "其他"] = Field(description="问题类别")
    urgency: Literal["低", "中", "高"] = Field(description="紧急程度")
    summary: str = Field(description="一句话摘要")


# 系统提示约束抽取行为：缺失信息不要编造
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "你是客服质检员。根据用户原话抽取工单字段。"
            "只能依据原文判断；信息不足时 urgency 用「低」，summary 如实说明缺什么。"
            "不要编造订单号或物流单号。",
        ),
        ("human", "{complaint}"),
    ]
)

model = init_chat_model(
    "deepseek:deepseek-v4-flash",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)

# 填模板 → 结构化输出
chain = prompt | model.with_structured_output(Ticket)

ticket = chain.invoke(
    {"complaint": "我想问问会员积分怎么查，不着急。"}
)
print(ticket)

输出：

category='咨询' urgency='低' summary='用户咨询会员积分查询方法，不着急。'

这条输入没有任何着急信号，urgency 正确落到「低」。这里 schema 与 prompt 在同一件事上互相加强：Literal["低","中","高"] 限定取值，系统提示补充「信息不足时用低」。少任何一层，边界样本都更容易翻车。

调试建议：出问题先拆开看中间结果——先确认模板填对，再看结构化结果：

messages = prompt.format_messages(complaint="……")
print(messages)
print(model.with_structured_output(Ticket).invoke(messages))

7. 解析失败怎么处理 #

结构化输出不是百分之百成功。模型可能漏字段、枚举写错、数值越界；供应商也可能直接报 400。

先分清两层失败，处理方式截然不同：

失败层	含义	常见表现	优先处理
协议 / 调用失败	请求本身不被接受	HTTP 400、thinking 冲突	关 thinking、换策略、查集成配置
解析 / 校验失败	有返回但不符合 schema	`ValidationError` / `parsing_error`	加强 description、加约束、重试或人工复核

本节三个手段，由易到难：

include_raw=True：失败时别直接炸，先拿到错误信息
schema 约束：尽量让非法值在校验阶段暴露
简单重试：把错误反馈给模型，再抽一次

7.1. include_raw=True：同时拿到原文与解析结果 #

默认 include_raw=False：解析失败直接抛异常，适合「快速失败」脚本。
生产接口更常见：记下 raw、看 parsing_error，再决定重试或转人工。

设为 True 时，返回字典三件套如下：

键	含义
`raw`	原始 `AIMessage`（便于日志 / token 用量）
`parsed`	成功时为对象，失败时多为 `None`
`parsing_error`	失败原因；成功时为 `None`

7.1.1. 4.include_raw.py #

from typing import Literal

from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv

load_dotenv(override=True)


class Ticket(BaseModel):
    """客服工单摘要。"""

    category: Literal["退款", "物流", "咨询", "其他"] = Field(description="问题类别")
    urgency: Literal["低", "中", "高"] = Field(description="紧急程度")
    summary: str = Field(description="一句话摘要")


model = init_chat_model(
    "deepseek:deepseek-v4-flash",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)

# include_raw=True：解析错误不会直接炸进程，而是放进 parsing_error
extractor = model.with_structured_output(Ticket, include_raw=True)

result = extractor.invoke("我的快递三天没更新了，很着急。")

print("parsed:", result["parsed"])
print("parsing_error:", result["parsing_error"])
print("raw type:", type(result["raw"]).__name__)

# 成功时常见形态：
# {
#   "raw": AIMessage(...),
#   "parsed": Ticket(...),
#   "parsing_error": None,
# }

输出：

parsed: category='物流' urgency='高' summary='快递三天未更新物流信息，用户很着急'
parsing_error: None
raw type: AIMessage

业务里可写成：

if result["parsing_error"] is not None:
    # 记日志、人工复核、或触发重试
    print("解析失败：", result["parsing_error"])
else:
    ticket = result["parsed"]

容易忽略的一点：include_raw=True 会改变返回值类型（对象变成 dict）。适合两种用法——排查时临时打开，或生产接口一开始就用、把 raw 写进日志。不要在业务代码里开开关关，取值逻辑会跟着变形。

另外，raw 里带有 usage_metadata，是统计 token 成本最方便的入口：

print(result["raw"].usage_metadata)

{'input_tokens': 375, 'output_tokens': 76, 'total_tokens': 451,
 'input_token_details': {'cache_read': 256}, 'output_token_details': {}}

顺便能看出：这条简单抽取花了 375 个输入 token，多半花在工具定义 JSON 上——schema 字段越多、描述越长，每次调用的固定成本越高。字段别贪多，用不上的别加。

7.2. 用校验约束减少「脏数据」 #

与其事后正则清洗，不如在 schema 里写清边界。有个容易误解的点先说清：

ge/le 会翻译成 minimum/maximum 一并发给模型（§3.1 验证过），所以不只是「事后拦一道」，更是「事前告诉模型范围」。多数时候模型自己就收敛了，根本走不到 Pydantic 报错。

实测「打 10 分」「打 100 分」这类越界输入如下：

这东西绝了，必须打 10 分！物流也快。  -> rating=5 sentiment=正面
打 100 分，无可挑剔                -> rating=5 sentiment=正面
还行吧，3 分                      -> rating=3 sentiment=中性

三条都收敛到合法范围，没有触发 ValidationError。

那 Pydantic 这道校验还有用吗？有——模型不听话时的兜底。直接构造越界值，就能看到拦截效果和报错内容：

from pydantic import ValidationError

try:
    ProductRating(rating=10, sentiment="正面")
except ValidationError as e:
    print(e)

输出：

1 validation error for ProductRating
rating
  Input should be less than or equal to 5 [type=less_than_equal, input_value=10, input_type=int]
    For further information visit https://errors.pydantic.dev/2.13/v/less_than_equal

注意这条报错本身就很适合喂回给模型（Input should be less than or equal to 5，还带实际输入值），正是 §7.3 重试策略的基础。

所以两层是配合，不是二选一：

层	作用	什么时候起效
JSON Schema 里的 `minimum`/`maximum`/`enum`	引导模型产出合法值	每次调用，降低出错概率
Pydantic 校验	挡住漏网的非法值	模型不听话时，抛 `ValidationError`

7.2.1. 5.validation_constraints.py #

from typing import Literal

from pydantic import BaseModel, Field, ValidationError
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv

load_dotenv(override=True)


class ProductRating(BaseModel):
    """商品评分抽取结果。"""

    # ge / le：评分必须在 1～5
    rating: int = Field(description="评分，只能是 1 到 5 的整数", ge=1, le=5)
    sentiment: Literal["正面", "负面", "中性"] = Field(description="情感倾向")
    comment: str = Field(description="简短评论原文摘要")


model = init_chat_model(
    "deepseek:deepseek-v4-flash",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)
extractor = model.with_structured_output(ProductRating)

# 原文写「10 分」时，合格模型应收敛到 5；若仍越界，Pydantic 会校验失败
try:
    rating = extractor.invoke("这东西绝了，必须打 10 分！物流也快。")
    print(rating)
except ValidationError as e:
    print("校验失败：", e)
except Exception as e:
    # 也可能是供应商侧错误或其它解析异常
    print("调用/解析失败：", type(e).__name__, e)

7.3. 简单重试：失败后再问一次 #

生产中常见策略：失败 → 把错误反馈给模型 → 再抽一次。
这和 Agent 的 ToolStrategy(handle_errors=True) 思路类似，这里先用手写循环看清机制。

注意：

重试要有上限，避免死循环烧额度
把具体校验错误喂回去，比只说「再试一次」有效（ValidationError 文本很好用，见 §7.2）
连续失败应转人工或降级，不要无限重试

7.3.1. 先看一个写错的版本 #

重试的直觉写法是「把上一轮回复和错误说明都追加到对话里」：

# 这段是错的，下面解释为什么
err = result["parsing_error"]
messages.append(result["raw"])                     # 上一轮的 AIMessage
messages.append(HumanMessage(f"上次输出无法通过校验：{err}。请重新抽取。"))

这段代码在第一次就成功时看不出问题（break 掉了，根本没执行到）。一旦真进入重试——也就是它唯一有用的时候——就会炸：

BadRequestError: Error code: 400 - {'error': {'message': "An assistant message with
'tool_calls' must be followed by tool messages responding to each 'tool_call_id'.
(insufficient tool messages following tool_calls message)", ...}}

原因回到 §4.2：结构化输出走 tool calling，result["raw"] 是一条带 tool_calls 的 AIMessage。OpenAI 兼容协议有硬规定：带 tool_calls 的助手消息后面，必须紧跟对每个 tool_call_id 的 ToolMessage。中间插一条 HumanMessage 就违反契约。

打印消息序列就能直接看到问题：

print([type(m).__name__ for m in messages])
# ['HumanMessage', 'AIMessage', 'HumanMessage']
#                  ^^^^^^^^^^ 带 tool_calls，后面却不是 ToolMessage

这条协议约束不只影响结构化输出。第 24 章手写 Agent 循环时，「模型请求了工具但循环被提前切断」也会撞上同一个 400，处理一样：要么补齐 ToolMessage，要么别把这条 AIMessage 带进下一轮。

7.3.2. 两种正确写法 #

方案 A：补齐 ToolMessage。 保留完整对话轨迹，给每个 tool_call 补一条应答消息：

from langchain_core.messages import ToolMessage

raw = result["raw"]
messages.append(raw)
# 给每个 tool_call 补一条应答，满足协议要求
for tc in raw.tool_calls:
    messages.append(ToolMessage(content="上次输出未通过校验", tool_call_id=tc["id"]))
messages.append(HumanMessage("请严格按 schema 重新抽取。"))

消息序列变成 ['HumanMessage', 'AIMessage', 'ToolMessage', 'HumanMessage']，协议满足，可以调用成功。

方案 B：不带上一轮的 AIMessage。 只把错误信息写进新的人类消息即可：

messages.append(
    HumanMessage(f"上次输出无法通过校验：{err}。请严格按 schema 重新抽取。")
)

消息序列是 ['HumanMessage', 'HumanMessage']，同样能跑通。

两种都实测有效，按需求选：

	方案 A（补 `ToolMessage`）	方案 B（丢弃上一轮）
模型能看到自己上次错在哪	能（有原始 `tool_calls`）	只能看到你的文字描述
消息数与 token 消耗	更多	更少
代码复杂度	要遍历 `tool_calls`	一行
适合	需要模型对照修正的复杂 schema	大多数抽取场景

下面完整示例用方案 B：抽取任务的 schema 通常不复杂，把校验错误讲清楚就够。

7.3.3. 6.retry_on_parse_error.py #

from typing import Literal

from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from dotenv import load_dotenv

load_dotenv(override=True)


class Ticket(BaseModel):
    """客服工单摘要。"""

    category: Literal["退款", "物流", "咨询", "其他"] = Field(description="问题类别")
    urgency: Literal["低", "中", "高"] = Field(description="紧急程度")
    summary: str = Field(description="一句话摘要，不超过 30 字")


model = init_chat_model(
    "deepseek:deepseek-v4-flash",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)
# include_raw=True 才能拿到 parsing_error 而不是直接抛异常
extractor = model.with_structured_output(Ticket, include_raw=True)

user_text = "我的快递三天没更新了，很着急。"

# 只保留最初的用户消息；重试时追加错误说明，不追加上一轮的 AIMessage
messages = [
    HumanMessage(
        "请抽取工单字段。category 只能是：退款/物流/咨询/其他；"
        f"urgency 只能是：低/中/高。\n用户原话：{user_text}"
    )
]

MAX_ATTEMPTS = 3
ticket = None

for attempt in range(1, MAX_ATTEMPTS + 1):
    result = extractor.invoke(messages)

    # 成功就收工
    if result["parsing_error"] is None and result["parsed"] is not None:
        ticket = result["parsed"]
        print(f"第 {attempt} 次成功")
        break

    # 失败：把具体校验错误写进新的人类消息，让模型看到自己错在哪
    err = result["parsing_error"]
    print(f"第 {attempt} 次失败：{err}")
    messages.append(
        HumanMessage(f"上次输出无法通过校验：{err}。请严格按 schema 重新抽取。")
    )

# 重试用尽仍失败要有明确出口，别让 None 悄悄流进下游
if ticket is None:
    print(f"重试 {MAX_ATTEMPTS} 次仍失败，转人工处理")
else:
    print("最终结果：", ticket)

输出（这条输入很规整，第一次就通过）：

第 1 次成功
最终结果： category='物流' urgency='高' summary='快递三天未更新，用户很着急'

想看到重试真正发生，可以故意把 schema 收紧到模型难满足，例如 summary: str = Field(description="摘要", max_length=5)，再观察循环行为。

最后一段的 if ticket is None 不要省。 重试耗尽后 ticket 仍是 None，若直接 return ticket 交给下游，就成了「静默的 None」——报错会出现在离现场很远的地方。要么转人工，要么抛异常，总之要有明确出口。

8. TypedDict：只要字典、不要 Pydantic 实例时 #

不是所有场景都需要 Pydantic 实例。若只想快速拿到 dict，且暂时不依赖运行时校验，可用 TypedDict。

但要清楚代价是什么：

TypedDict / 纯 JSON Schema 路径，通常不会像 Pydantic 那样做完整运行时校验。
类型提示更多服务「读代码的人 / 静态检查」，不等于自动拦脏数据。

8.1. 7.typed_dict_output.py #

from typing import Literal
from typing_extensions import Annotated, TypedDict

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv

load_dotenv(override=True)


class TicketDict(TypedDict):
    """客服工单摘要（字典版）。"""

    category: Annotated[Literal["退款", "物流", "咨询", "其他"], ..., "问题类别"]
    urgency: Annotated[Literal["低", "中", "高"], ..., "紧急程度"]
    summary: Annotated[str, ..., "一句话摘要"]


model = init_chat_model(
    "deepseek:deepseek-v4-flash",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)

extractor = model.with_structured_output(TicketDict)
result = extractor.invoke("想退货，订单还没发货。")
print(type(result), result)  # <class 'dict'> {...}

输出：

<class 'dict'> {'category': '退款', 'urgency': '中', 'summary': '用户想退货，但订单尚未发货'}

8.2. Annotated[类型, ..., "描述"] 这个写法怎么读 #

TypedDict 没有 Field，描述只能靠 Annotated 挂上去。三个位置含义如下：

位置	内容	说明
第 1 个	`Literal["退款", ...]`	字段类型，和 Pydantic 一样
第 2 个	`...`（`Ellipsis`）	默认值占位；`...` 表示必填
第 3 个	`"问题类别"`	字段描述，作用等同 `Field(description=...)`

它确实生效——把工具定义打出来，和 Pydantic 版几乎一样：

{
  "name": "TicketDict",
  "description": "客服工单摘要（字典版）。",
  "parameters": {
    "properties": {
      "category": {
        "description": "问题类别",
        "enum": ["退款", "物流", "咨询", "其他"],
        "type": "string"
      }
    },
    "required": ["category", "urgency", "summary"]
  }
}

所以 TypedDict 的代价不在「模型端」，而在「解析端」：发给模型的约束一样，拿回来后没有 Pydantic 那层校验与转换——没有 ge/le、没有 field_validator（§5.4 那套归一化用不了）、也没有默认值填充。

怎么选可以看这张表：

需求	建议
要校验、嵌套、默认值、后续强类型	Pydantic `BaseModel`（本章默认）
只要 dict、结构简单	`TypedDict`
与外部系统交换 schema	JSON Schema（部分供应商需指定 `method`）

9. 结构化输出可以流式吗 #

可以，但流出来的东西和第 3 章的文本流式完全不是一回事，容易误用，值得单独说清。

文本流式是增量的：每个 chunk 是一小段新文本，要自己拼。结构化输出的流式是累积的：每个 chunk 都是完整对象，字段逐渐被填满。

extractor = model.with_structured_output(Ticket)

for i, chunk in enumerate(extractor.stream("快递三天没更新，很着急。")):
    print(f"chunk {i}: {chunk}")

输出：

chunk 0: category='物流' urgency='高' summary=''
chunk 1: category='物流' urgency='高' summary='快递'
chunk 2: category='物流' urgency='高' summary='快递三天'
chunk 3: category='物流' urgency='高' summary='快递三天未'
chunk 4: category='物流' urgency='高' summary='快递三天未更新'
chunk 5: category='物流' urgency='高' summary='快递三天未更新物流'
chunk 6: category='物流' urgency='高' summary='快递三天未更新物流信息'
chunk 7: category='物流' urgency='高' summary='快递三天未更新物流信息，'
chunk 8: category='物流' urgency='高' summary='快递三天未更新物流信息，客户'
chunk 9: category='物流' urgency='高' summary='快递三天未更新物流信息，客户着急'

（chunk 条数取决于生成文本长度，每次运行可能略有不同，重点看形态。）

关键差别，用错就会出 bug：

每个 chunk 都是完整对象，直接用最后一个即可，千万不要把 chunk 拼接起来。
文本流式要 full += chunk，结构化流式要 last = chunk。

注意 chunk 0 已有 category 和 urgency——Pydantic 必须凑齐所有必填字段才能构造合法实例，所以是攒够了才开始吐。

TypedDict 版没有这个限制，从空字典开始：

chunk 0: {}
chunk 1: {'category': ''}
chunk 2: {'category': '物流'}
chunk 3: {'category': '物流', 'urgency': ''}
chunk 4: {'category': '物流', 'urgency': '高'}
chunk 5: {'category': '物流', 'urgency': '高', 'summary': ''}
...
chunk 16: {'category': '物流', 'urgency': '高', 'summary': '快递三天未更新物流信息，客户非常着急。'}

什么时候值得用？ 抽取任务通常是「等结果入库」，用户不看过程，invoke 就够。流式更适合长文本摘要类字段——例如给人看的报告，summary 有几百字，边生成边渲染体验更好。字段少而短的分类抽取，流式往往只增加复杂度。

10. method 参数：换一种约束模型的方式 #

with_structured_output 默认走 tool calling（§3.1 见过的工具定义）。但这不是唯一路径，method 可以换：

`method`	原理	有没有把 schema 发给模型	DeepSeek 实测
`"function_calling"`（默认）	schema 变成工具定义，强制模型调用	有	可用
`"json_schema"`	走厂商原生的结构化输出接口	有	可用
`"json_mode"`	只要求「回一段合法 JSON」	没有	不可用（见下）

第三列是重点：下面会看到它直接决定能不能用。

# 三种 method 依次试一遍，失败的把异常类型和前 90 个字符打出来
for m in ["function_calling", "json_schema", "json_mode"]:
    try:
        extractor = model.with_structured_output(Ticket, method=m)
        print(f"{m:18s} OK -> {extractor.invoke('想退货，订单还没发货。')}")
    except Exception as e:
        print(f"{m:18s} {type(e).__name__}: {str(e)[:90]}")

输出：

function_calling   OK -> category='退款' urgency='中' summary='客户想退货，订单尚未发货。'
json_schema        OK -> category='退款' urgency='中' summary='客户想退货，但订单尚未发货。'
json_mode          BadRequestError: Error code: 400 - {'error': {'message': "Prompt must contain the word 'json' in some form

json_mode 那条报错很好懂：该模式对应 OpenAI 兼容接口的 response_format={"type": "json_object"}，而接口要求提示词里必须出现 json 字样，否则直接 400。

于是很自然会想：按它说的做，在提示词里加上「请用 json 返回」不就行了？关键在这一步。 加上后不再报 400，却换了个地方失败：

# 提示词里补上 json 字样，绕过 400；连问两次看是否稳定
for n in range(2):
    try:
        got = model.with_structured_output(Ticket, method="json_mode").invoke(
            "想退货，订单还没发货。请用 json 返回。"
        )
        print(f"第 {n + 1} 次 OK -> {got}")
    except Exception as e:
        print(f"第 {n + 1} 次 {type(e).__name__}: {str(e)[:120]}")

输出：

第 1 次 OutputParserException: Failed to parse Ticket from completion {"action": "return_request", "order_status": "unshipped"}. Got: 3 validation erro
第 2 次 OutputParserException: Failed to parse Ticket from completion {"action": "return_request", "order_status": "unshipped", "message": "您的订单尚未发货，可以

模型确实回了合法 JSON，但字段名是它自己编的——action、order_status，跟 Ticket 要的 category / urgency / summary 毫无关系。两次运行编出同一套键名，说明不是随机抽风。

根因看一眼绑定参数就清楚：

# 把这条链的第一步拿出来，看它到底往接口上绑了什么
runnable = model.with_structured_output(Ticket, method="json_mode")
for step in runnable.steps:
    print(f"{type(step).__name__}  kwargs={getattr(step, 'kwargs', None)}")

输出：

_ChatModelBinding  kwargs={'response_format': {'type': 'json_object'}, 'ls_structured_output_format': {'kwargs': {'method': 'json_mode'}, 'schema': <class '__main__.Ticket'>}}
PydanticOutputParser  kwargs=None

请求里只有 response_format={'type': 'json_object'}，没有工具定义，也没有 schema（ls_structured_output_format 是给 LangSmith 追踪的元数据，不发给模型）。也就是说：约束只到「必须是 JSON」这一层，字段叫什么、有哪些，模型完全不知道。 猜错是必然的；解析器最后发现对不上，才抛 OutputParserException。

三点结论如下：

前两种在 DeepSeek 上抽到的字段一致（category、urgency 都对，只有 summary 措辞会有些随机差异），没必要折腾——保持默认就好。
json_mode 不适合结构化抽取，因为它不传 schema。定位是「提示词里已写清格式，接口只保证别掺杂闲话」，也就是得自己在提示词里描述字段，再配合 PydanticOutputParser.get_format_instructions() 才能用。既然 function_calling 能把 schema 直接交给模型，就没理由退回这条路。
不同供应商支持的 method 不一样。 遇到「默认方式报错」时，换 method 值得一试，具体以各家 Providers 文档为准。

11. Agent 里的结构化输出 #

前面都是模型层：model.with_structured_output(...)。
若已在用 create_agent，也可在 Agent 层声明最终结构：response_format=...。

两者对比：

	模型层 `with_structured_output`	Agent 层 `response_format`
入口	直接 `extractor.invoke(...)`	`agent.invoke({"messages": ...})`
结果位置	调用返回值本身	`result["structured_response"]`
还要不要工具循环	通常没有	可以同时带 tools（视策略与模型能力）

response_format 常见两种策略如下：

策略	含义	何时用
ProviderStrategy	走厂商原生 structured output / `response_format`	OpenAI 等明确支持原生结构化输出的模型
ToolStrategy	把 schema 当成特殊工具，靠 tool calling 拿到结构	DeepSeek 等不支持原生 `response_format` 的模型

对本文默认的 DeepSeek：不要只写 response_format=Ticket（可能自动走 ProviderStrategy 并报 400），应显式用 ToolStrategy(Ticket)。

11.1. 8.agent_response_format.py #

# 从 typing 导入 Literal，用于把字段取值限制为若干固定标签
from typing import Literal

# 从 pydantic 导入 BaseModel 与 Field，用于定义带校验的数据结构
from pydantic import BaseModel, Field

# 从 langchain.agents 导入 create_agent，用于创建 Agent
from langchain.agents import create_agent

# 从 langchain.agents.structured_output 导入 ToolStrategy，用工具调用方式拿到结构化结果
from langchain.agents.structured_output import ToolStrategy

# 从 langchain.chat_models 导入 init_chat_model，用于初始化聊天模型
from langchain.chat_models import init_chat_model

# 从 dotenv 导入 load_dotenv，用于加载环境变量
from dotenv import load_dotenv

# 加载 .env；override=True 表示覆盖已存在的同名变量
load_dotenv(override=True)


# 定义客服工单摘要的数据结构
class Ticket(BaseModel):
    # 类文档字符串：说明该结构表示客服工单摘要
    """客服工单摘要。"""

    # 问题类别：只能是退款 / 物流 / 咨询 / 其他四个标签之一
    category: Literal["退款", "物流", "咨询", "其他"] = Field(description="问题类别")

    # 紧急程度：只能是低 / 中 / 高
    urgency: Literal["低", "中", "高"] = Field(description="紧急程度")

    # 一句话摘要，便于列表展示或入库
    summary: str = Field(description="一句话摘要")


# 初始化聊天模型；deepseek-v4 默认开启 thinking，结构化输出场景下先关闭
model = init_chat_model(
    # 指定使用的模型名称
    "deepseek:deepseek-v4-flash",
    # 温度设为 0，尽量让输出更稳定、可复现
    temperature=0,
    # 通过 extra_body 关闭 thinking，避免强制 tool_choice 时出现 400
    extra_body={"thinking": {"type": "disabled"}},
)

# 创建 Agent：模型层之外，在 Agent 层用 response_format 约束最终结构化结果
agent = create_agent(
    # 传入上面初始化好的聊天模型
    model=model,
    # 本示例不做工具调用业务，工具列表置空
    tools=[],
    # DeepSeek 不支持原生 response_format，显式用 ToolStrategy；handle_errors=True 表示解析失败时把错误反馈给模型重试
    response_format=ToolStrategy(Ticket, handle_errors=True),
    # 系统提示：要求只依据用户原话抽取，不要编造字段
    system_prompt="根据用户原话抽取工单，不要编造。",
)

# 调用 Agent，传入用户消息，触发工单抽取
result = agent.invoke(
    # messages 里放一条用户原话：付款两天未发货且着急
    {"messages": [{"role": "user", "content": "付款了两天还没发货，挺着急。"}]}
)

# 打印 Agent 返回的结构化结果（Ticket 实例）
print(result["structured_response"])

# 需要看对话轨迹时仍可用 result["messages"]

本章重点仍在模型层抽取；Agent 策略、与工具并用等，放到第 9～10 章再展开。
完整说明见官方 Structured output。

12. DeepSeek 使用注意 #

本文默认 DeepSeek，结构化输出有几条「几乎必踩」的坑如下：

现象	原因	处理
`with_structured_output` 报 400	deepseek-v4 默认 thinking，与强制 tool_choice 冲突	`extra_body={"thinking": {"type": "disabled"}}`
Agent `response_format=Ticket` 报 `response_format type is unavailable`	自动走了 ProviderStrategy，DeepSeek 不支持原生 structured output	改为 `response_format=ToolStrategy(Ticket)`
字段乱跳、枚举不稳	temperature 偏高或描述含糊	`temperature=0`；补强 `Field(description=...)`
可选字段被填成字符串 `"null"`	描述里写了「填 null」，模型照字面填	加 `field_validator` 归一化（§5.4）
必填字段凭空出现假值	该可选的字段声明成了必填	加 `default=None`（§3.3、§3.4）
重试时报 `insufficient tool messages`	带 `tool_calls` 的 `AIMessage` 后面接了 `HumanMessage`	补 `ToolMessage` 或别带上一轮（§7.3）
`method="json_mode"` 报 400	该模式要求提示词里出现 `json` 字样	用默认的 `function_calling`（§10）
补上 `json` 字样后改报 `OutputParserException`，字段名全不对	`json_mode` 不把 schema 发给模型，只要求回合法 JSON	同上；这个模式不适合结构化抽取（§10）

以各家 Providers 文档为准；本文 DeepSeek 路径默认不改 method。

13. 实用约定与坑 #

把前面经验收成清单，写业务代码时对照用：

约定	说明
传类不传实例	`with_structured_output(Ticket)` ✔ / `Ticket()` ✘
description 写给人看也写给模型看	含糊描述 = 含糊字段；模型看到的就是这句话（§3.1）
可选字段必须给 `default=None`	只写或 None`会进`required`，模型被迫填空字符串（§3.3）
`str 或 None` 字段配归一化校验器	拦住字符串 `"null"` / `"无"` / `""`（§5.4）
抽取任务 temperature 偏低	常用 `0`～`0.2`
先 prompt 再结构化	约束「别编造」放在系统提示里更稳
调试先 `include_raw`	同时看 `parsed`、`parsing_error` 和 `raw.tool_calls`
流式结果取最后一个，不要拼接	结构化流式是累积式完整对象（§9）
重试要有明确出口	耗尽次数后转人工或抛异常，别让 `None` 流进下游（§7.3.3）
别和「自由聊天回复」混用同一出口	给 UI 展示的话术 vs 给库表的字段，分开两条链

有一条值得单独强调，贯穿本章所有坑：

格式正确 ≠ 内容正确。 Pydantic 能保证拿到结构合法的 Order，不能保证 order_id 不是编的、不是字符串 "null"。schema 管结构，归一化校验器与可选性设计管内容。

口诀：

Schema 定结构，Prompt 定边界，校验器兜脏数据，失败要可观测。

14. 练习 #

看清工具定义：给 Ticket 随便加两个字段，用 §3.1 的方法把工具定义打出来，确认 description 与 enum 都到位。
可选字段矩阵：把 order_id 分别写成 str | None = Field(default=None) 与 str | None = Field(...)，各跑一段没有订单号的文本，对比 model_json_schema()["required"] 与实际返回值。
复现字符串污染：用 §5.1 的完整 Order schema 跑 5～8 条缺失字段文本，统计有多少次填成字符串 "null"；再加上 §5.4 的校验器，确认全部归一成 None。
触发重试：把 summary 改成 Field(description="摘要", max_length=5)，观察 §7.3.3 循环真正跑起来时的输出。
踩一次 400：故意用 §7.3.1 的错误写法，亲眼看一遍 insufficient tool messages，再换成方案 A 或 B 修好。
流式对比：同一 schema 分别用 Pydantic 与 TypedDict 跑 stream()，观察第一个 chunk 的差别，并解释原因。
对照 Agent：同一段投诉，分别用 with_structured_output 与 create_agent(response_format=ToolStrategy(...))，对比结果字段是否一致。

15. 本章小结 #

结构化输出让模型返回可校验对象，适合抽取 / 分类 / 填表 / 入库。
优先用 Pydantic BaseModel + Field(description=...) 定义 schema。
入口 API：model.with_structured_output(Schema)；可与 ChatPromptTemplate 组成 prompt | extractor。
底层是 tool calling：schema 变成工具定义，数据在 raw.tool_calls[0]["args"] 里，content 是空的。搞不清状况时把工具定义打出来看（§3.1）。
可选字段一定给 default=None。只写 | None 会进 required，模型被迫填空字符串；必填且信息不足时，模型会编一个格式完美的假值。
str | None 字段配一个 mode="before" 的归一化校验器，把字符串 "null" / "无" / "" 统一成 None。实测 16 个可选字段里，有 6 个被填成了字符串。
失败处理：include_raw、校验约束、重试；Agent 侧可用 response_format / ToolStrategy(handle_errors=...)。重试时别把带 tool_calls 的 AIMessage 直接接 HumanMessage，会报 400。
结构化输出可以流式，但每个 chunk 都是完整对象；取最后一个，不要拼接。
method 保持默认 function_calling 即可；本文用 DeepSeek 时记得 关闭 thinking，并保持偏低的 temperature。

下一章：Runnable 与 LCEL——把 prompt | model 升级成管道：并行、分支、流式与可组合的处理链。

导航菜单