Nuxt3中API间ChatGPT流式数据传输问题求助

问题背景 #

正在将Next.js实现的ChatGPT API流式响应功能迁移到Nuxt3应用，目标是向前端实时返回分块数据，预期数据流程：
(a)Nuxt3前端 <-> (b)Nitro包装API <-> (c)Nitro上的ChatGPT API处理器 <-> (d)ChatGPT API

环境配置 #

• nuxt: ^3.2.3
• Windows 11
• node.js 16

当前问题 #

(c)处理器可正常从ChatGPT API实时拉取流数据，但(b)与(c)之间的数据传输异常，推测是(c)返回分块数据或(b)接收流的环节存在问题。

关键代码 #

(b) 包装API：/server/api/response.post.ts #

import { OpenAIStream, OpenAIStreamPayload } from '@/server/api/openaistream'

type RequestData = {
  currentModel: string
  message: string
}

export default defineEventHandler(async (event) => {
  const { message } = await readBody(event);

  const payload: OpenAIStreamPayload = {
    model: 'gpt-3.5-turbo',
    messages: [{ role: 'user', content: message }],
    temperature: 0.7,
    top_p: 1,
    frequency_penalty: 0,
    presence_penalty: 0,
    max_tokens: 500,
    stream: true,
    n: 1,
  }

  const stream = await OpenAIStream(payload);
  return new Response(stream);
})

(c) ChatGPT处理器：/server/api/openaistream.ts #

import {
  createParser,
  ParsedEvent,
  ReconnectInterval,
} from 'eventsource-parser';

export type ChatGPTAgent = 'user' | 'system';
export interface ChatGPTMessage {
  //abbreviated
}
export interface OpenAIStreamPayload {
  //abbreviated
}

export async function OpenAIStream(payload: OpenAIStreamPayload) {
  const encoder = new TextEncoder();
  const decoder = new TextDecoder();

  let counter = 0;

  const res = await fetch('https://api.openai.com/v1/chat/completions', {
    headers: {
      'Content-Type': 'application/json',
      Authorization: `Bearer ${process.env.OPENAI_API_KEY ?? ''}`,
    },
    method: 'POST',
    body: JSON.stringify(payload),
  });

  const stream = new ReadableStream({
    async start(controller) {
      // callback
      function onParse(event: ParsedEvent | ReconnectInterval) {
        if (event.type === 'event') {
          const data = event.data;
          // https://beta.openai.com/docs/api-reference/completions/create#completions/create-stream
          if (data === '[DONE]') {
            console.log("END!");
            controller.close();
            return;
          }
          try {
            const json = JSON.parse(data);
            const text = json.choices[0].delta?.content || '';
            // check if it receives chunked data from chatgpt API
            console.log(text);
            if (counter < 2 && (text.match(/\n/) || []).length) {
              // this is a prefix character (i.e., "\n\n"), do nothing
              return;
            }
            const queue = encoder.encode(text);
            controller.enqueue(queue);
            counter++;
          } catch (e) {
            // maybe parse error
            controller.error(e);
          }
        }
      }

      for await (const chunk of res.body as any) {
        //check if it can decode the chunk data to send as stream data
        console.log(decoder.decode(chunk))
        parser.feed(decoder.decode(chunk));
      }
    },
  });

  return stream;
}

排查思路与解决方案 #

1. 修复openaistream.ts中的未初始化parser问题 #

代码中调用了parser.feed但未创建parser实例，这会导致无法解析ChatGPT返回的SSE流数据。在start函数内添加parser初始化：

async start(controller) {
  // 新增：初始化parser
  const parser = createParser(onParse);
  
  // 原有的onParse函数和后续代码...
}

2. 适配Nitro的流式响应处理 #

Nuxt3的Nitro（基于h3）不能直接返回Response对象来处理流式响应，需要改用h3提供的sendStream方法，或者手动处理响应流：

方式一：使用sendStream

修改/server/api/response.post.ts的返回逻辑：

import { sendStream } from 'h3'
import { OpenAIStream, OpenAIStreamPayload } from '@/server/api/openaistream'

// ...其他代码

export default defineEventHandler(async (event) => {
  // ...获取payload和stream的代码
  
  // 设置正确的响应头
  event.node.res.setHeader('Content-Type', 'text/plain; charset=utf-8')
  event.node.res.setHeader('Transfer-Encoding', 'chunked')
  event.node.res.setHeader('Cache-Control', 'no-cache')
  event.node.res.setHeader('Connection', 'keep-alive')
  
  return sendStream(event, stream)
})

方式二：手动管道流到响应

如果sendStream不生效，可手动处理流的读取与写入：

export default defineEventHandler(async (event) => {
  // ...获取payload和stream的代码
  
  // 设置响应头
  event.node.res.setHeader('Content-Type', 'text/plain; charset=utf-8')
  event.node.res.setHeader('Transfer-Encoding', 'chunked')
  event.node.res.setHeader('Cache-Control', 'no-cache')
  event.node.res.setHeader('Connection', 'keep-alive')

  const reader = stream.getReader()
  const pump = async () => {
    const { done, value } = await reader.read()
    if (done) {
      event.node.res.end()
      return
    }
    event.node.res.write(value)
    await pump()
  }

  await pump()
  return event.node.res
})

3. 优化chunk解码方式 #

处理多字节字符时，使用流模式解码避免截断，修改openaistream.ts中的解码逻辑：

for await (const chunk of res.body as any) {
  console.log(decoder.decode(chunk, { stream: true }))
  parser.feed(decoder.decode(chunk, { stream: true }));
}

处理完所有chunk后，可调用decoder.decode()完成最终解码（可选）。

内容的提问来源于stack exchange，提问作者Yuji N.