Claude API 中转站

主题

Claude API 流式输出完整指南：提升用户体验的实时响应方案 ​

先说结论：

Claude API 是 Anthropic 提供的大模型接口，尤其适合长文本处理、严谨写作、复杂代码分析和高质量内容生成。
对于国内开发者来说，实际使用时最常见的问题并不是"Claude 好不好"，而是"怎么接、怎么调、怎么选版本"。

国内最推荐 Claude API 中转站平台：

Claude API 中转站 平台地址：https://jeniya.cn

Claude API 中转站 平台地址：https://jeniya.top

Claude API 中转站 平台地址：https://jeniya.chat

在使用 Claude API 时，流式输出（Streaming） 是提升用户体验的关键技术。这篇文章详细讲解：

  • 什么是流式输出？为什么需要它？
  • Claude API 流式调用如何实现？
  • Python SDK 流式输出完整示例
  • 错误处理和重连机制
  • 性能优化建议

什么是流式输出 ​

传统请求 vs 流式输出 ​

传统请求模式：

用户发送请求 → 等待... → 一次性返回完整结果

流式输出模式：

用户发送请求 → 逐字返回 → 实时显示 → 完成

为什么需要流式输出 ​

场景传统模式流式输出
长文本生成用户等待时间长，体验差实时看到内容，体验好
交互式对话响应延迟明显对话更流畅自然
调试开发难以判断进度可以实时观察输出
用户感知系统好像"卡住了"系统正在"思考"

核心价值：流式输出让用户感知到系统正在工作，而不是"卡死"。

Claude API 流式调用基础 ​

基本原理 ​

Claude API 的流式输出基于 SSE（Server-Sent Events） 技术：

  1. 客户端发送请求，设置 stream=True
  2. 服务端返回事件流
  3. 客户端逐个接收事件并处理
  4. 流结束或出错时关闭连接

关键参数 ​

python
response = client.messages.create(
    model="claude-sonnet-4-6",
    messages=[{"role": "user", "content": "写一首诗"}],
    stream=True  # 开启流式输出
)

Python SDK 流式输出示例 ​

基础示例 ​

python
import os
from anthropic import Anthropic

def stream_chat(prompt: str):
    """基础流式输出示例"""
    client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
    
    with client.messages.stream(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
    
    print()  # 换行

if __name__ == "__main__":
    stream_chat("请写一首关于编程的五言绝句")

使用 OpenAI SDK 格式（中转站） ​

如果你使用兼容 OpenAI 格式的中转站：

python
from openai import OpenAI

def stream_chat_openai(prompt: str):
    """使用 OpenAI SDK 格式的流式输出"""
    client = OpenAI(
        api_key="YOUR_API_KEY",
        base_url="https://jeniya.cn/v1"
    )
    
    stream = client.chat.completions.create(
        model="claude-sonnet-4-6",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )
    
    for chunk in stream:
        if chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)
    
    print()

if __name__ == "__main__":
    stream_chat_openai("请解释什么是机器学习")

进阶：完整的流式输出封装 ​

下面是一个生产环境可用的流式输出封装：

python
import os
import time
from anthropic import Anthropic
from typing import Generator, Optional

class ClaudeStreamer:
    """Claude API 流式输出封装类"""
    
    def __init__(self, api_key: Optional[str] = None):
        self.client = Anthropic(
            api_key=api_key or os.getenv("ANTHROPIC_API_KEY")
        )
    
    def stream(
        self,
        prompt: str,
        model: str = "claude-sonnet-4-6",
        max_tokens: int = 2048,
        temperature: float = 0.7
    ) -> Generator[str, None, None]:
        """
        流式生成文本
        
        Args:
            prompt: 用户输入
            model: 模型名称
            max_tokens: 最大输出长度
            temperature: 随机性参数
            
        Yields:
            逐个生成的文本片段
        """
        with self.client.messages.stream(
            model=model,
            max_tokens=max_tokens,
            temperature=temperature,
            messages=[{"role": "user", "content": prompt}]
        ) as stream:
            for text in stream.text_stream:
                yield text
    
    def stream_with_callback(
        self,
        prompt: str,
        callback: callable,
        **kwargs
    ) -> str:
        """
        带回调的流式输出
        
        Args:
            prompt: 用户输入
            callback: 每收到一段文本就调用的函数
            **kwargs: 传递给 stream() 的其他参数
            
        Returns:
            完整的生成文本
        """
        full_text = ""
        for chunk in self.stream(prompt, **kwargs):
            full_text += chunk
            if callback:
                callback(chunk)
        return full_text
    
    def stream_to_file(
        self,
        prompt: str,
        output_path: str,
        **kwargs
    ) -> str:
        """
        流式输出到文件
        
        Args:
            prompt: 用户输入
            output_path: 输出文件路径
            **kwargs: 传递给 stream() 的其他参数
            
        Returns:
            完整的生成文本
        """
        full_text = ""
        with open(output_path, 'w', encoding='utf-8') as f:
            for chunk in self.stream(prompt, **kwargs):
                full_text += chunk
                f.write(chunk)
                f.flush()  # 实时写入
        return full_text


# 使用示例
def main():
    streamer = ClaudeStreamer()
    
    # 示例1：直接迭代
    print("=== 直接迭代 ===")
    for chunk in streamer.stream("用三句话介绍 Python"):
        print(chunk, end="", flush=True)
    print("\n")
    
    # 示例2：带回调函数
    print("=== 带回调函数 ===")
    def on_chunk(chunk):
        # 可以在这里做额外处理，比如更新 UI
        pass
    
    result = streamer.stream_with_callback(
        "解释什么是 API",
        callback=on_chunk
    )
    print(f"完整结果: {result}\n")
    
    # 示例3：输出到文件
    print("=== 输出到文件 ===")
    streamer.stream_to_file(
        "写一篇关于人工智能的短文",
        "output.txt"
    )
    print("已保存到 output.txt")

if __name__ == "__main__":
    main()

错误处理与重连机制 ​

流式输出过程中可能遇到各种错误，需要妥善处理：

python
import os
import time
from anthropic import Anthropic, APIError, APIConnectionError, RateLimitError

def robust_stream(prompt: str, max_retries: int = 3):
    """带错误处理和重试的流式输出"""
    client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
    
    for attempt in range(max_retries):
        try:
            with client.messages.stream(
                model="claude-sonnet-4-6",
                max_tokens=1024,
                messages=[{"role": "user", "content": prompt}]
            ) as stream:
                for text in stream.text_stream:
                    yield text
            break  # 成功完成，退出重试循环
            
        except RateLimitError:
            # 限流错误，等待后重试
            wait_time = 2 ** attempt  # 指数退避
            print(f"\n触发限流，{wait_time}秒后重试...")
            time.sleep(wait_time)
            continue
            
        except APIConnectionError:
            # 连接错误
            print(f"\n连接失败，正在重试 ({attempt + 1}/{max_retries})...")
            time.sleep(1)
            continue
            
        except APIError as e:
            # 其他 API 错误
            print(f"\nAPI 错误: {e}")
            if attempt < max_retries - 1:
                time.sleep(1)
                continue
            raise

# 使用示例
if __name__ == "__main__":
    for chunk in robust_stream("解释什么是深度学习"):
        print(chunk, end="", flush=True)

Web 应用中的流式输出 ​

Flask 示例 ​

python
from flask import Flask, Response, request, stream_with_context
from anthropic import Anthropic
import os

app = Flask(__name__)
client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

@app.route('/stream', methods=['POST'])
def stream():
    """流式输出 API 端点"""
    data = request.json
    prompt = data.get('prompt', '')
    
    def generate():
        with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}]
        ) as stream:
            for text in stream.text_stream:
                yield f"data: {text}\n\n"
        yield "data: [DONE]\n\n"
    
    return Response(
        stream_with_context(generate()),
        mimetype='text/event-stream'
    )

if __name__ == '__main__':
    app.run(debug=True)

FastAPI 示例 ​

python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from anthropic import Anthropic
import os

app = FastAPI()
client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

@app.post("/stream")
async def stream_chat(prompt: str):
    """流式输出 API 端点"""
    
    async def generate():
        with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}]
        ) as stream:
            for text in stream.text_stream:
                yield f"data: {text}\n\n"
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(
        generate(),
        media_type="text/event-stream"
    )

性能优化建议 ​

1. 合理设置 max_tokens ​

python
# 不推荐：设置过小，容易截断
max_tokens=100

# 推荐：根据实际需求设置
max_tokens=1024  # 一般对话
max_tokens=4096  # 长文本生成

2. 缓冲区管理 ​

python
class BufferManager:
    """流式输出缓冲区管理"""
    
    def __init__(self, buffer_size: int = 10):
        self.buffer_size = buffer_size
        self.buffer = []
    
    def add(self, chunk: str) -> str:
        """添加内容到缓冲区，达到阈值时返回"""
        self.buffer.append(chunk)
        if len(self.buffer) >= self.buffer_size:
            return self.flush()
        return ""
    
    def flush(self) -> str:
        """清空缓冲区并返回内容"""
        result = "".join(self.buffer)
        self.buffer = []
        return result

3. 超时设置 ​

python
from anthropic import Anthropic

client = Anthropic(
    api_key=os.getenv("ANTHROPIC_API_KEY"),
    timeout=60.0  # 设置60秒超时
)

常见问题 ​

1. 流式输出中途断开怎么办？ ​

添加心跳检测和自动重连机制，保存已接收的内容，断开后从断点继续。

2. 如何判断流式输出结束？ ​

  • 正常结束：收到 [DONE] 事件或流自然关闭
  • 异常结束：捕获异常并处理

3. 流式输出比普通请求慢吗？ ​

实际上流式输出的首字响应时间更快，总时间基本相同。用户感知上会觉得流式更快。

总结 ​

流式输出的核心价值：

  • 提升用户体验，减少等待焦虑
  • 适合长文本生成场景
  • 便于实时展示和调试

实现要点：

  • 使用 stream=True 开启流式输出
  • 使用上下文管理器确保资源释放
  • 做好错误处理和重试机制
  • 在 Web 应用中正确处理 SSE 事件

相关阅读 ​