第 2 章 · 02 流式响应(streaming)

第 2 章 · 02 流式响应(streaming)

本节摘要:流式响应(streaming)让 Claude 边生成边返回——用户不必等整段回复写完才看到第一个字,首 token 延迟从「整段生成时间」降到「模型开始输出的瞬间」。本节讲清它的工程用法:Python 的 messages.stream() 上下文管理器与 text_stream、TypeScript 的 for await 迭代与 finalMessage()、六种核心事件(message_start/content_block_start/content_block_delta/content_block_stop/message_delta/message_stop)各自的触发时机,以及如何处理文本、思考(thinking)、工具调用(input_json_delta)三类增量内容。我们还讲了一个关键工程要点:大 max_tokens 必须配流式,否则 SDK 会拒绝预估超过 10 分钟的非流式请求。读完本节,你能用流式改善体验,并知道何时用高层 helper、何时下钻到原始事件。

内容来源:Anthropic 官方 Claude API 文档 python/claude-api/streaming.md 与 typescript/claude-api/streaming.md(随 Claude Code 分发,从泄露素材库提取),汉化并套用体系化模板。

学习目标

阅读完本节,你应当能够:

1. 用 Python 的 messages.stream() 或 TypeScript 的 messages.stream() 写出最简流式调用,实时打印增量文本。
2. 说出六种核心流事件的触发时机与各自携带的信息。
3. 在同一条流里区分文本增量、思考增量、工具输入增量三类内容并分别处理。
4. 用 .get_final_message() / .finalMessage() 在流式结束后拿到完整消息与用量。
5. 解释为什么大 max_tokens 必须配流式,以及如何处理流中断后的不完整响应。

一、为什么要流式:首 token 延迟与体验

非流式调用下,你必须等模型把整段回复全部生成完,才能收到第一个字节。回复越长,等待越久——一个 3000 字的技术解释,用户可能要盯 20 秒空白屏幕。流式则把生成过程切成一个个 token(或 token 组)实时推给你,用户几乎在模型开始输出的瞬间就看到文字流淌。

非流式:  [请求] ────────等20秒──────── [整段返回]
流式:    [请求] ──0.3秒──> 第一个字 ──逐字流出──> [结束]
         ↑首 token 延迟从这里开始         ↑用户体验从这里开始
​

流式不仅改善体验,还有一个硬性工程价值:避免超时。SDK 对非流式请求有一个「预估时长上限」——如果它判断这次请求可能跑超过约 10 分钟(大 max_tokens 加复杂任务),会直接抛 ValueError 拒绝执行,因为闲置长连接会被网络中间层掐断。此时唯一的出路就是开流式。

💡 官方默认建议:凡是可能涉及长输入、长输出或大 max_tokens 的请求,默认就用流式。这能既改善体验又规避超时。如果你不需要逐事件处理,用 SDK 的 .get_final_message() / .finalMessage() 在流结束后拿完整响应即可——它兼顾了流式的「不超时」与非流式的「拿完整结果」。

二、最简流式:Python 与 TypeScript

两门主力语言的流式 API 形态高度一致,都是「拿到一个流对象,迭代它」。

Python——用 messages.stream() 上下文管理器,配合 text_stream 拿纯文本增量:

with client.messages.stream(
    model="claude-opus-4-8",
    max_tokens=64000,
    messages=[{"role": "user", "content": "Write a story"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)   # flush=True 让字立即显示
​

TypeScript——用 messages.stream() 返回的异步可迭代对象,手动过滤事件:

const stream = client.messages.stream({
  model: "claude-opus-4-8",
  max_tokens: 64000,
  messages: [{ role: "user", content: "Write a story" }],
});

for await (const event of stream) {
  if (event.type === "content_block_delta" && event.delta.type === "text_delta") {
    process.stdout.write(event.delta.text);
  }
}
​

💡 务必 flush:Python 要加 flush=True,TypeScript 要用 process.stdout.write() 而非 console.log()——否则输出会被缓冲,用户看到的还是一坨坨而非逐字流淌。这是流式最常被忽略的细节。

高层 vs 低层 helper

Python 提供两层 API:推荐用 messages.stream()(高层,自动累积状态、暴露 text_stream 与 get_final_message());如果只想要原始事件迭代、省内存,可以给 messages.create() 传 stream=True,但这样不会自动累积最终消息。

# 低层:原始事件迭代,无最终消息累积
for event in client.messages.create(
    model="claude-opus-4-8",
    max_tokens=64000,
    messages=[{"role": "user", "content": "Write a story"}],
    stream=True,
):
    print(event.type)
​

异步版本(Python)只需换成 async with 与 async for:

async with async_client.messages.stream(
    model="claude-opus-4-8",
    max_tokens=64000,
    messages=[{"role": "user", "content": "Write a story"}],
) as stream:
    async for text in stream.text_stream:
        print(text, end="", flush=True)
​

三、六种核心事件:流的骨架

理解流,关键是理解这六种事件按什么顺序触发、各携带什么信息。下图是一条「思考 + 文本」消息的完整事件流时序:

一次完整流的事件序列大致如下(以 SSE 原始格式示意,SDK 已替你解析):

message_start        ← 流开始
content_block_start  ← 第一个块(可能是 thinking)
content_block_delta  ← 思考增量(多条)
content_block_stop   ← 思考块结束
content_block_start  ← 第二个块(text)
content_block_delta  ← 文本增量(多条)
content_block_stop   ← 文本块结束
message_delta        ← 带 stop_reason 与 usage
message_stop         ← 流结束
​

⚠️ 关键认知:一条消息可能包含多个内容块——先是一段思考(thinking),再是一段文本,甚至中间还有工具调用块。所以处理流时,要按 content_block_start 的块类型分流,而不是假设「整条流只有文本」。

四、处理三类增量内容

开启扩展思考(adaptive thinking)后,同一条流里会混着思考增量(thinking_delta)与文本增量(text_delta);有工具调用时还会有工具输入增量(input_json_delta)。处理方式是按事件类型分支:

with client.messages.stream(
    model="claude-opus-4-8",
    max_tokens=64000,
    thinking={"type": "adaptive", "display": "summarized"},  # Opus 4.8 等
    messages=[{"role": "user", "content": "Analyze this problem"}],
) as stream:
    for event in stream:
        if event.type == "content_block_start":
            if event.content_block.type == "thinking":
                print("\n[Thinking...]")
            elif event.content_block.type == "text":
                print("\n[Response:]")
        elif event.type == "content_block_delta":
            if event.delta.type == "thinking_delta":
                print(event.delta.thinking, end="", flush=True)
            elif event.delta.type == "text_delta":
                print(event.delta.text, end="", flush=True)
            elif event.delta.type == "input_json_delta":
                pass  # 工具输入正在流式拼装,可暂不处理
​

💡 工具输入也是流式的:当模型决定调用工具时,工具的 JSON 参数不是一次性给出,而是通过多条 input_json_delta 逐片拼成。如果你要在用户面前实时展示「正在调用 get_weather,参数是…」,可以累积这些增量;否则等到 content_block_stop 再统一解析即可。第 2 章 01 节的 Tool Runner 已替你处理了拼装。

五、流式 + 工具调用:Tool Runner 的两层循环

Tool Runner(见第 2 章 01 节)支持流式——给 client.beta.messages.toolRunner(...) 传 stream: true 即可。此时循环结构是两层:外层迭代每次工具循环(每个回合一条消息流),内层处理该回合的流事件。

const runner = client.beta.messages.toolRunner({
  model: "claude-opus-4-8",
  max_tokens: 64000,
  tools: [getWeather],
  messages: [{ role: "user", content: "What's the weather in Paris and London?" }],
  stream: true,
});

// 外层:每次工具循环一个流
for await (const messageStream of runner) {
  // 内层:处理该回合的流事件
  for await (const event of messageStream) {
    if (event.type === "content_block_delta" && event.delta.type === "text_delta") {
      process.stdout.write(event.delta.text);
    }
  }
}
​

Python 同理:client.beta.messages.tool_runner(..., stream=True),每个迭代产出一个流,用 get_final_message() 拿该回合累积消息。这种模式让你在 agent 循环里也能享受流式的实时反馈——用户能实时看到模型「正在思考→正在查天气→正在组织回复」的全过程。

六、拿到最终消息与用量

流式结束后,你通常还想拿到完整消息对象(用于记录用量、判断 stop_reason、存入对话历史)。SDK 提供了 helper,无需自己累积:

with client.messages.stream(
    model="claude-opus-4-8",
    max_tokens=64000,
    messages=[{"role": "user", "content": "Hello"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final_message = stream.get_final_message()    # Python
    print(f"\n\nTokens used: {final_message.usage.output_tokens}")
​

const stream = client.messages.stream({ /* ... */ });
for await (const event of stream) { /* ... */ }
const finalMessage = await stream.finalMessage();   // TypeScript
console.log(`Tokens used: ${finalMessage.usage.output_tokens}`);
​

⚠️ 不要自己包 Promise:TypeScript 文档特别提醒,别把 .on() 事件手动包进 new Promise() 来等结束——finalMessage() 内部已处理了完成、错误、中断所有状态,自己包反而容易漏边界情况。

用量信息除了在 finalMessage().usage 里,也会通过 message_delta 事件实时推送(event.usage.output_tokens)。想在流式过程中实时显示 token 计数,就监听 message_delta:

elif event.type == "message_delta":
    if event.usage and event.usage.output_tokens is not None:
        total_tokens = event.usage.output_tokens
​

七、错误处理与中断恢复

流式中途可能出错:网络断开(APIConnectionError)、被限流(RateLimitError)、服务端 5xx(APIStatusError)。处理方式与非流式类似,但要注意流中断后你手里可能只有不完整内容:

try:
    with client.messages.stream(
        model="claude-opus-4-8", max_tokens=64000,
        messages=[{"role": "user", "content": "Write a story"}],
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
except anthropic.APIConnectionError:
    print("\nConnection lost. Please retry.")
except anthropic.RateLimitError:
    print("\nRate limited. Please wait and retry.")
except anthropic.APIStatusError as e:
    print(f"\nAPI error: {e.status_code}")
​

💡 工程要点:(1)流中断后已打印的内容是「不完整响应」,如果要存对话历史,需重新发请求而非把残文存进去;(2)SDK 默认对 429 与 5xx 自动指数退避重试(默认 2 次),只有需要更强重试逻辑时才自己写;(3)给流式请求设合理超时,长任务尤其要。

本节要点回顾

1. 流式的核心价值:首 token 延迟从「整段生成时间」降到「模型开始输出瞬间」,并规避大 max_tokens 非流式请求的超时拒绝(SDK 拒绝预估超 10 分钟的非流式请求)。
2. 默认开流式:凡长输入/长输出/大 max_tokens 都默认流式;不需要逐事件就用 .get_final_message() / .finalMessage() 拿完整结果。
3. 六种核心事件:message_start → content_block_start → content_block_delta(多条) → content_block_stop → message_delta(带 stop_reason、usage) → message_stop。
4. 一条流可有多个块:先思考、再文本、可能还有工具调用,按 content_block_start 的块类型分流。
5. 三类增量:text_delta(文本)、thinking_delta(思考)、input_json_delta(工具输入,逐片拼)。
6. Tool Runner 两层循环:外层迭代每个回合的流,内层处理事件,实现 agent 循环里的实时反馈。
7. 务必 flush:Python 加 flush=True,TypeScript 用 process.stdout.write();中断后内容不完整,需重发而非存残文。

下一节进入提示缓存(prompt caching)——把长 system prompt 与工具定义缓存住,让重复前缀的成本与延迟降到十分之一。它和流式是 Agent 应用的两大成本利器。