在现代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") 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 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" ) 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, 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: 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 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" )

---

五、流式输出的错误处理

常见错误类型

• APIConnectionError：网络连接失败 → 使用重试机制

• RateLimitError：超出速率限制 → 指数退避重试

• APIStatusError：API返回错误状态码 → 记录日志，通知用户

• TimeoutError：请求超时 → 设置合理超时时间

健壮的错误处理实现

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}")

---

六、最佳实践总结

推荐做法：

• 长文本生成必用流式：超过200 tokens的响应建议使用流式输出

• 设置合理的max_tokens：避免无限制生成，控制成本

• 实现重试机制：针对网络错误和速率限制使用指数退避

• 使用异步客户端：在Web服务中使用AsyncAnthropic提升并发性能

• 前端实时更新：使用EventSource或fetch + ReadableStream处理流数据

• 监控token使用：记录每次请求的token消耗，控制成本

避免的做法：

• 不要在流中缓存全部内容再显示：这失去了流式的意义

• 不要忽略错误处理：流中断会导致用户看到不完整内容

• 不要设置过短的超时：长文本生成可能需要数十秒

---

总结

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

• 快速原型开发：Anthropic SDK + stream() 上下文管理器

• 兼容OpenAI生态：openai库 + jiekou.ai base_url

• Web后端服务：FastAPI/Flask + SSE推送

• 高并发场景：AsyncAnthropic + 异步处理

• 生产环境：完善的重试机制 + 监控告警

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