> ## Documentation Index
> Fetch the complete documentation index at: https://www.yuan111.asia/doc/llms.txt
> Use this file to discover all available pages before exploring further.

# 05 · 打字机与结构化：流式输出 + 让模型交出 JSON

> stream() 打字机输出 + with_structured_output 让模型交出类型正确的 JSON。

# 05 · 打字机与结构化：流式输出 + 让模型交出 JSON

## 开头的现象

小林的多轮对话跑通了，但他被一个问题折磨得够呛：**每次问模型，都要干等两三秒，然后"啪"地一下，一整段回答突然蹦出来。** 他用 `time.time()` 一测：invoke 花了 3.2 秒，然后一次性打印了全部 200 字。

他盯着那个"唰"一下全出来的画面，心里想：**"网页上的 AI 都是打字机效果，一个字一个字蹦，我的怎么像倒垃圾？用户体验也太差了。"**

## 第一幕：小林发现了 stream() —— 同一个插口的另一种插法

小林去翻 LangChain 文档，看到 `BaseChatModel` 上除了 `invoke()`，还挂着一个叫 `stream()` 的方法。文档里写着：**"stream 把一次性返回变成增量返回——每生成一点，就吐给你一点。"**

他改了一行代码：

```python theme={null}
import time

# 版本 A：invoke 一次性（小林原来的写法）
t0 = time.time()
resp = model.invoke("从 1 数到 5，每行一个数字")
print(f"耗时 {time.time()-t0:.1f}s，一次性输出：{resp.content}")

# 版本 B：stream 流式（小林的新写法）
t0 = time.time()
collected = ""
for chunk in model.stream("从 1 数到 5，每行一个数字"):
    collected \+= chunk.content or ""
    print(chunk.content, end="", flush=True)   # flush=True：立即打印，不打缓冲
print(f"\n耗时 {time.time()-t0:.1f}s，流式输出完毕")
```

运行结果让他一愣：

```text theme={null}
版本 A：耗时 3.2s，一次性输出：1 2 3 4 5
版本 B：1 2 3 4 5（逐个蹦出来），耗时 0.8s 就看到了第一个数字
```

**总耗时其实没变短**（模型生成 5 个数字还是那么久），但**体验天差地别**：版本 B 第一秒就有内容可看，版本 A 得干等三秒。小林恍然大悟：**"invoke 和 stream 是同一个插口的两种插法——一个要完整回答，一个要边生成边吐。"**

他打印了 stream 的每个 chunk，发现每个 chunk 都是一个小号的"消息块"（AIMessageChunk），第一块往往是个"空壳"（只有 id，content 是空的），最后一块带着 `usage_metadata` 和 `stop_reason: "end_turn"`。他心想：**"原来流式是一串碎片，最后一块负责'收尾'说'我生成完了'。"**

> 想验证打字机效果吗？跑 `05_streaming.py`。你会看到：版本 A 憋三秒一次性蹦出，版本 B 一秒内开始逐个蹦字——同一个 `model`，只换了一个方法名。

## 第二幕：小林想"让模型直接交 JSON"——第一次失败了

小林的老板有个新需求：客服机器人要把用户的投诉自动整理成"工单"——意图是什么、订单号是多少、紧急程度几级。小林想：让模型直接输出 JSON 不就行了？

他先试了最笨的办法——在 prompt 里写"请输出 JSON"：

```python theme={null}
resp = model.invoke("把这句话整理成JSON：我上周买的东西没发货！订单号20260701001，明天就要用了！")
print(resp.content)
# → "好的，我为您整理如下：\n{\n  \"intent\": \"投诉\",\n  \"order_id\": ...\n}\n希望这能帮到您！"
```

小林看着输出里的"好的，我为您整理如下"和"希望这能帮到您！"，血压上来了：**"模型给我夹带私货！我要的是纯 JSON，它给我包了一层客套话！"** 他被迫写正则去剥那些客套话，剥完发现 JSON 偶尔还缺字段、字段类型还不对（order\_id 应该是字符串，模型有时给数字）。

他心想：**"靠 prompt 劝模型输出 JSON，就像靠嘴劝一个话痨少说两句——偶尔行，但永远不可靠。"**

## 第三幕：with\_structured\_output() —— 让模型"变成"只输出蓝图的模型

小林翻到了 `BaseChatModel` 上的另一个方法：`with_structured_output()`。文档说：**"你给它一个 Pydantic 蓝图类，它返回一个'只会按蓝图输出'的模型。"**

他写了一个 Pydantic 类（蓝图），然后把模型"包装"了一下：

```python theme={null}
from pydantic import BaseModel, Field

class SupportTicket(BaseModel):
    intent: str = Field(description="用户意图：物流查询/退换货/投诉/其他")
    order_id: str | None = Field(description="订单号，用户没提则为空", default=None)
    urgency: int = Field(description="紧急程度 1~5，5 最紧急")

# 关键：把模型"变成"只输出 SupportTicket 的模型
structured_model = model.with_structured_output(SupportTicket)

ticket = structured_model.invoke(
    "我上周买的东西到现在没发货！订单号20260701001，你们到底在搞什么？？？明天就要用了！"
)
print(ticket)
# → intent='投诉' order_id='20260701001' urgency=5   ← 一个真正的 Pydantic 对象！
```

小林盯着输出，眼睛亮了：**没有客套话、没有多余字段、类型完全正确**——`urgency` 是 `int`（可以直接 `ticket.urgency \+ 1` 参与运算），`order_id` 是字符串。他直接写业务逻辑：

```python theme={null}
if ticket.intent == "投诉":
    print(f"→ 转人工处理（意图：{ticket.intent}，紧急度：{ticket.urgency}）")
print(f"→ 关联订单号：{ticket.order_id}")
```

他彻底明白了 `with_structured_output()` 干了什么：**它内部让模型输出 JSON，然后用 Pydantic 帮你校验、转类型、补默认值——模型给的字段对不上，它帮你纠正或报错。你拿到手的不是"碰运气的字符串"，是"类型正确的对象"。**

他还发现：**`with_structured_output()` 返回的对象不是 AIMessage，而是你定义的 Pydantic 类的实例**——`movie.year` 是 `int` 可以直接算。这跟"先拿 AIMessage 再手动解析"是两条完全不同的路。

> 想验证"蓝图"魔法吗？跑 `05_structured.py`。第二个例子（SupportTicket）输入一段乱糟糟的投诉，输出一个规整的工单对象——字段、类型、默认值全对。

### 🔧 技术细节：这一章涉及的关键类和签名

**① 流式——`model.stream()` 返回什么**

```python theme={null}
model.stream(input)   # → 迭代器，每个元素是 AIMessageChunk
for chunk in model.stream("你好"):
    print(chunk.content)     # 一片文字
    print(chunk.response_metadata)  # 首块是空壳，末块带 stop_reason/usage
# 流式完成后的"收尾"信号：chunk.response_metadata["stop_reason"] == "end_turn"
```

**② `StrOutputParser`——剥壳取文本（langchain\_core.output\_parsers）**

```python theme={null}
StrOutputParser().parse(text: str) -> str    # 纯透传，等于 lambda x: x
# 但 invoke 输入可以是 AIMessage，自动取 .content
parser.invoke(ai_msg)   # → str
```

**③ `with_structured_output`——让模型"只按蓝图输出"**

```python theme={null}
model.with_structured_output(
    schema,                # Pydantic 类，如 SupportTicket
    method="function_calling",  # 底层实现：function_calling / json_mode / json_schema
    **kwargs,
) -> Runnable   # invoke 后直接返回 Pydantic 对象（不是 AIMessage！）
```

**④ Pydantic 蓝图类——结构化的"模具"**

```python theme={null}
from pydantic import BaseModel, Field
class SupportTicket(BaseModel):
    intent: str = Field(description="用户意图")
    order_id: str | None = Field(default=None, description="订单号，没提则为空")
    urgency: int = Field(description="紧急程度 1~5")
# with_structured_output 返回：SupportTicket(intent='投诉', order_id='...', urgency=5)
```

> 想验证吗？`type(parser.invoke(ai_msg))` 是 `str`；`type(structured.invoke("..."))` 是 `SupportTicket`——两个出口完全不同。

## 第四幕：边界——结构化输出不是什么都能干

小林用了一周，摸清了 `with_structured_output()` 的边界：

**它能干的**：把模型输出固定成你定义的蓝图（字段名、类型、默认值、枚举）。

**它不能干的**：

* **不能保证"内容正确"**——它保证"格式对"，不保证"数据对"。模型可能把 `order_id` 抽错（比如从"20260701001"里抽成"2026"），Pydantic 只校验类型，不校验语义。
* **不是所有模型都原生支持**——它底层靠模型的"工具调用"或"JSON 模式"能力。老模型不支持时，langchain 会降级用 prompt 硬劝，效果打折。

小林还发现一个坑：**结构化输出的 JSON 结构，跟你直接 `model.invoke()` 返回的 AIMessage 是两码事**。他打印了 `with_structured_output` 返回对象的 `model_dump()`，看到了规整的三个字段；而直接 invoke 返回的是带 `response_metadata`/`usage_metadata` 的 AIMessage。**两条路，出口不同，别搞混。**

> 想验证边界吗？把 `SupportTicket` 的 `intent` 枚举改成 `"物流查询/退换货/投诉/其他"`，然后故意输入一句不提订单号的话——`order_id` 会变成 `None`（默认值生效），但 `intent` 可能被模型猜成别的（Pydantic 不会纠正语义）。

## 结论（小林用一周换来的）

**`stream()` 是打字机：把一次性返回变成增量碎片，总耗时不变但首字更快、体验更顺。`with_structured_output(蓝图类)` 是打印机：让模型只输出你定义的 JSON 形状，Pydantic 帮你校验和转类型，拿到的直接是能用的对象。** 一个解决"怎么出"（流式），一个解决"出什么形状"（结构化）。边界是：流式不省时，结构化不保证内容对——它只保证格式对。

## 复现信号：什么时候你会想起这一章

1. **用户抱怨"回答要等半天才蹦出来"**——把 `invoke()` 换成 `stream()`，for 循环逐个吐。
2. **你想让模型输出 JSON 给程序用**——别用 prompt 劝，用 `with_structured_output(Pydantic类)`。
3. **你看到模型输出夹带"好的，以下是……"**——这就是没结构化的信号，该上蓝图了。
4. **你想校验模型输出对不对**——结构化保证格式，内容对错靠你自己判断。

小林搞定了打字机和 JSON，但他的好奇心又上来了：**模型回答之前，脑子里到底在想什么？能不能把思考过程也打出来？**——[翻到第 6 篇：看得见的思考](/doc/doc/narrative-course/06-思考模式)。