同样是问 Claude 一个问题，同步调用要等 10 秒才吐出整段答案，流式调用第 1 秒就开始逐字往外蹦。用户体验的差别，比"省了多少钱"更直接——因为用户看得见。

一、先说结论：用户等不起整段，但等得起逐字

我在香港做 FinTech，做的工具里有一块是对话式的 AI 助手——用户问一个财务问题，Claude 生成分析。

最开始我用同步调用（messages.create），问题来了：一个稍微复杂点的问题，Claude 要生成几十秒，这几十秒里界面是一动不动的。用户以为卡死了，关掉重开，重开又等几十秒。体验烂到我自己都看不下去。

直到我切到流式输出（stream=True）：Claude 生成一个 token，界面就蹦一个字。用户看得见进展，等待的焦虑感瞬间消失。哪怕总时长没变，主观感受天差地别。

这篇文章我把 Claude 流式输出的原理、事件类型、生产级踩坑一次性讲清楚。核心结论先放这：任何用户实时等待的对话场景，都必须用流式输出——它不是优化，是底线。

收藏提示①：文末的流式封装函数可以直接复用，套到你的对话工具里，用户立刻能感受到差别。

二、环境信息

项	版本
Python	3.13.12
anthropic SDK	最新版
模型	claude-sonnet-4-6（示例用）
说明	流式输出不改变计费，输入输出 token 与非流式完全一样

三、最简实现：一个参数，逐字输出

同步调用和流式调用的差别，本质上就一个 stream=True：

import anthropic

client = anthropic.Anthropic()

# 同步：等全部生成完，一次性返回
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "帮我分析一下这个季度的现金流"}],
)
print(response.content[0].text)   # 几十秒后，整段打印

# 流式：逐 token 返回，实时打印
with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "帮我分析一下这个季度的现金流"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)   # 每个 chunk 实时打印

上面用 text_stream 是 SDK 提供的最高层封装——它已经帮你把底层的增量事件拼接成了一段段可读文本。对"只想让文字逐字蹦出来"的场景，这就够了。

收藏提示②：stream.text_stream 是最省事的入口，一个 for 循环 + flush=True 就能实现逐字输出。记住 flush=True——不写的话，Python 会缓冲，你看到的还是"卡一下然后整段蹦"。

四、深入一点：7 种事件类型

上图：橙色高亮的 content_block_delta 是核心——逐字拿文本就在这里。其他事件负责开始/结束/心跳。

text_stream 够用，但如果你想做打字指示器、实时 token 计数、工具调用的进度动画，就得碰底层事件流了。

Claude 的流式响应由一串事件组成，主要就这几种：

事件类型	含义	你什么时候关心它
message_start	流开始，含元信息	埋点、计时开始
content_block_start	一个内容块开始	多块内容时标记起点
content_block_delta	增量内容（核心）	逐字拿文本就在这里
content_block_stop	内容块结束	块级收尾
message_delta	消息级变化（stop_reason / usage）	实时 token 计数
message_stop	流结束	结束收尾
ping	心跳保活	一般忽略

其中 content_block_delta 里还分三种子类型：text_delta（文本）、input_json_delta（工具参数的增量 JSON）、thinking_delta（扩展思考）。

手动处理事件流的完整写法：

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "写一首关于调试的俳句"}],
) as stream:
    for event in stream:
        if event.type == "content_block_delta" and event.delta.type == "text_delta":
            print(event.delta.text, end="", flush=True)
        elif event.type == "message_delta":
            # 实时拿到当前已生成的 output token 数
            print(f"\n[tokens so far: {event.usage.output_tokens}]", end="")
        elif event.type == "message_stop":
            print("\n[Done]")

五、踩坑记录：这四个坑，别踩

1. 别对非 delta 事件访问 .text：事件流里只有 content_block_delta 才有 delta.text，其他事件类型直接访问会抛 AttributeError。必须先用 event.type == "content_block_delta" 过滤。这是新手最常见的报错。
2. flush=True 不能省：Python 的 stdout 有缓冲，不写 flush=True，你会看到"卡几秒 → 整段蹦出来"，等于流式白做。写进循环里。
3. 流式也会中途断：连接可能在返回 200 之后、流进行到一半时断开（限流、网络抖动）。要 try/except anthropic.APIError 包住整个流，断的时候给用户展示"已生成的部分 + 重试提示"，而不是直接崩。
4. 客户端断开要提前止损：如果用户关了页面，你的后端还在继续生成，等于白白烧 output token。做 Web 服务时，在循环里检查连接是否还活着，断了就退出流——async with 块会在退出时自动调用 stream.close() 停止计费。

六、流式 vs 批量：什么时候用哪个

上图：一句话核心——用户等不等？等就流式，不等就批量。流式不省钱（全价），但买的是"看得见进展"；批量省一半，但用户只能等几小时。

上一篇讲了 Batch API（省 50%），这篇讲 Streaming（实时）。两个正好是一对，决策就一个问题：

场景	选择	核心原因
用户实时盯着屏幕	流式	首字延迟（TTFT）最重要，逐字蹦出来留住用户
批量分类、离线评测	Batch	用户不等，省 50%
结果几分钟内就要	同步	Batch 至少几小时

一句话：交互走流式，批处理走 Batch。流式不省一分钱（token 计价与非流式完全一样），但它买的是用户体验——而这个体验，直接决定你的产品会不会被用户关掉。

七、写在最后

流式输出这件事，技术上毫无门槛——就一个 stream=True，加一个 for 循环。但它对用户体验的影响，比很多人以为的大得多：一个会"逐字蹦"的 AI，和一个"卡半天吐一大段"的 AI，用户对它们的耐心完全是两回事。

把流式接进去之后，你还会发现一个连锁反应：用户更愿意问更长、更复杂的问题了——因为他不再担心"卡死"，知道系统在动。这对一个对话产品来说，是比任何 prompt 技巧都底层的改进。

需要说明的是，事件类型、SDK 接口这些会随版本演进，本文以写作时的官方文档为准，动手前查一下最新版 SDK 的接口说明。

收藏提示③：如果这篇帮你把对话工具的体验提上来了，收藏 + 点赞，下次接流式时直接回来抄代码。