在NextJS中修改OpenAI流式响应的实现方案咨询

一、用Vercel AI库实现修改流式数据 #

完全可以用Vercel的ai库实现这个需求，库本身提供了灵活的流式数据处理钩子，能在返回响应前修改每一段数据。

以Next.js API路由为例，你可以在调用流式接口时，通过自定义处理函数修改返回的内容格式：

import { openai } from '@ai-sdk/openai';
import { streamText } from 'ai';

export const runtime = 'edge';

export async function POST(req: Request) {
  const { conversation } = await req.json();

  const result = await streamText({
    model: openai('gpt-4'),
    messages: conversation,
    onStreamChunk: ({ chunk }) => {
      // 这里修改chunk内容，返回自定义格式
      const customChunk = {
        type: 'chat',
        data: {
          role: 'assistant',
          content: chunk.text,
        },
        myProperty: false,
      };
      // 返回修改后的字符串，要符合SSE格式
      return `data: ${JSON.stringify(customChunk)}\n\n`;
    },
  });

  return result.toDataStreamResponse();
}

这样就能直接通过库的钩子完成数据修改，同时保持流式传输的特性，不需要自己处理底层流的转换。

二、不用库的原生正确实现 #

你之前的代码问题在于混淆了Node.js的Transform流和Web Streams API的TransformStream，NextResponse只接受Web Streams标准的流对象，而且不需要额外创建WritableStream来转发数据，直接用pipeThrough连接转换流即可。

以下是修正后的代码：

import { OpenAI } from 'openai';

const openai = new OpenAI();

export const runtime = 'edge';

export async function POST(req: Request) {
  const { conversation } = await req.json();

  // 调用OpenAI流式接口，获取Node.js可读流
  const completion = await openai.chat.completions.create({
    model: 'gpt-4',
    stream: true,
    messages: conversation,
  });

  // 将Node.js流转换为Web Stream（Node.js 18+支持）
  const webStream = ReadableStream.from(completion);

  // 创建Web Streams标准的TransformStream
  const transformStream = new TransformStream({
    async transform(chunk, controller) {
      const chunkStr = new TextDecoder().decode(chunk);
      // 分割SSE的多行数据（可能一个chunk包含多个data块）
      const lines = chunkStr.split('\n').filter(line => line.trim() !== '');

      for (const line of lines) {
        if (line.startsWith('data: ')) {
          const dataStr = line.replace('data: ', '');
          if (dataStr === '[DONE]') continue;

          try {
            const data = JSON.parse(dataStr);
            const content = data.choices[0].delta.content; // 注意流式返回是delta，不是完整message
            if (content) {
              const customData = {
                type: 'chat',
                data: {
                  role: 'assistant',
                  content: content,
                },
                myProperty: false,
              };
              // 将自定义数据转为SSE格式并推送
              controller.enqueue(new TextEncoder().encode(`data: ${JSON.stringify(customData)}\n\n`));
            }
          } catch (err) {
            console.error('解析SSE数据失败:', err);
          }
        }
      }
    },
  });

  // 连接流并返回响应
  return new NextResponse(webStream.pipeThrough(transformStream), {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
    },
  });
}

关键修正点： #

• 使用Web Streams的TransformStream替代Node.js的Transform，符合NextResponse的要求；
• 处理流式返回的delta字段（OpenAI流式响应返回的是增量内容，不是完整的message）；
• 正确分割SSE的多行数据（一个chunk可能包含多个data行）；
• 手动设置SSE响应头，确保客户端能正确解析流式数据；
• 将Node.js可读流转换为Web Stream，用ReadableStream.from()完成适配（Node.js 18+内置支持）。

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