> ## 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.

# 04 · 输出解析与流式：StrOutputParser 与打字机

# 04 · 输出解析与流式：StrOutputParser 与打字机

> **本片目标**：解决"输出"两个痛点——① 剥壳取文本（AIMessage → str）；② 流式输出（逐字显示，SSE 网络模型）。同时第一次接触 **LCEL 管道（`|`）**，为第 09 篇 RAG 组装埋伏笔。
> **新增规定性：4**（输出用 `StrOutputParser`；流式用 `.stream()`；**`|` 管道开始出现**）
> **数据字典**：StrOutputParser / AIMessageChunk。
> **进程线程模型**：`stream()` 是生成器——**逐块 yield**，每块到达就处理，不等待全部。
> **网络模型**：流式 = **SSE**（Server-Sent Events，HTTP 长连接逐事件推送）。

***

## 1. 上集回顾

第 03 篇我们有模板了。但两个输出问题没解决：

1. `model.invoke(messages)` 返回 AIMessage，我想要纯字符串——每次 `.content` 太啰嗦，而且后面接 RAG 链时每个环节都要"字符串进、字符串出"，AIMessage 这个壳会碍事。
2. 模型生成 100 个 token 要 2 秒，`invoke` 是**憋到最后才返回**。用户盯着空白 2 秒——体验极差。ChatGPT 那种逐字蹦出来是怎么做到的？

***

## 2. StrOutputParser：剥壳取文本

```python theme={null}
from langchain_core.output_parsers import StrOutputParser

parser = StrOutputParser()
text = parser.invoke(ai_msg)    # AIMessage → str
# 等价于: text = ai_msg.content
```

### 2.1 StrOutputParser 返回值（1.x 注意！）

```python theme={null}
from langchain_core.output_parsers import StrOutputParser

parser = StrOutputParser()
text = parser.invoke(ai_msg)    # AIMessage → TextAccessor（str 的子类）
```

| 方法       | 签名                 | 返回值           | 说明             |                                       |
| -------- | ------------------ | ------------- | -------------- | ------------------------------------- |
| `invoke` | \`invoke(AIMessage | str) -> str\` | `TextAccessor` | 收 AIMessage 自动取 `.content`，收 str 原样返回 |

> **⚠️ 1.5.x 实测类型**：`StrOutputParser().invoke(ai_msg)` 返回的不是裸 `str`，而是 `langchain_core.messages.base.TextAccessor`。它是 `str` 的子类（`isinstance(x, str) == True`，`str(x)` 正常），但 `type(x).__name__` 是 `'TextAccessor'`。**所有字符串操作（拼接、切片、正则）都照常工作**，只是类型名不同——别被调试输出吓到。这是 langchain-core 1.0 引入的过渡类（为了兼容旧的 `.text()` 方法调用）。

**它为什么存在**：`StrOutputParser` 实现了一个通用接口 `BaseOutputParser`。以后你要自定义解析器（比如第 05 篇的 Pydantic 解析），只需实现 `parse` 方法。它让"模型输出 → 应用所需格式"这件事有了统一的抽象。

***

## 3. 流式输出：`.stream()` 与 AIMessageChunk

```python theme={null}
for chunk in model.stream("讲一个笑话"):
    print(chunk.content, end="", flush=True)
```

### 3.1 数据字典：AIMessageChunk

| 字段                  | 类型              | 说明                                      |                             |
| ------------------- | --------------- | --------------------------------------- | --------------------------- |
| `content`           | `str`           | **当前这一块**的文本碎片（不是累积的！）                  |                             |
| `id`                | \`str           | None\`                                  | 与最终 AIMessage 相同的 id        |
| `response_metadata` | `dict`          | 中间块基本为空；**末块**有 `stop_reason` 和 `usage` |                             |
| `usage_metadata`    | \`UsageMetadata | None\`                                  | 中间块为 None；**末块**有完整 token 数 |

**流式协议特征（实测）**：

* **第一块是"空壳"**：`id` 已就位、`content` 为空——用于先确定这次 run 的 id。
* **中间块**：`content` 是碎片（几个字），`usage_metadata` 是 `None`。
* **末块**：`content` 可能为空，但 `response_metadata["stop_reason"] == "end_turn"`、`usage_metadata` 有值——**表示生成结束**。

```python theme={null}
for i, chunk in enumerate(model.stream("你好")):
    print(f"块{i}: content={chunk.content!r} usage={chunk.usage_metadata}")
# 块0: content=''       usage=None          ← 空壳（确立 id）
# 块1: content='你好'    usage=None
# 块2: content='！'      usage=None
# 块3: content=''       usage={'input_tokens':..,'output_tokens':..}  ← 末块（带 token 数）
```

### 3.2 手动累积（你通常不用做）

框架层面，`invoke = 把 stream 所有块拼起来`。你不需要手动拼，但要知道：**invoke 和 stream 是同一个底层的两种视图**。

***

## 4. LCEL 管道（`|`）——预告

第 03 篇的模板 + 第 04 篇的解析器，可以串起来：

```python theme={null}
chain = prompt | model | StrOutputParser()
text = chain.invoke({"company": "Z.ai", "history": [], "question": "你好"})
# 输入 dict → 模板 → AIMessage → str
```

**`|` 是什么**：LangChain 表达式语言（LCEL）。`a | b` 表示"a 的输出作为 b 的输入"。它要求两边的对象都实现 `Runnable` 协议（都有 `invoke`）。

**类型流**：

```
dict {"company":...} → prompt.invoke → PromptValue
  → 转消息列表 → model.invoke → AIMessage
  → parser.invoke → str
```

> 现在只需要眼熟。第 09 篇 RAG 组装会用到完整的 `|` 组合（含 `RunnablePassthrough`）。先记住：**prompt | model | parser 是 LangChain 1.x 的标准三段式**。

***

## 5. 进程线程模型

```
for chunk in model.stream(...):
    # 生成器：每次迭代，当前线程被 SSE 推送唤醒一次
    # 处理完这块（打印/攒起来），挂起等下一块
```

**关键**：`stream()` 是惰性生成器。它不是"一次性取完再返回"，而是"来一块给一块"。阻塞发生在**每次迭代**（等待网络），但每次只等一小段（几十毫秒），所以用户体验是"逐字蹦"。

***

## 6. 网络模型：SSE 长连接

```
model.stream("你好")
  └─ POST .../v1/messages  （带 stream: true）
       │
       └─ 连接保持打开（HTTP 长连接），服务端逐事件推送：
           event: content_block_start   ← 空壳
           event: content_block_delta   ← {"delta":{"text":"你"}}
           event: content_block_delta   ← {"delta":{"text":"好"}}
           event: message_delta         ← {"stop_reason":"end_turn"}
           event: message_stop
       │
       └─ 每个事件被 SDK 包装成 AIMessageChunk，逐块 yield
```

| 网络特征 | 值                                   |
| ---- | ----------------------------------- |
| 协议   | HTTPS + **SSE**（Server-Sent Events） |
| 请求   | 一次 POST，`stream: true`              |
| 响应   | 分块推送（chunked transfer），非一次性 JSON    |
| 结束信号 | `message_delta` 事件里的 `stop_reason`  |

> 本机已装 `httpx-sse 0.4.3`——就是 SDK 解析 SSE 流的底层库（第 12 篇细讲）。

**实测时序（智谱 glm-4.7，2026-08）**：

* 正常时：`stream` 首个内容块 ≈ **0.5s** \< `invoke` 总耗时 ≈ **2.0s**——首 token 到达更快，这是流式最核心的体验优势。
* 偶发：首 token 延迟可能飙到 10s+（API 侧波动）。**结论：流式降低的是"首字等待"，不保证"总时间"更短**——`invoke` 和 `stream` 生成同样内容的总耗时差不多，流式的价值是**早开始显示**。

***

## 7. 验证：跑起来

配套代码 `code/04_streaming.py`：

1. `StrOutputParser` 剥壳演示；
2. `model.stream()` 逐块打印 content + usage（看到空壳/碎片/末块三态）；
3. `prompt | model | parser` 管道第一次完整跑通；
4. 计时对比 `invoke` vs `stream` 完成时间（验证 invoke 憋到最后）。

```powershell theme={null}
cd enterprise-rag-course\code
python 04_streaming.py
```

**预期输出（节选）**：

```
===== ① StrOutputParser =====
parser.invoke(AIMessage) → TextAccessor: 你好！...
（TextAccessor 是 str 的子类，所有字符串操作照常）

===== ② 流式输出 =====
块0: content=''       usage=None          ← 空壳
块1: content='我是'    usage=None
...
块N: content=''       usage={'input_tokens': 9, 'output_tokens': 23, ...}   ← 末块

===== ③ 管道 prompt | model | parser =====
回答: 你好！我是Z.ai公司的智能助手。...
```

***

## 8. 边界

* **`chunk.content` 是碎片不是累积**——想拿完整答案，要么自己 `''.join`，要么直接用 `invoke`。
* **`invoke` 也能拿到流式数据**——`invoke` 内部就是消费 stream。区别只在"要不要逐块处理"。
* **SSE 需要长连接**——代理/网关配置不当会切断流式。生产环境要注意超时设置（第 15 篇）。
* **`StrOutputParser` 对多模态 content 无能为力**——content 是块列表（图片等）时，解析器只取字符串部分。

***

## 推荐资料（延伸阅读）

* [LangChain 官方文档 · 流式](https://docs.langchain.com/oss/python/langchain/streaming) —— stream/astream 与事件流
* [langchain-core API 参考 · output\_parsers](https://reference.langchain.com/python/langchain-core/output_parsers/) —— 解析器家族的全部实现
* [LangChain 官方文档 · 事件流式](https://docs.langchain.com/oss/python/langchain/event-streaming) —— 事件驱动的流式进阶

***

## 9. 未完待续

"输出是文本"解决了，但业务往往要的是**结构化数据**：客服工单要 `{"intent": "投诉", "order_id": "2026..."}`、RAG 要带引用来源、报表要数字。让模型"保证输出 JSON 且字段正确"——这就是结构化输出。

→ [05 · 结构化输出](/doc/doc/enterprise-rag-course/05-结构化输出)