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

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

1. 上集回顾

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

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

2. StrOutputParser：剥壳取文本

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 注意！）

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

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 有值——表示生成结束。

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 篇的解析器，可以串起来：

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：

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

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 是块列表（图片等）时，解析器只取字符串部分。

9. 未完待续

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

RAG 教程

小林学 LangChain

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

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

1. 上集回顾

2. StrOutputParser：剥壳取文本

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

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

3.1 数据字典：AIMessageChunk

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

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

5. 进程线程模型

6. 网络模型：SSE 长连接

7. 验证：跑起来

8. 边界

推荐资料（延伸阅读）

9. 未完待续

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

​1. 上集回顾

​2. StrOutputParser：剥壳取文本

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

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

​3.1 数据字典：AIMessageChunk

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

​4. LCEL 管道（|）——预告

​5. 进程线程模型

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

​7. 验证：跑起来

​8. 边界

​推荐资料（延伸阅读）

​9. 未完待续

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

1. 上集回顾

2. StrOutputParser：剥壳取文本

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

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

3.1 数据字典：AIMessageChunk

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

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

5. 进程线程模型

6. 网络模型：SSE 长连接

7. 验证：跑起来

8. 边界

推荐资料（延伸阅读）

9. 未完待续