第 1 篇的第一次调用已经跑通：create() 发出请求，等模型把整段回复生成完，一次性拿回完整消息。这个过程里屏幕是静止的——回复越长，静止越久，几百 token 的答案要干等十几秒才整块出现。流式输出改变的就是这一点：回复不再一次性到达，而是边生成边送达，代码收到一段处理一段，用户体验从"等完整块"变成"看打字机"。
本文是《Anthropic SDK 从入门到实战》第 2 篇，把 create() 的流式形态讲透：先用 stream=True 逐个消费原生事件，再理清事件流水线的固定顺序，然后换成官方更推荐的 stream 上下文管理器，对比同步与异步两种消费方式，最后做一个带统计的终端打字机。全文按"价值 → 基础姿势 → 事件流水线 → 便捷姿势 → 异步消费 → 实战 → 总结"组织。
1、为什么需要流式输出
非流式的短板集中在一个指标上：首字延迟（Time To First Token，TTFT）——从发出请求到用户看到第一个字的时间。非流式模式下它等于网络往返加整段生成时长，一段 500 token 的回复按每秒几十 token 的速度生成，用户要盯十几秒空屏；流式模式下模型起笔后第一段增量就发出，首字延迟通常压缩到一两秒，剩下的是"读字跟上生成"的体验。
生成慢不是缺陷，是原理。模型逐 token 自回归生成——第 N 个 token 依赖前 N-1 个 token 的结果，整段回复注定要花时间一个一个"打"出来，总时长无法压缩。流式做的事是把已生成的部分先送出来：总时长没变，用户的等待被摊薄到了阅读过程里。
工程层面还有第二个问题：超时。HTTP 连接长时间没有数据会被中间层掐断，所以 SDK 对非流式调用有一个保护——估算生成时长会超过默认超时上限（约 10 分钟）的非流式请求直接抛 ValueError 拒绝执行。而当前一代模型的最大输出已经到 128K token，想把 max_tokens 开到这个量级，流式是唯一可行路径。两种模式对比如下：
| 对比项 | 非流式 create | 流式 stream |
|---|---|---|
| 返回时机 | 全部生成完成后一次返回 | 首段生成后即开始送达 |
| 首字延迟 | 约等于总生成时长 | 通常一两秒 |
| 长输出 | 可能触发超时保护 | 边生成边送达，无此限制 |
需要注意：流式不改变计费，也不改变生成内容，改变的只是到达方式——token 总量一样算钱，结果一字不差。所以选择标准很纯粹：面向人显示就用流式，机器间后台调用用非流式也毫无问题。
2、基础姿势：stream 开关
在第 1 篇的调用上加一个参数，就是最小的流式示例：
import anthropic
from dotenv import load_dotenv
load_dotenv()
client = anthropic.Anthropic()
events = client.messages.create(
model="claude-sonnet-5-5",
max_tokens=512,
messages=[{"role": "user", "content": "一句话解释什么是流式输出"}],
stream=True, # 唯一改动：打开流式开关
)
for event in events:
print(event.type)
解读：stream=True 不改变请求语义——model、max_tokens、messages、system 与非流式完全一致（2026-10-21 API 版本起 temperature 等采样参数已移除，流式同样不要带），参数清单仍以官方 Messages API 参考为准。改变的是返回：create() 不再阻塞到生成结束才返回 Message 对象，而是立刻返回一个事件流，for 循环每收到一个事件就执行一次循环体。运行后事件名依次打出，形态如下：
message_start
content_block_start
content_block_delta
content_block_delta
content_block_delta
content_block_stop
message_delta
message_stop
解读：中间的 content_block_delta 会重复很多次，每条对应一段文本增量，这就是流式的粒度。真正取文本只需过滤这一类事件，读取它的增量负载：
for event in events:
if event.type == "content_block_delta":
if event.delta.type == "text_delta":
print(event.delta.text, end="", flush=True)
解读：event.delta 是增量对象，纯文本回复里它的 type 为 text_delta，正文在 text 字段。end="" 阻止 print 换行，flush=True 强制立即上屏——这两个参数是打字机效果的最低配置。此外还有两类辅助事件：ping 是服务端的心跳保活，忽略即可；error 在生成中途出错时出现，SDK 会把它抛成异常而不是静默吞掉。完整事件定义见官方流式文档。
3、事件流水线全景
第 2 节的事件不是随机到达，而是一条固定顺序的流水线。一次纯文本回复的时序如下：
解读：外层是消息级事件，一头一尾各一次；内层是内容块级事件——第 1 篇讲过回复是块数组，流式就是把数组里的块"边装配边交付"，每个块的开启、增量、结束各对应一类事件。一次回复有多个块时，中间三层会按块重复。六类核心事件的职责如下：
| 事件 | 到达时机 | 携带内容 |
|---|---|---|
message_start |
开头一次 | 消息元信息与输入用量 |
content_block_start |
每个内容块开始 | 块类型与序号 |
content_block_delta |
每段增量 | text_delta 等增量负载 |
content_block_stop |
每个内容块结束 | 块完结标记 |
message_delta |
生成收尾时 | 停止原因与累计输出用量 |
message_stop |
末尾一次 | 消息完结标记 |
两个细节值得记住。一是**stop_reason 只在 message_delta 里出现**——message_start 携带的消息对象里它还是空值，流式代码要判断结束原因，得等倒数第二个事件。二是用量分布在两处：message_start 带输入侧的 input_tokens，message_delta 带累计的 output_tokens，后者随增量增长实时更新，拿来做进度指示正合适。
为了建立实感，把一条 content_block_delta 事件序列化成 JSON，线上传输的原始形状如下：
{
"type": "content_block_delta",
"index": 0,
"delta": { "type": "text_delta", "text": "流式" }
}
解读：index 标明这段增量属于第几个内容块——纯文本回复通常是 0，一次回复出现多个块时靠它把增量归位到正确的块；delta.text 就是要上屏的片段。SDK 已把这些 JSON 解析成 Python 对象，代码里属性访问即可；原始形状的价值在调试网络层、或日后用其他语言对接时能对上口径。
4、便捷姿势：上下文管理器
第 2 节的写法把六类事件全部交给手动处理，灵活但啰嗦。多数场景只需要"逐字拿文本、结束后拿完整结果"，SDK 为此提供了更高一层的 stream 上下文管理器，官方 SDK 文档把它列为推荐用法：
with client.messages.stream(
model="claude-sonnet-5-5",
max_tokens=512,
messages=[{"role": "user", "content": "一句话解释什么是流式输出"}],
) as stream:
for text in stream.text_stream: # 只产出文本增量
print(text, end="", flush=True)
final = stream.get_final_message() # 迭代结束后取完整消息
print(f"\n停止原因：{final.stop_reason}")
print(f"输出 token：{final.usage.output_tokens}")
解读：text_stream 是过滤好的文本迭代器，等价于第 2 节两层 if 的展开；get_final_message() 返回 SDK 在幕后自动累积的完整 Message 对象——content 块数组、stop_reason、usage 一应俱全，形状与第 1 篇的非流式响应完全一致，累积逻辑的实现在源码仓库里可以直接读到。换句话说，流式与非流式可以共用同一套后续处理代码：过程用流式改善体验，结果落回同一个对象。with 语法还保证连接正常关闭，中途抛异常也不泄漏。
两点补充。管理器里同样能直接迭代原生事件（for event in stream），stream=True 的独立价值在于不做任何累积、内存占用更低，处理超大输出时更合适；另外同一个流里 text_stream 与原生事件迭代不要混用——两者消费同一串事件，混用会互相"抢"数据，取其一即可。
5、同步与异步两种消费
前面全部是同步代码：迭代期间当前线程被阻塞，对脚本和命令行工具正合适。但如果消费方是 Web 服务，一个线程被一个流占满，并发能力立刻见顶——这类场景该换 AsyncAnthropic：
import asyncio
import anthropic
from dotenv import load_dotenv
load_dotenv()
client = anthropic.AsyncAnthropic()
async def main():
events = await client.messages.create(
model="claude-sonnet-5-5",
max_tokens=512,
messages=[{"role": "user", "content": "一句话解释什么是流式输出"}],
stream=True,
)
async for event in events: # 异步迭代，等待时让出线程
if event.type == "content_block_delta":
if event.delta.type == "text_delta":
print(event.delta.text, end="", flush=True)
asyncio.run(main())
解读：异步版的事件类型与流水线和同步版完全一致，差别只有两处——客户端类换成 AsyncAnthropic，迭代换成 await 加 async for。等待下一段事件期间事件循环是空闲的，同一个循环可以同时挂几十路流，这是 Web 服务做打字机接口的标准形态。顺带一提，两个客户端类共享同一套事件类型定义，只是传输层的两种驱动方式，本文的事件知识对两者通用。两种客户端的适用场景对比如下：
| 对比项 | 同步 Anthropic | 异步 AsyncAnthropic |
|---|---|---|
| 迭代写法 | for event in events |
async for event in events |
| 等待行为 | 阻塞当前线程 | 让出事件循环 |
| 典型场景 | 脚本、CLI、批处理 | Web 服务、高并发网关 |
| 并发手段 | 多线程或多进程 | 单循环跑多任务 |
选择标准一句话：单路消费用同步，一路以上就该异步。
6、实战：终端打字机
把前面的零件组装起来：逐字上屏、记录首字延迟、结束后打印总用量。完整代码如下：
import time
import anthropic
from dotenv import load_dotenv
load_dotenv()
client = anthropic.Anthropic()
start = time.perf_counter() # 请求发出时刻
first = None # 首字时刻
chars = 0 # 累计字符数
with client.messages.stream(
model="claude-sonnet-5-5",
max_tokens=2048,
messages=[{"role": "user", "content": "用两百字介绍 SSE 协议"}],
) as stream:
for text in stream.text_stream:
if first is None:
first = time.perf_counter()
chars += len(text)
print(text, end="", flush=True)
final = stream.get_final_message()
total = time.perf_counter() - start
ttft = first - start
print(f"\n—— 字符 {chars}，总耗时 {total:.1f} s，首字 {ttft:.2f} s")
print(f"—— 输出 token {final.usage.output_tokens}，停止原因 {final.stop_reason}")
解读：计时用 time.perf_counter()，首字时刻在第一段增量到达时记录一次。运行时终端逐字蹦字，结束后追加两行统计——首字延迟是流式价值最直观的数字，通常只有总耗时的零头；output_tokens 直接取自 get_final_message()，不必自己数。首字延迟还是横向对比的常用口径：换模型、换网络路径各测一轮，数字会明显分层，第 9 篇讲选型时会再用到它。把提问换成让模型写长文，再对比首字与总耗时，对流式的收益会有更具体的体感。
7、总结
本篇在第 1 篇的非流式调用上加了 stream=True：六类事件按 message_start、内容块三连、message_delta、message_stop 的固定顺序到达，文本藏在 content_block_delta 的 text_delta 里；日常首选 with client.messages.stream(...) 上下文管理器——text_stream 逐字拿文本，get_final_message() 落回与非流式一致的完整对象；Web 服务用 AsyncAnthropic 异步迭代。流式不改变计费与结果，只改变到达方式，但它既是长输出的唯一路径，也是交互体验的下限保障。
留两个练习：其一，给第 6 节的打字机加速率统计——用字符数除以总耗时减首字延迟的差值得到每秒字符数，再用 output_tokens 同样换算每秒 token 数，对 claude-sonnet-5-5 与 claude-haiku-5-5 各测三次取平均，你会拿到两组真实的生成速度数字；其二，统计一次回复里各类事件的出现次数，验证 message_start 与 message_stop 恰好各一次、content_block_delta 与文本段数对应。对于准备把 Claude 接进自己界面的开发者，建议现在就把打字机跑起来——首字延迟的差距一旦亲眼见过，就回不去了。下一篇是第 3 篇：结构化输出与 JSON 校验，让回复从自由文本变成可靠的机器可读数据。
