SSE 流式输出前端接收与断线重连实战：基于 HolySheep AI API 的完整指南

当 DeepSeek V3.2 仅需 $0.42/MTok、Gemini 2.5 Flash 为 $2.50/MTok、Claude Sonnet 4.5 高达 $15/MTok、GPT-4.1 为 $8/MTok 时，你以为选最便宜的就完事了？错！真正的成本杀手藏在结算汇率里。

以每月 100 万 Token 输出量为例，按官方美元汇率 ¥7.3=$1 计算：DeepSeek 需 ¥306.6，而 HolySheep 按立即注册专享的 ¥1=$1 无损汇率，仅需 ¥42——节省超过 85%。这才是中转站的核心价值：不仅聚合多模型，更用企业级汇率让你的人民币花出美元效果。

一、SSE 流式输出的技术原理

Server-Sent Events（SSE）是一种基于 HTTP 的单向下推技术，服务器通过 text/event-stream MIME 类型持续推送数据。与 WebSocket 不同，SSE 轻量、无协议握手，特别适合 AI 补全、进度展示、日志推送等场景。HolySheep AI API 完全兼容 OpenAI 格式的流式响应，前端接收逻辑完全一致。

二、前端实现：EventSource 与 Fetch 流式读取

现代浏览器环境下，推荐使用 Fetch API 配合 ReadableStream 处理 SSE，比传统的 EventSource 更灵活（支持 POST 请求、中断流式响应）。

2.1 标准 Fetch 流式实现

<!DOCTYPE html>
<html lang="zh-CN">
<head>
  <meta charset="UTF-8">
  <title>SSE 流式对话演示</title>
  <style>
    body { font-family: -apple-system, sans-serif; max-width: 800px; margin: 40px auto; padding: 0 20px; }
    #chat { border: 1px solid #e0e0e0; border-radius: 8px; padding: 16px; min-height: 300px; margin-bottom: 16px; }
    .message { margin: 8px 0; padding: 8px 12px; border-radius: 8px; }
    .user { background: #007AFF; color: white; margin-left: 20%; }
    .assistant { background: #F2F2F7; margin-right: 20%; }
    #input-area { display: flex; gap: 12px; }
    input { flex: 1; padding: 12px; border: 1px solid #ddd; border-radius: 8px; font-size: 16px; }
    button { padding: 12px 24px; background: #007AFF; color: white; border: none; border-radius: 8px; cursor: pointer; }
    button:disabled { background: #ccc; }
  </style>
</head>
<body>
  <h2>🟢 HolySheep AI 流式对话示例</h2>
  <div id="chat"></div>
  <div id="input-area">
    <input type="text" id="userInput" placeholder="输入你的问题..." value="请用50字介绍量子计算">
    <button id="sendBtn" onclick="sendMessage()">发送</button>
  </div>

  <script>
    const HOLYSHEEP_BASE_URL = 'https://api.holysheep.ai/v1';
    // ⚠️ 替换为你的 HolySheep API Key
    const API_KEY = 'YOUR_HOLYSHEEP_API_KEY';

    let abortController = null;
    let isStreaming = false;

    async function sendMessage() {
      const input = document.getElementById('userInput');
      const sendBtn = document.getElementById('sendBtn');
      const chat = document.getElementById('chat');
      const message = input.value.trim();
      
      if (!message || isStreaming) return;

      // 追加用户消息
      const userDiv = document.createElement('div');
      userDiv.className = 'message user';
      userDiv.textContent = message;
      chat.appendChild(userDiv);

      // 创建助手消息气泡（用于流式追加内容）
      const assistantDiv = document.createElement('div');
      assistantDiv.className = 'message assistant';
      assistantDiv.id = 'current-assistant';
      chat.appendChild(assistantDiv);

      input.value = '';
      sendBtn.disabled = true;
      isStreaming = true;
      let fullResponse = '';

      try {
        // 创建 AbortController 用于中断请求
        abortController = new AbortController();

        const response = await fetch(${HOLYSHEEP_BASE_URL}/chat/completions, {
          method: 'POST',
          headers: {
            'Content-Type': 'application/json',
            'Authorization': Bearer ${API_KEY}
          },
          body: JSON.stringify({
            model: 'deepseek-chat',  // 支持 deepseek/chatgpt/claude/gemini 等
            messages: [{ role: 'user', content: message }],
            stream: true  // 关键：开启流式输出
          }),
          signal: abortController.signal
        });

        if (!response.ok) {
          throw new Error(HTTP ${response.status}: ${response.statusText});
        }

        // 方式一：使用 ReadableStream（推荐，现代浏览器）
        const reader = response.body.getReader();
        const decoder = new TextDecoder();

        while (true) {
          const { done, value } = await reader.read();
          if (done) break;

          const chunk = decoder.decode(value, { stream: true });
          // 解析 SSE 格式：data: {...}\n\n
          const lines = chunk.split('\n');
          
          for (const line of lines) {
            if (line.startsWith('data: ')) {
              const data = line.slice(6);
              if (data === '[DONE]') continue;
              
              try {
                const parsed = JSON.parse(data);
                const content = parsed.choices?.[0]?.delta?.content;
                if (content) {
                  fullResponse += content;
                  assistantDiv.textContent = fullResponse;
                  // 实时滚动到底部
                  chat.scrollTop = chat.scrollHeight;
                }
              } catch (e) {
                // 忽略解析错误（可能是不完整的 JSON）
              }
            }
          }
        }

      } catch (error) {
        if (error.name === 'AbortError') {
          assistantDiv.textContent += '\n[已中断]';
        } else {
          assistantDiv.textContent = ❌ 请求失败: ${error.message};
          console.error('Stream error:', error);
        }
      } finally {
        isStreaming = false;
        sendBtn.disabled = false;
        abortController = null;
      }
    }

    // 支持 Ctrl+Enter 发送
    document.getElementById('userInput').addEventListener('keydown', (e) => {
      if (e.key === 'Enter' && !e.shiftKey) {
        e.preventDefault();
        sendMessage();
      }
    });
  </script>
</body>
</html>

2.2 企业级封装：自动重连与错误恢复

/**
 * HolySheep 流式 API 客户端
 * 支持：自动重连、错误恢复、连接状态监听、并发限制
 */

class HolySheepStreamClient {
  constructor(options = {}) {
    this.baseUrl = options.baseUrl || 'https://api.holysheep.ai/v1';
    this.apiKey = options.apiKey || '';
    this.model = options.model || 'deepseek-chat';
    this.maxRetries = options.maxRetries || 3;
    this.retryDelay = options.retryDelay || 1000; // ms
    this.reconnectInterval = options.reconnectInterval || 3000; // ms
    this.maxReconnectAttempts = options.maxReconnectAttempts || 5;
    
    // 连接状态
    this.isConnected = false;
    this.isReconnecting = false;
    this.currentAbortController = null;
    this.reconnectAttempts = 0;
    
    // 事件监听器
    this.listeners = {
      token: [],      // 每个 token 回调
      done: [],       // 完成回调
      error: [],      // 错误回调
      reconnecting: [], // 重连中回调
      connected: []   // 连接成功回调
    };
  }

  // 事件系统
  on(event, callback) {
    if (this.listeners[event]) {
      this.listeners[event].push(callback);
    }
    return this;
  }

  off(event, callback) {
    if (this.listeners[event]) {
      this.listeners[event] = this.listeners[event].filter(cb => cb !== callback);
    }
    return this;
  }

  emit(event, data) {
    if (this.listeners[event]) {
      this.listeners[event].forEach(cb => {
        try {
          cb(data);
        } catch (e) {
          console.error(Event handler error for ${event}:, e);
        }
      });
    }
  }

  // 核心：发送流式请求
  async send(messages, options = {}) {
    const model = options.model || this.model;
    const temperature = options.temperature ?? 0.7;
    const maxTokens = options.maxTokens || 4096;

    // 清理之前的连接
    this.abort();

    return new Promise((resolve, reject) => {
      const streamHandler = async (retryCount = 0) => {
        this.currentAbortController = new AbortController();
        let fullContent = '';
        let finishReason = null;

        try {
          const response = await fetch(${this.baseUrl}/chat/completions, {
            method: 'POST',
            headers: {
              'Content-Type': 'application/json',
              'Authorization': Bearer ${this.apiKey}
            },
            body: JSON.stringify({
              model,
              messages,
              stream: true,
              temperature,
              max_tokens: maxTokens,
              ...options.extraParams
            }),
            signal: this.currentAbortController.signal
          });

          if (!response.ok) {
            const errorText = await response.text();
            throw new Error(HTTP ${response.status}: ${errorText});
          }

          this.isConnected = true;
          this.reconnectAttempts = 0;
          this.emit('connected', { model, timestamp: Date.now() });

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

          while (true) {
            const { done, value } = await reader.read();
            if (done) break;

            const chunk = decoder.decode(value, { stream: true });
            const lines = chunk.split('\n');

            for (const line of lines) {
              if (!line.startsWith('data: ')) continue;
              
              const data = line.slice(6);
              if (data === '[DONE]') {
                finishReason = 'stop';
                continue;
              }

              try {
                const parsed = JSON.parse(data);
                const delta = parsed.choices?.[0]?.delta;
                
                if (delta?.content) {
                  fullContent += delta.content;
                  this.emit('token', { 
                    token: delta.content, 
                    fullContent,
                    usage: parsed.usage 
                  });
                }

                if (parsed.choices?.[0]?.finish_reason) {
                  finishReason = parsed.choices[0].finish_reason;
                }
              } catch (e) {
                // 忽略不完整 JSON
              }
            }
          }

          this.isConnected = false;
          this.emit('done', { 
            content: fullContent, 
            finishReason,
            retryCount 
          });
          resolve({ content: fullContent, finishReason });

        } catch (error) {
          this.isConnected = false;

          if (error.name === 'AbortError') {
            this.emit('done', { content: fullContent, finishReason: 'abort' });
            resolve({ content: fullContent, finishReason: 'abort' });
            return;
          }

          console.error(Stream error (attempt ${retryCount + 1}):, error);

          // 重试逻辑
          if (retryCount < this.maxRetries) {
            const delay = this.retryDelay * Math.pow(2, retryCount); // 指数退避
            console.log(Retrying in ${delay}ms...);
            
            await new Promise(r => setTimeout(r, delay));
            return streamHandler(retryCount + 1);
          }

          this.emit('error', { error: error.message, retryCount });
          reject(error);
        }
      };

      streamHandler(0);
    });
  }

  // 中断当前请求
  abort() {
    if (this.currentAbortController) {
      this.currentAbortController.abort();
      this.currentAbortController = null;
    }
  }

  // 主动断线重连
  async reconnect() {
    if (this.isReconnecting) return;
    
    this.isReconnecting = true;
    this.abort();

    while (this.reconnectAttempts < this.maxReconnectAttempts) {
      this.reconnectAttempts++;
      this.emit('reconnecting', { 
        attempt: this.reconnectAttempts,
        maxAttempts: this.maxReconnectAttempts 
      });

      try {
        // 测试连接
        const testResponse = await fetch(${this.baseUrl}/models, {
          headers: { 'Authorization': Bearer ${this.apiKey} }
        });
        
        if (testResponse.ok) {
          this.isReconnecting = false;
          this.reconnectAttempts = 0;
          return true;
        }
      } catch (e) {
        console.warn(Reconnect attempt ${this.reconnectAttempts} failed);
      }

      await new Promise(r => setTimeout(r, this.reconnectInterval));
    }

    this.isReconnecting = false;
    throw new Error(Failed to reconnect after ${this.maxReconnectAttempts} attempts);
  }

  // 销毁实例
  destroy() {
    this.abort();
    this.listeners = { token: [], done: [], error: [], reconnecting: [], connected: [] };
  }
}

// ============ 使用示例 ============

const client = new HolySheepStreamClient({
  apiKey: 'YOUR_HOLYSHEEP_API_KEY',
  model: 'deepseek-chat',
  maxRetries: 3,
  retryDelay: 1000
});

// 监听每个 token
client.on('token', ({ token, fullContent }) => {
  process.stdout.write(token);  // 实时输出
});

// 监听完成
client.on('done', ({ content, finishReason }) => {
  console.log('\n\n--- 完整响应 ---');
  console.log(content);
  console.log('--- 结束 ---');
});

// 监听错误
client.on('error', ({ error, retryCount }) => {
  console.error(请求失败 (已重试 ${retryCount} 次): ${error});
});

// 监听重连
client.on('reconnecting', ({ attempt, maxAttempts }) => {
  console.log(🔄 正在第 ${attempt}/${maxAttempts} 次重连...);
});

// 发送请求
(async () => {
  try {
    await client.send([
      { role: 'system', content: '你是一个有帮助的AI助手。' },
      { role: 'user', content: '解释一下什么是RESTful API' }
    ]);
  } catch (e) {
    console.error('最终错误:', e.message);
  }
})();

三、断线重连的工程实践

生产环境中，网络波动、服务器重启、超时中断都会导致流式连接中断。一个健壮的重连机制需要考虑：

指数退避：重试间隔从 1s → 2s → 4s → 8s，避免对服务器造成压力
最大重试次数：防止无限循环，建议 3-5 次
连接心跳：定期 ping/pong 检测连接存活状态
UI 状态同步：让用户知道当前是「连接中」还是「重连中」

四、Vue/React 集成示例

// React Hook 封装
import { useState, useRef, useCallback, useEffect } from 'react';

export function useHolySheepStream(apiKey) {
  const [messages, setMessages] = useState([]);
  const [isStreaming, setIsStreaming] = useState(false);
  const [error, setError] = useState(null);
  const abortControllerRef = useRef(null);

  const sendMessage = useCallback(async (content) => {
    if (!apiKey || isStreaming) return;

    // 添加用户消息
    const userMsg = { role: 'user', content, id
相关资源
📚 AI API 技术文章库
💰 查看价格
📖 开发者文档
🚀 免费注册
相关文章

一、SSE 流式输出的技术原理

二、前端实现：EventSource 与 Fetch 流式读取

2.1 标准 Fetch 流式实现

2.2 企业级封装：自动重连与错误恢复

三、断线重连的工程实践

四、Vue/React 集成示例

相关资源

相关文章

🔥 推荐使用 HolySheep AI