本文介绍 Claude API Streaming 流式输出的配置和解析方法。Claude API 的流式输出基于 SSE，可以将模型生成过程中的内容分批返回，常用于 AI 聊天机器人、AI 写作、代码生成、长文总结和 Agent 工具调用等场景。

文章将重点说明如何开启 stream 参数，如何使用 cURL、Node.js 和 Python 接收流式响应，如何解析 message_start、content_block_delta、text_delta、message_delta 等事件，以及如何在前端展示过程中处理拼接、渲染、中断和异常重试问题。

先说结论：Claude API 流式输出适合什么场景？

Claude API Streaming 并不会让模型本身“思考得更快”。它真正提升的是用户感知上的速度。

更具体一点，它主要优化这几个指标：

• TTFT：Time To First Token，也就是首个有效文本返回时间；
• 首屏可读时间：用户看到第一句话、第一段内容的时间；
• 用户感知等待时间：从“空等转圈”变成“内容正在生成”；
• 可中断成本：用户点停止后，可以尽早取消后续输出，避免继续消耗。

一般来说，下面这些场景很适合使用 Claude API 流式输出：

场景	是否推荐 Streaming	原因
AI 聊天机器人	推荐	用户需要尽快看到反馈
AI 写作、长文总结	推荐	输出内容较长，流式体验提升明显
代码生成、代码解释	推荐	用户可以边看边判断是否需要停止
Agent / Tool Use	推荐	工具调用过程可以及时展示出来
后台批处理	不推荐	不需要实时展示，非流式更简单
短文本分类、打标签	不推荐	输出太短，Streaming 收益有限
严格 JSON 结构化输出	谨慎使用	流式 JSON 需要额外拼接和校验

所以核心结论其实很简单：Streaming 提升的是感知性能，不一定会明显缩短完整响应的总耗时。

---

Claude API Streaming 的基本原理：SSE 和事件流

Claude API Streaming 基于 SSE，也就是 Server-Sent Events。

普通的非流式请求，会等模型把结果全部生成完，再一次性返回一个 JSON。流式请求则不同，它会持续返回一系列事件，例如：

message_start
content_block_start
content_block_delta
content_block_stop
message_delta
message_stop

常见事件大概可以这样理解：

事件	作用
message_start	一条消息开始
content_block_start	一个内容块开始，可能是文本，也可能是工具调用
content_block_delta	内容增量，最常见的是 text_delta
content_block_stop	当前内容块结束
message_delta	消息级别的增量，可能包含 usage 信息
message_stop	整条消息结束
ping	保活事件
error	流式过程中发生错误

这里有个很重要的点：开发时不要把整个返回当成普通 JSON 一口气解析，而是要按事件类型分别处理。

尤其是在 Tool Use 场景里，input_json_delta 返回的是 partial_json，也就是 JSON 的一部分。它还不是完整 JSON，所以不能每收到一小段就直接 JSON.parse，否则很容易报错。

---

Claude API Streaming 最小配置：先用 cURL 跑通

最小配置其实很简单，请求体里加上 stream: true 就可以了。

curl -N https://api.anthropic.com/v1/messages \
  -H "x-api-key: $ANTHROPIC_API_KEY" \
  -H "anthropic-version: 2023-06-01" \
  -H "content-type: application/json" \
  -d '{
    "model": "claude-3-5-sonnet-20241022",
    "max_tokens": 800,
    "stream": true,
    "messages": [
      {
        "role": "user",
        "content": "用三段话解释 Claude API Streaming 的优势。"
      }
    ]
  }'

这里有几个地方很容易被忽略：

• -N：关闭 cURL 的输出缓冲。不加的话，你可能会误以为没有流式返回；
• stream: true：开启 Claude API Streaming；
• anthropic-version：指定 API 版本；
• max_tokens：控制最大输出长度，避免生成过长；
• model：模型名请以官方当前可用列表为准，不要直接照抄旧示例。

怎么判断是不是真的流式？很简单，如果终端里的内容是一段一段冒出来的，而不是等很久后一次性打印出来，说明流式链路基本没问题。

---

Node.js 接入 Claude API 流式输出

先安装 SDK：

npm install @anthropic-ai/sdk

一个基础示例如下：

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY,
});

const stream = await client.messages.create({
  model: "claude-3-5-sonnet-20241022",
  max_tokens: 800,
  stream: true,
  messages: [
    { role: "user", content: "写一段关于 Claude API 流式输出的说明。" }
  ],
});

let fullText = "";

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

  if (event.type === "message_delta") {
    // 部分 SDK / 事件中可能会在这里提供 usage 信息
    // 具体字段请以当前 SDK 返回为准
  }

  if (event.type === "message_stop") {
    console.log("\n\n生成结束");
  }
}

在生产环境里，建议加上 AbortController。用户点击“停止生成”时，可以立刻取消上游请求，避免继续生成和计费。

const controller = new AbortController();

const stream = await client.messages.create(
  {
    model: "claude-3-5-sonnet-20241022",
    max_tokens: 1200,
    stream: true,
    messages: [{ role: "user", content: "生成一篇长文。" }],
  },
  {
    signal: controller.signal,
  }
);

// 用户停止时调用
// controller.abort();

实际开发中，比较推荐下面这些做法：

做法	是否推荐
只拼接 text_delta	推荐
把所有事件都当文本拼接	不推荐
支持用户中断	推荐
只在前端做假打字机效果	不推荐

换句话说，真正的流式应该来自模型输出本身，而不是前端拿到完整内容后再模拟打字机效果。

---

Python 接入 Claude API 流式输出

先安装 SDK：

pip install anthropic

如果只是想拿到文本流，可以这样写：

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-3-5-sonnet-20241022",
    max_tokens=800,
    messages=[
        {"role": "user", "content": "解释 Claude API Streaming 配置的关键点。"}
    ],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

如果你需要更细地处理事件，比如区分文本、Tool Use、usage 等，可以遍历完整事件流：

with client.messages.stream(
    model="claude-3-5-sonnet-20241022",
    max_tokens=800,
    messages=[{"role": "user", "content": "输出一段 Markdown 示例。"}],
) as stream:
    full_text = ""

    for event in stream:
        if event.type == "content_block_delta":
            delta = event.delta
            if getattr(delta, "type", None) == "text_delta":
                full_text += delta.text
                print(delta.text, end="", flush=True)

        if event.type == "message_stop":
            print("\n生成完成")

Python 服务端尤其要注意两点。

第一，控制台输出时记得加 flush=True。否则本地看起来可能不像流式，而像攒了一段之后才输出。

第二，如果通过 Web 框架转发给浏览器，要确认响应缓冲已经关闭。不然即使 Claude API 是流式返回，浏览器也可能还是一次性收到内容。

---

前端怎么展示 Claude 流式输出：EventSource 和 Fetch Stream

浏览器端不要直接请求 Claude API，因为这样会暴露 API Key。比较安全、也更符合生产环境的架构是：

浏览器 → 业务后端 → Claude API

也就是说，前端只请求自己的业务后端。鉴权、调用 Claude、转发流、记录日志这些事情，都交给后端处理。

如果后端提供的是 SSE GET 接口，前端可以用 EventSource：

const es = new EventSource("/api/chat-stream?conversationId=123");

es.onmessage = (event) => {
  appendText(event.data);
};

es.onerror = () => {
  es.close();
};

不过很多聊天接口需要 POST 请求，还要携带复杂 body 或自定义鉴权。这种情况下，更推荐使用 fetch + ReadableStream：

const controller = new AbortController();

const res = await fetch("/api/chat-stream", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ message: "你好，介绍一下 Claude API 流式输出" }),
  signal: controller.signal,
});

const reader = res.body.getReader();
const decoder = new TextDecoder();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;

  const chunk = decoder.decode(value, { stream: true });
  appendText(chunk);
}

// 停止生成
// controller.abort();

前端渲染也别太激进。不要每来一个 token 就重新渲染整篇 Markdown，这样很容易卡，尤其是长文本和代码块。

更稳妥的做法是：每 30–100ms 批量刷新一次。代码块还没闭合时，可以先按纯文本展示，等生成结束后再做完整 Markdown 渲染和代码高亮。这样用户体验会稳定得多。

---

服务端转发 SSE 的生产配置

很多人本地测试时流式好好的，一上线就变成一次性返回。原因通常不是代码错了，而是代理、网关或 CDN 把响应缓冲了。

后端转发 SSE 时，建议设置这些响应头：

Content-Type: text/event-stream
Cache-Control: no-cache, no-transform
Connection: keep-alive
X-Accel-Buffering: no

一个 Node.js / Express 示例：

app.post("/api/chat-stream", async (req, res) => {
  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache, no-transform");
  res.setHeader("Connection", "keep-alive");
  res.setHeader("X-Accel-Buffering", "no");

  const controller = new AbortController();

  req.on("close", () => {
    controller.abort();
  });

  try {
    const stream = await client.messages.create(
      {
        model: "claude-3-5-sonnet-20241022",
        max_tokens: 1000,
        stream: true,
        messages: [{ role: "user", content: req.body.message }],
      },
      { signal: controller.signal }
    );

    for await (const event of stream) {
      if (
        event.type === "content_block_delta" &&
        event.delta?.type === "text_delta"
      ) {
        res.write(`data: ${JSON.stringify(event.delta.text)}\n\n`);
      }
    }

    res.write("event: done\ndata: {}\n\n");
    res.end();
  } catch (err) {
    res.write(`event: error\ndata: ${JSON.stringify({ message: "stream error" })}\n\n`);
    res.end();
  }
});

如果前面还有 Nginx，需要关闭代理缓冲：

location /api/chat-stream {
    proxy_pass http://backend;
    proxy_buffering off;
    proxy_cache off;
    proxy_read_timeout 300s;
}

另外，CDN、Serverless、API Gateway 这些中间层也可能缓冲响应。部署前一定要实际验证：内容是不是逐段返回，而不是最后一次性吐出来。

---

Tool Use 场景下怎么解析流式事件

Tool Use 的流式处理会比普通文本复杂一点，因为工具参数可能会通过 input_json_delta 分段返回。

比如你可能收到类似这样的内容：

content_block_start: tool_use
content_block_delta: input_json_delta partial_json="{\"query\":\"Claude"
content_block_delta: input_json_delta partial_json=" Streaming\"}"
content_block_stop

错误做法是每收到一段就立刻解析：

JSON.parse(partialJson); // 容易报错

因为这时候的 partial_json 还不是完整 JSON。

正确做法是按 content_block 的 index 缓存起来，等 content_block_stop 到了之后，再统一解析：

const toolJsonBuffer = new Map();

function handleEvent(event) {
  if (
    event.type === "content_block_delta" &&
    event.delta?.type === "input_json_delta"
  ) {
    const old = toolJsonBuffer.get(event.index) || "";
    toolJsonBuffer.set(event.index, old + event.delta.partial_json);
  }

  if (event.type === "content_block_stop") {
    const json = toolJsonBuffer.get(event.index);
    if (json) {
      const args = JSON.parse(json);
      // 执行工具调用
    }
  }
}

Tool Use 这里最关键的一点是：文本、工具参数、thinking delta 不要粗暴拼成同一个字符串。
不同类型的事件应该分开处理，否则后面很容易出现解析失败、展示混乱或者状态不同步的问题。

---

性能实测：流式和非流式到底差在哪？

做 Claude API 性能优化时，建议至少记录下面这些指标：

指标	定义
TTFT	请求发出到首个有效文本 token 返回
TTL	请求发出到 message_stop 或完整响应结束
Tokens/sec	输出 token 数 / 生成耗时
首屏可读时间	前端出现可读句子或段落的时间
中断节省	用户停止后减少的后续输出 token

测试时不要只测一种任务，最好覆盖不同长度和不同前端渲染方式：

测试类型	输出长度	目的
短文本	100–300 tokens	看短任务是否值得流式
中等文本	800–1500 tokens	观察 TTFT 和总耗时
长文本	3000+ tokens	评估用户感知收益
前端逐 token 渲染	不限	观察是否卡顿
前端节流渲染	不限	对比渲染优化效果

如果要发布正式测试数据，建议表格里补充测试日期、模型、地区、SDK 版本、网络环境、重复次数和统计口径。一个可复现的测试表可以这样设计：

模式	任务	TTFT	TTL	Tokens/sec	结论
非流式	短文本	无首字返回	待实测	待实测	实现简单
流式	短文本	待实测	待实测	待实测	体验略好
非流式	长文本	无首字返回	待实测	待实测	等待感明显
流式	长文本	待实测	待实测	待实测	首屏反馈明显更好

实际测下来，通常会得到一个比较稳定的结论：流式对 TTFT 和首屏可读时间帮助最大，但 TTL 往往和非流式接近。

因此，不太建议把 Streaming 宣传成“总耗时一定更短”。更准确的说法应该是：用户能更早看到内容，也能更早决定是否中断。

---

Claude API Streaming 性能优化清单

Claude API 的性能优化，可以从五个层面来做。

1. 降低首字延迟

想让第一个字更快出现，可以从这些地方入手：

• 精简 system prompt；
• 减少无关上下文；
• 根据任务复杂度选择合适模型；
• 避免过长 prefill；
• 后端尽量复用连接，减少冷启动影响。

2. 控制输出长度

输出越长，总耗时和成本通常越高。所以要主动控制生成范围：

• 设置合理的 max_tokens；
• 长文任务给出明确结构；
• 不要让模型无限扩写；
• 支持用户点击停止，并立即 abort 上游请求。

3. 优化前端渲染

前端渲染做不好，流式体验也会被拖垮。建议注意这些点：

• 不要每个 token 都重新渲染整篇 Markdown；
• 使用 30–100ms 的批量刷新；
• 代码块没闭合时，先不要急着做高亮解析；
• 页面上展示“正在生成”，而不是只有一个简单 loading。

4. 提升稳定性

线上环境更复杂，要把异常情况提前考虑进去：

• 正确处理 ping 保活；
• 设置请求超时；
• 捕获断流错误；
• 避免无限重试；
• 使用 request id 做日志追踪。

5. 做好成本监控

Streaming 本身不等于省钱，但它能帮助用户更早中断。成本监控可以这样做：

• 记录 input tokens 和 output tokens；
• 记录用户是否中断；
• 长任务优先使用流式；
• 后台任务可以使用非流式；
• 日志注意脱敏，避免保存敏感内容。

---

常见问题与排错

为什么设置了 stream: true 还是一次性返回？

先检查三个地方：cURL 是否加了 -N，后端是否缓冲响应，Nginx / CDN 是否开启了 proxy buffering。

为什么前端收不到 SSE？

确认响应头是不是 text/event-stream，再检查浏览器请求是否被网关压缩、缓存或合并了。

为什么 Nginx 部署后不再逐字输出？

通常是 proxy_buffering 没关。需要在对应的 location 里设置：

proxy_buffering off;

为什么 Markdown 代码块显示错乱？

流式输出时，代码块可能还没生成完整。建议先按纯文本增量展示，等结束后再做完整 Markdown 渲染。

Tool Use 的 JSON 为什么解析失败？

因为 partial_json 只是 JSON 的一部分，不是完整 JSON。必须缓存到 content_block_stop 之后再解析。

Streaming 会不会更省钱？

不一定。Streaming 不会改变 token 单价，也不会改变模型计费逻辑。它可能带来的成本优势，主要来自用户提前停止生成，从而减少后续输出。

Streaming 和 WebSocket 应该选哪个？

如果只是模型单向输出，优先选 SSE，简单稳定。
如果需要双向实时通信、多人协作，或者复杂状态同步，再考虑 WebSocket。

---

最佳实践：推荐架构和上线 Checklist

推荐架构如下：

浏览器
  ↓
业务后端：鉴权、限流、日志、调用 Claude API、转发 SSE
  ↓
Claude API / 兼容接入平台

上线前建议逐项检查：

• API Key 只放在后端，不进入浏览器；
• 请求体设置了 stream: true；
• 明确指定 model、max_tokens、anthropic-version；
• 只拼接 text_delta，不要把所有事件混在一起；
• Tool Use 的 partial_json 等 block 结束后再解析；
• 后端 SSE 响应头设置正确；
• Nginx / CDN / Serverless 不缓冲响应；
• 支持 AbortController 停止生成；
• 记录完整回答、usage 和 request id；
• 前端 Markdown 渲染做节流；
• 长文本和聊天使用 Streaming，后台批处理使用非流式。

最后总结一句：Claude API Streaming 的配置本身不复杂，真正容易出问题的是事件解析、服务端转发、前端渲染和线上稳定性。

如果只是本地跑一个 demo，stream: true 基本就够了。
但如果要放进真实业务里，就不能只关注能不能输出，还要把 SSE 响应头、代理缓冲、用户中断、Tool Use 拼接、usage 统计和性能指标一起设计好。