ChatGPT Streaming中断问题解析：如何稳定接收完整消息流

1次阅读

没有评论

共计 2336 个字符，预计需要花费 6 分钟才能阅读完成。

在集成 ChatGPT 流式 API 时，开发者经常会遇到 streaming interrupted 错误，导致消息接收不完整。这种情况通常由以下几个原因引起：

网络波动：不稳定的网络连接可能导致数据包丢失或延迟
API 限流：服务端可能因请求频率过高而主动断开连接
前后端协作问题：不恰当的消息处理逻辑可能导致缓冲区溢出或消息丢失

在选择流式传输方案时，我们需要考虑不同技术的适用场景：

短轮询：简单但效率低，不适合实时场景
长轮询：减少了空请求，但仍有一定延迟
SSE(Server-Sent Events)：适合服务器向客户端的单向通信
WebSocket：全双工通信，最适合 ChatGPT 流式场景

消息 ID 序列化与校验：为每个消息块分配唯一 ID，便于重组和校验
指数退避重试策略：在连接中断时，采用渐进式重试间隔
前端消息缓冲区设计：合理设置缓冲区大小和刷新策略

// 使用 axios 处理流式响应
const axios = require('axios');

async function streamChatGPT(prompt) {
  try {
    const response = await axios({
      method: 'post',
      url: 'https://api.openai.com/v1/chat/completions',
      responseType: 'stream',
      data: {
        model: 'gpt-3.5-turbo',
        messages: [{role: 'user', content: prompt}],
        stream: true
      },
      headers: {'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`
      }
    });

    response.data.on('data', (chunk) => {
      // 处理每个数据块
      const lines = chunk.toString().split('\n').filter(line => line.trim() !== '');
      for (const line of lines) {if (line.startsWith('data:')) {const data = line.replace('data:', '');
          if (data === '[DONE]') {console.log('Stream completed');
            return;
          }
          try {const parsed = JSON.parse(data);
            // 处理消息逻辑
          } catch (err) {console.error('Error parsing JSON:', err);
          }
        }
      }
    });

    response.data.on('error', (err) => {console.error('Stream error:', err);
      // 实现指数退避重试逻辑
    });

  } catch (err) {console.error('Request failed:', err);
  }
}

import React, {useState, useEffect} from 'react';

function ChatGPTStream({prompt}) {const [message, setMessage] = useState('');
  const [buffer, setBuffer] = useState([]);

  useEffect(() => {const eventSource = new EventSource(`/api/chat?prompt=${encodeURIComponent(prompt)}`);

    eventSource.onmessage = (event) => {
      try {const data = JSON.parse(event.data);
        // 使用缓冲区处理消息块
        setBuffer(prev => [...prev, data]);

        // 定期刷新 UI 以避免频繁渲染
        if (buffer.length % 5 === 0) {setMessage(buffer.map(b => b.choices[0].delta.content).join(''));
        }
      } catch (err) {console.error('Error parsing message:', err);
      }
    };

    eventSource.onerror = (err) => {console.error('EventSource failed:', err);
      // 实现重连逻辑
    };

    return () => eventSource.close();
  }, [prompt]);

  return <div>{message}</div>;
}