流式输出与细粒度工具流

用户不喜欢等待。流式输出让 AI 的回复像打字一样逐字出现，大幅提升用户体验。本文讲解 Claude API 的 SSE 流式输出、工具调用的流式传输，以及前端渐进式渲染的完整方案。

你将学到什么

Claude 流式输出的工作原理（SSE）
Python 和 TypeScript 中的流式实现
工具调用的细粒度流式传输
前端渐进式渲染方案

基础流式输出

Python 实现

import anthropic

client = anthropic.Anthropic()

# 流式输出
with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=2048,
    messages=[{"role": "user", "content": "讲一个关于 AI 的故事"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

print()  # 换行

TypeScript 实现

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

const client = new Anthropic();

async function streamResponse() {
  const stream = client.messages.stream({
    model: "claude-sonnet-4-6",
    max_tokens: 2048,
    messages: [{ role: "user", content: "讲一个关于 AI 的故事" }],
  });

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

  const finalMessage = await stream.finalMessage();
  console.log("\nTokens used:", finalMessage.usage);
}

SSE 事件类型

Claude 流式输出基于 Server-Sent Events（SSE），包含以下事件：

事件	说明
`message_start`	消息开始，包含 metadata
`content_block_start`	内容块开始（text 或 tool_use）
`content_block_delta`	内容增量（逐字输出）
`content_block_stop`	内容块结束
`message_delta`	消息级别更新（stop_reason、usage）
`message_stop`	消息结束

工具调用的流式传输

当 Claude 调用工具时，工具的参数也会流式传输。这意味着你可以：

在参数还没完全生成时就开始准备执行
向用户展示 "正在调用 xxx 工具..."

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=4096,
    tools=tools,
    messages=[{"role": "user", "content": "搜索最新的 AI 新闻"}]
) as stream:
    for event in stream:
        if event.type == "content_block_start":
            if event.content_block.type == "tool_use":
                print(f"\n正在调用工具: {event.content_block.name}")

        elif event.type == "content_block_delta":
            if event.delta.type == "text_delta":
                print(event.delta.text, end="", flush=True)
            elif event.delta.type == "input_json_delta":
                # 工具参数的增量 JSON
                print(f"  参数片段: {event.delta.partial_json}")

        elif event.type == "message_stop":
            print("\n流结束")

前端渐进式渲染

Next.js + AI SDK

// app/api/chat/route.ts
import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();

export async function POST(req: Request) {
  const { messages } = await req.json();

  const stream = client.messages.stream({
    model: "claude-sonnet-4-6",
    max_tokens: 2048,
    messages,
  });

  // 将 SDK stream 转为 ReadableStream
  const readableStream = new ReadableStream({
    async start(controller) {
      for await (const event of stream) {
        if (
          event.type === "content_block_delta" &&
          event.delta.type === "text_delta"
        ) {
          controller.enqueue(
            new TextEncoder().encode(event.delta.text)
          );
        }
      }
      controller.close();
    },
  });

  return new Response(readableStream, {
    headers: { "Content-Type": "text/plain; charset=utf-8" },
  });
}

前端消费

async function chat(message: string) {
  const response = await fetch("/api/chat", {
    method: "POST",
    body: JSON.stringify({ messages: [{ role: "user", content: message }] }),
  });

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

  while (reader) {
    const { done, value } = await reader.read();
    if (done) break;
    const chunk = decoder.decode(value);
    // 将 chunk 追加到 UI 中
    appendToChat(chunk);
  }
}

实战练习

Tip: 实现一个流式聊天界面。

用 Python 实现基础流式输出
在 Next.js 中实现 SSE 接口
前端实现打字机效果的渐进式渲染

关键要点

Note: 本文核心总结

流式输出基于 SSE，逐字/逐块传输
client.messages.stream() 是最简单的流式 API
工具调用参数也支持细粒度流式传输
前端用 ReadableStream 消费，实现打字机效果