Claude 流式 API 使用教程

Claude 流式 API（Streaming）允许服务端在生成响应的同时逐步推送数据，大幅减少用户等待时间。本文详细介绍 Claude SSE 流式协议、Python 和 Node.js 实现，以及错误处理最佳实践。

什么是流式输出？

传统 API 调用需要等待模型生成完整响应后一次性返回，而流式输出（Streaming）使用 Server-Sent Events（SSE）协议，在模型生成每个 token 时即时推送给客户端。对于长文本生成场景，流式输出可以将首字节响应时间从几十秒缩短到几百毫秒。

SSE 协议基础

Claude 流式 API 基于标准 SSE 协议，事件格式如下：

event: message_start
data: {"type":"message_start","message":{...}}

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}

event: message_stop
data: {"type":"message_stop"}

主要事件类型包括：

• message_start：消息开始，包含模型信息和 usage
• content_block_start：内容块开始
• content_block_delta：增量文本内容
• content_block_stop：内容块结束
• message_delta：消息级别的增量信息（如 stop_reason、output token 数）
• message_stop：消息结束

Python 流式调用

import anthropic

client = anthropic.Anthropic(
    api_key="your-api-key",
    base_url="https://claude4u.com/v1"  # 轻舟 AI 中转地址
)

# 方式一：使用 stream 上下文管理器（推荐）
with client.messages.stream(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "用 Python 写一个快速排序"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

# 获取最终消息（包含 usage 信息）
final_message = stream.get_final_message()
print(f"\n总 token：{final_message.usage.input_tokens + final_message.usage.output_tokens}")

方式二：原始 SSE 事件

# 如果需要处理每个 SSE 事件
with client.messages.stream(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}]
) as stream:
    for event in stream:
        if event.type == "content_block_delta":
            print(event.delta.text, end="")
        elif event.type == "message_delta":
            print(f"\nStop reason: {event.delta.stop_reason}")

Node.js 流式调用

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

const client = new Anthropic({
  apiKey: 'your-api-key',
  baseURL: 'https://claude4u.com/v1'
});

// 方式一：使用 stream 辅助方法
const stream = client.messages.stream({
  model: 'claude-sonnet-4-20250514',
  max_tokens: 1024,
  messages: [{ role: 'user', content: '用 JavaScript 写一个防抖函数' }]
});

stream.on('text', (text) => {
  process.stdout.write(text);
});

const finalMessage = await stream.finalMessage();
console.log('\nUsage:', finalMessage.usage);

// 方式二：for-await 异步迭代
const response = await client.messages.create({
  model: 'claude-sonnet-4-20250514',
  max_tokens: 1024,
  stream: true,
  messages: [{ role: 'user', content: 'Hello' }]
});

for await (const event of response) {
  if (event.type === 'content_block_delta') {
    process.stdout.write(event.delta.text);
  }
}

错误处理与重连

流式连接可能因网络波动中断，需要做好异常处理：

import anthropic
import time

def stream_with_retry(messages, max_retries=3):
    client = anthropic.Anthropic(
        api_key="your-api-key",
        base_url="https://claude4u.com/v1"
    )
    for attempt in range(max_retries):
        try:
            with client.messages.stream(
                model="claude-sonnet-4-20250514",
                max_tokens=1024,
                messages=messages
            ) as stream:
                collected = []
                for text in stream.text_stream:
                    collected.append(text)
                    print(text, end="", flush=True)
                return "".join(collected)
        except anthropic.APIConnectionError:
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)
                continue
            raise
        except anthropic.RateLimitError:
            time.sleep(10)
            continue

流式输出注意事项

• 流式模式下 usage 信息在 message_start 和 message_delta 事件中分别返回输入和输出 token 数
• 客户端断开连接后应及时关闭流，避免资源浪费
• 流式响应的总 token 消耗与非流式完全一致，不会额外计费
• 建议设置合理的超时时间，避免长时间等待无响应的连接