基于NestJS与Angular实现OpenAI流式响应的问题排查

问题1：流式响应结束后自动重复调用接口 #

EventSource默认会在连接断开（包括正常结束）时自动重试，这是重复调用的核心原因，需要前后端配合处理：

后端（NestJS）处理 #

确保流式响应结束时发送明确的结束标记，并主动关闭响应流：

@Post('chat-stream')
async chatStream(@Body() body: ChatRequest, @Res() res: Response) {
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');

  const stream = await openai.chat.completions.create({
    model: 'gpt-3.5-turbo',
    messages: body.messages,
    stream: true,
  });

  for await (const chunk of stream) {
    const content = chunk.choices[0]?.delta?.content || '';
    if (content) {
      res.write(`data: ${JSON.stringify({ content })}\n\n`);
      await new Promise(resolve => setTimeout(resolve, 10));
    }
  }

  // 发送结束标记
  res.write('data: [DONE]\n\n');
  res.end(); // 主动关闭响应连接
}

前端（Angular）处理 #

监听消息事件，收到结束标记后主动关闭EventSource实例：

private eventSource: EventSource | null = null;

startChat(messages: ChatMessage[]) {
  // 先关闭已有连接
  if (this.eventSource) {
    this.eventSource.close();
  }

  const url = `/api/chat-stream?messages=${encodeURIComponent(JSON.stringify(messages))}`;
  this.eventSource = new EventSource(url);

  this.eventSource.onmessage = (event) => {
    if (event.data === '[DONE]') {
      this.eventSource?.close(); // 收到结束信号后主动终止连接
      return;
    }
    const data = JSON.parse(event.data);
    // 处理流式返回的内容
    console.log(data.content);
  };

  this.eventSource.onerror = () => {
    this.eventSource?.close(); // 出错时也主动关闭
  };
}

问题2：无法通过EventSource发送POST数据 #

EventSource仅支持GET请求，无法直接携带POST请求体，推荐两种解决方案：

方式1：用Fetch API模拟SSE（优先推荐） #

通过Fetch发送POST请求，手动处理流式响应，完全兼容POST传参，效果和SSE一致：

async startChat(messages: ChatMessage[]) {
  const response = await fetch('/api/chat-stream', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({ messages }),
  });

  if (!response.body) return;

  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格式的消息
    const messageChunks = chunk.split('\n\n').filter(msg => msg.startsWith('data:'));
    for (const msg of messageChunks) {
      const data = msg.slice(5); // 移除"data: "前缀
      if (data === '[DONE]') {
        return;
      }
      const content = JSON.parse(data).content;
      // 处理流式内容
      console.log(content);
    }
  }
}

后端无需修改，保持原有流式输出逻辑即可。

方式2：将数据编码到GET参数（仅适合小体量数据） #

如果对话历史、上下文数据量较小，可以将JSON数据编码后放到URL参数中，但受限于URL长度限制，不推荐用于大量数据场景。

内容的提问来源于stack exchange，提问作者Robouste