新增国内直连 Base URL： https://api.highwayapi.ai/openai，原域名继续提供服务，详见产品文档

Claude API流式输出实现：Server-Sent Events详解

分类：技术交流发布时间：建议阅读时长：58 分钟
作者：sodope llm

引言

在现代AI应用开发中，流式输出（Streaming） 已成为提升用户体验的关键技术。传统的请求-响应模式需要等待模型生成完整内容后才能返回结果，而流式输出允许模型边生成边传输，用户可以实时看到内容逐字出现，极大地降低了感知延迟。

本教程将深入介绍如何使用 Claude API 实现流式输出，涵盖 Server-Sent Events（SSE）协议原理、Anthropic 原生 SDK 用法、OpenAI 兼容接口调用，以及实际 Web 应用构建。

一、什么是流式输出（SSE）

1.1 Server-Sent Events 协议简介

Server-Sent Events（SSE） 是一种基于 HTTP 的服务器推送技术，允许服务器向客户端持续发送数据流。与 WebSocket 不同，SSE 是单向的（服务器 → 客户端），非常适合 AI 文本生成场景。

SSE 消息格式如下：

data: {"type": "content_block_delta", "delta": {"text": "Hello"}}
data: {"type": "content_block_delta", "delta": {"text": " World"}}
data: [DONE]

1.2 SSE 与传统请求的对比

特性传统请求-响应SSE 流式输出
响应延迟等待全部生成完成实时逐步接收
用户体验长时间等待白屏内容实时显示
连接方式短连接持久连接
服务器压力需保持完整内容边生成边释放
适用场景短文本、批处理长文本、对话应用

1.3 Claude API 的流式事件类型

Claude API 在流式模式下会发送以下事件：

  • message_start：消息开始
  • content_block_start：内容块开始
  • content_block_delta：内容增量（实际文本）
  • content_block_stop：内容块结束
  • message_delta：消息元数据更新（如 token 使用量）
  • message_stop：消息结束

二、Anthropic SDK 流式调用

2.1 安装依赖

pip install anthropic

2.2 基础流式调用

import anthropic
client = anthropic.Anthropic(api_key="your-api-key")
# 使用 stream() 上下文管理器
with client.messages.stream(
    model="claude-opus-4-5",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "请写一首关于春天的诗"}
    ]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
print()  # 换行

2.3 获取完整响应信息

import anthropic
client = anthropic.Anthropic(api_key="your-api-key")
with client.messages.stream(
    model="claude-opus-4-5",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "解释量子纠缠原理"}
    ]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    
    # 流结束后获取完整消息对象
    final_message = stream.get_final_message()
    print(f"\n\n--- 统计信息 ---")
    print(f"输入 tokens: {final_message.usage.input_tokens}")
    print(f"输出 tokens: {final_message.usage.output_tokens}")
    print(f"停止原因: {final_message.stop_reason}")

2.4 使用原始事件流（低级接口）

import anthropic
client = anthropic.Anthropic(api_key="your-api-key")
# 使用 stream_raw() 获取原始 SSE 事件
with client.messages.stream(
    model="claude-opus-4-5",
    max_tokens=512,
    messages=[{"role": "user", "content": "你好"}]
) as stream:
    for event in stream:
        print(f"事件类型: {event.type}")
        if hasattr(event, 'delta') and hasattr(event.delta, 'text'):
            print(f"文本增量: {event.delta.text}")

2.5 异步流式调用

import asyncio
import anthropic
async def stream_response():
    client = anthropic.AsyncAnthropic(api_key="your-api-key")
    
    async with client.messages.stream(
        model="claude-opus-4-5",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": "用Python实现快速排序"}
        ]
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)
asyncio.run(stream_response())

三、使用 OpenAI 格式流式调用 jiekou.ai

jiekou.ai 提供了与 OpenAI API 兼容的接口，可以使用 openai Python 库直接调用 Claude 模型。

3.1 安装依赖

pip install openai

3.2 基础流式调用

from openai import OpenAI
client = OpenAI(
    api_key="your-jiekou-ai-api-key",
    base_url="https://api.jiekou.ai/v1"
)
stream = client.chat.completions.create(
    model="claude-opus-4-5",
    messages=[
        {"role": "user", "content": "请介绍Python异步编程"}
    ],
    stream=True,
    max_tokens=1024
)
for chunk in stream:
    if chunk.choices[0].delta.content is not None:
        print(chunk.choices[0].delta.content, end="", flush=True)
print()

3.3 带系统提示的流式调用

from openai import OpenAI
client = OpenAI(
    api_key="your-jiekou-ai-api-key",
    base_url="https://api.jiekou.ai/v1"
)
stream = client.chat.completions.create(
    model="claude-opus-4-5",
    messages=[
        {
            "role": "system",
            "content": "你是一位专业的Python开发专家，回答简洁明了，多用代码示例。"
        },
        {
            "role": "user",
            "content": "如何优化Python代码性能？"
        }
    ],
    stream=True,
    max_tokens=2048,
    temperature=0.7
)
full_response = ""
for chunk in stream:
    delta_content = chunk.choices[0].delta.content
    if delta_content is not None:
        print(delta_content, end="", flush=True)
        full_response += delta_content
print(f"\n\n总字符数: {len(full_response)}")

3.4 多轮对话流式调用

from openai import OpenAI
client = OpenAI(
    api_key="your-jiekou-ai-api-key",
    base_url="https://api.jiekou.ai/v1"
)
def chat_with_stream(messages: list) -> str:
    """流式多轮对话函数"""
    stream = client.chat.completions.create(
        model="claude-opus-4-5",
        messages=messages,
        stream=True,
        max_tokens=1024
    )
    
    full_response = ""
    for chunk in stream:
        content = chunk.choices[0].delta.content
        if content:
            print(content, end="", flush=True)
            full_response += content
    
    print()
    return full_response
# 维护对话历史
conversation = []
while True:
    user_input = input("\n你: ")
    if user_input.lower() in ["exit", "quit", "退出"]:
        break
    
    conversation.append({"role": "user", "content": user_input})
    
    print("Claude: ", end="")
    response = chat_with_stream(conversation)
    
    conversation.append({"role": "assistant", "content": response})

四、构建实时对话 Web 应用

4.1 Flask + SSE 后端

from flask import Flask, Response, request, jsonify, stream_with_context
from openai import OpenAI
import json
app = Flask(__name__)
client = OpenAI(
    api_key="your-api-key",
    base_url="https://api.jiekou.ai/v1"
)
@app.route("/chat/stream", methods=["POST"])
def chat_stream():
    data = request.json
    messages = data.get("messages", [])
    
    def generate():
        try:
            stream = client.chat.completions.create(
                model="claude-opus-4-5",
                messages=messages,
                stream=True,
                max_tokens=2048
            )
            
            for chunk in stream:
                content = chunk.choices[0].delta.content
                if content:
                    # SSE 格式
                    yield f"data: {json.dumps({'content': content})}\n\n"
            
            yield "data: [DONE]\n\n"
            
        except Exception as e:
            yield f"data: {json.dumps({'error': str(e)})}\n\n"
    
    return Response(
        stream_with_context(generate()),
        mimetype="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no"
        }
    )
if __name__ == "__main__":
    app.run(debug=True, port=5000)

4.2 前端 JavaScript 接收 SSE

<!DOCTYPE html>
<html lang="zh">
<head>
    <meta charset="UTF-8">
    <title>Claude 实时对话</title>
    <style>
        #chat-box { height: 400px; overflow-y: auto; border: 1px solid #ccc; padding: 10px; }
        #input-area { display: flex; gap: 10px; margin-top: 10px; }
        #user-input { flex: 1; padding: 8px; }
    </style>
</head>
<body>
    <div id="chat-box"></div>
    <div id="input-area">
        <input id="user-input" type="text" placeholder="输入消息...">
        <button onclick="sendMessage()">发送</button>
    </div>
    <script>
        const messages = [];
        
        async function sendMessage() {
            const input = document.getElementById('user-input');
            const chatBox = document.getElementById('chat-box');
            const userText = input.value.trim();
            
            if (!userText) return;
            
            messages.push({ role: 'user', content: userText });
            chatBox.innerHTML += `<p><strong>你:</strong> ${userText}</p>`;
            input.value = '';
            
            // 创建 AI 回复容器
            const aiPara = document.createElement('p');
            aiPara.innerHTML = '<strong>Claude:</strong> ';
            chatBox.appendChild(aiPara);
            
            const response = await fetch('/chat/stream', {
                method: 'POST',
                headers: { 'Content-Type': 'application/json' },
                body: JSON.stringify({ messages })
            });
            
            const reader = response.body.getReader();
            const decoder = new TextDecoder();
            let aiResponse = '';
            
            while (true) {
                const { done, value } = await reader.read();
                if (done) break;
                
                const chunk = decoder.decode(value);
                const lines = chunk.split('\n');
                
                for (const line of lines) {
                    if (line.startsWith('data: ')) {
                        const data = line.slice(6);
                        if (data === '[DONE]') break;
                        
                        try {
                            const parsed = JSON.parse(data);
                            if (parsed.content) {
                                aiResponse += parsed.content;
                                aiPara.innerHTML = `<strong>Claude:</strong> ${aiResponse}`;
                                chatBox.scrollTop = chatBox.scrollHeight;
                            }
                        } catch (e) {}
                    }
                }
            }
            
            messages.push({ role: 'assistant', content: aiResponse });
        }
        
        document.getElementById('user-input').addEventListener('keydown', (e) => {
            if (e.key === 'Enter') sendMessage();
        });
    </script>
</body>
</html>

4.3 FastAPI 异步版本

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from openai import AsyncOpenAI
import json
app = FastAPI()
client = AsyncOpenAI(
    api_key="your-api-key",
    base_url="https://api.jiekou.ai/v1"
)
class ChatRequest(BaseModel):
    messages: list
    model: str = "claude-opus-4-5"
    max_tokens: int = 2048
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    async def generate():
        async with client.chat.completions.with_streaming_response.create(
            model=request.model,
            messages=request.messages,
            stream=True,
            max_tokens=request.max_tokens
        ) as response:
            async for chunk in response.iter_lines():
                if chunk.startswith("data: "):
                    data = chunk[6:]
                    if data != "[DONE]":
                        yield f"data: {data}\n\n"
        
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(
        generate(),
        media_type="text/event-stream"
    )

五、流式输出的错误处理

5.1 常见错误类型

错误类型原因处理方式
APIConnectionError网络连接失败重试机制
RateLimitError超出速率限制指数退避重试
APIStatusErrorAPI 返回错误状态码记录日志，通知用户
StreamInterruptedError流中断断点续传或重新请求
TimeoutError请求超时设置合理超时时间

5.2 健壮的错误处理实现

import anthropic
import time
import logging
from typing import Generator
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def stream_with_retry(
    client: anthropic.Anthropic,
    messages: list,
    max_retries: int = 3,
    initial_delay: float = 1.0
) -> Generator[str, None, None]:
    """带重试机制的流式输出"""
    
    for attempt in range(max_retries):
        try:
            with client.messages.stream(
                model="claude-opus-4-5",
                max_tokens=1024,
                messages=messages
            ) as stream:
                for text in stream.text_stream:
                    yield text
                return  # 成功完成，退出函数
                
        except anthropic.RateLimitError as e:
            wait_time = initial_delay * (2 ** attempt)
            logger.warning(f"速率限制，{wait_time}秒后重试... (尝试 {attempt + 1}/{max_retries})")
            time.sleep(wait_time)
            
        except anthropic.APIConnectionError as e:
            logger.error(f"连接错误: {e}")
            if attempt == max_retries - 1:
                raise
            time.sleep(initial_delay)
            
        except anthropic.APIStatusError as e:
            logger.error(f"API错误 {e.status_code}: {e.message}")
            raise  # 不重试状态码错误
    
    raise Exception(f"在 {max_retries} 次尝试后仍然失败")
# 使用示例
client = anthropic.Anthropic(api_key="your-api-key")
messages = [{"role": "user", "content": "写一个Python爬虫示例"}]
try:
    for text in stream_with_retry(client, messages):
        print(text, end="", flush=True)
except Exception as e:
    print(f"\n请求失败: {e}")

5.3 超时控制

import asyncio
import anthropic
async def stream_with_timeout(prompt: str, timeout_seconds: int = 30):
    """带超时控制的异步流式输出"""
    client = anthropic.AsyncAnthropic(api_key="your-api-key")
    
    async def _stream():
        result = []
        async with client.messages.stream(
            model="claude-opus-4-5",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}]
        ) as stream:
            async for text in stream.text_stream:
                print(text, end="", flush=True)
                result.append(text)
        return "".join(result)
    
    try:
        return await asyncio.wait_for(_stream(), timeout=timeout_seconds)
    except asyncio.TimeoutError:
        print(f"\n[请求超时，已等待 {timeout_seconds} 秒]")
        return None
# 运行
result = asyncio.run(stream_with_timeout("解释机器学习的基本原理", timeout_seconds=60))

六、性能对比与最佳实践

6.1 流式 vs 非流式性能对比

指标非流式流式
首字节时间（TTFB）5-30 秒< 1 秒
用户感知延迟高低
服务器内存占用需缓存完整响应边传边释放
适合输出长度短文本（< 200 tokens）长文本（> 200 tokens）
实现复杂度低中等
错误处理难度简单需要处理中断

6.2 Token 计算对比

import anthropic
import time
client = anthropic.Anthropic(api_key="your-api-key")
prompt = "请详细介绍Python的异步编程模型，包括asyncio、协程、事件循环等概念"
# 非流式
start = time.time()
response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": prompt}]
)
non_stream_time = time.time() - start
print(f"非流式总耗时: {non_stream_time:.2f}s")
print(f"输出tokens: {response.usage.output_tokens}")
# 流式（测量首个token时间）
start = time.time()
first_token_time = None
total_tokens = 0
with client.messages.stream(
    model="claude-opus-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": prompt}]
) as stream:
    for text in stream.text_stream:
        if first_token_time is None:
            first_token_time = time.time() - start
        total_tokens += 1
stream_total_time = time.time() - start
print(f"\n流式首个token时间: {first_token_time:.2f}s")
print(f"流式总耗时: {stream_total_time:.2f}s")

6.3 最佳实践总结

✅ 推荐做法

  1. 长文本生成必用流式：超过 200 tokens 的响应建议使用流式输出
  2. 设置合理的 max_tokens：避免无限制生成，控制成本
  3. 实现重试机制：针对网络错误和速率限制使用指数退避
  4. 使用异步客户端：在 Web 服务中使用 AsyncAnthropic 提升并发性能
  5. 前端实时更新：使用 EventSource 或 fetch + ReadableStream 处理流数据
  6. 监控 token 使用：记录每次请求的 token 消耗，控制成本

❌ 避免的做法

  1. 不要在流中缓存全部内容再显示：这失去了流式的意义
  2. 不要忽略错误处理：流中断会导致用户看到不完整内容
  3. 不要设置过短的超时：长文本生成可能需要数十秒
  4. 不要在紧密循环中轮询：使用事件驱动模型替代轮询
# ✅ 正确：边接收边处理
with client.messages.stream(...) as stream:
    for text in stream.text_stream:
        update_ui(text)  # 立即更新UI
# ❌ 错误：积累后再处理
chunks = []
with client.messages.stream(...) as stream:
    for text in stream.text_stream:
        chunks.append(text)  # 失去了流式优势
update_ui("".join(chunks))

总结

本教程详细介绍了 Claude API 流式输出的完整实现方案：

场景推荐方案
快速原型开发Anthropic SDK + stream() 上下文管理器
兼容 OpenAI 生态openai 库 + jiekou.ai base_url
Web 后端服务FastAPI/Flask + SSE 推送
高并发场景AsyncAnthropic + 异步处理
生产环境完善的重试机制 + 监控告警

流式输出是构建优质 AI 应用的基础能力。掌握 SSE 协议、正确使用 SDK 流式接口、构建健壮的错误处理机制，将帮助你打造响应迅速、用户体验卓越的 AI 产品。

下一步建议：结合 WebSocket 实现双向实时通信，或探索 Claude 的工具调用（Tool Use）功能，构建更复杂的 AI Agent 应用。

分享：
联系我们