在Next.js API路由中高效传输OpenAI流式响应到客户端

// app/api/chat/route.ts
import { NextResponse } from 'next/server';
import OpenAI from 'openai'; // 确保已安装 openai 包

// 初始化OpenAI客户端
// 确保在环境变量中设置 OPENAI_API_KEY
const openai = new OpenAI({
  apiKey: process.env.OPENAI_API_KEY,
});

/**
 * 辅助函数：将异步迭代器转换为 ReadableStream
 * @param iterator 异步迭代器，每次 next() 返回 Uint8Array
 * @returns ReadableStream
 */
function iteratorToStream(iterator: AsyncIterator<Uint8Array>): ReadableStream<Uint8Array> {
  return new ReadableStream({
    async pull(controller) {
      const { value, done } = await iterator.next();
      if (done) {
        controller.close();
      } else {
        controller.enqueue(value);
      }
    },
  });
}

/**
 * POST 请求处理器，用于处理OpenAI流式对话请求
 */
export async function POST(request: Request) {
  // 从请求体中解析消息内容
  const { messages } = await request.json();

  try {
    // 调用OpenAI API创建聊天完成，并开启流式传输
    const oaiResponse = await openai.chat.completions.create({
      model: "gpt-3.5-turbo", // 推荐使用支持聊天的模型
      messages: messages,     // 客户端发送的对话消息
      stream: true,           // 开启流式传输
    });

    const encoder = new TextEncoder(); // 用于将字符串编码为 Uint8Array
    let completeMessage = ''; // 用于存储完整的响应内容（可选）

    // 异步生成器函数，用于逐块处理OpenAI响应并产出可流式传输的数据
    async function* makeIterator() {
      // 遍历OpenAI返回的每个数据块
      for await (const chunk of oaiResponse) {
        // 提取delta内容，即当前数据块的文本部分
        const delta = chunk.choices[0]?.delta?.content || '';
        completeMessage += delta; // 累积完整消息

        // 将每个delta封装为JSON对象并编码，然后产出
        // 这样做的好处是可以在流中传输结构化数据，例如除了文本还包含其他元数据
        yield encoder.encode(JSON.stringify({ type: "chunk", content: delta }) + '\n');
      }
      // (可选) 在流结束时发送一个包含完整内容或其他元数据的消息
      yield encoder.encode(JSON.stringify({ type: "done", full_content: completeMessage }) + '\n');
    }

    // 返回一个 Response 对象，其主体是转换后的 ReadableStream
    // 设置 Content-Type 为 text/plain; charset=utf-8，表示返回的是纯文本流
    // 客户端将按行解析这些JSON字符串
    return new Response(iteratorToStream(makeIterator()), {
      headers: { 'Content-Type': 'text/plain; charset=utf-8' },
    });

  } catch (error) {
    console.error("OpenAI API 调用失败:", error);
    // 错误处理：返回一个JSON错误响应
    return NextResponse.json({ error: "Failed to generate completion" }, { status: 500 });
  }
}

// components/ChatComponent.tsx
'use client'; // 标记为客户端组件 (App Router)

import React, { useState } from 'react';

export default function ChatComponent() {
  const [responseContent, setResponseContent] = useState('');
  const [isLoading, setIsLoading] = useState(false);

  const handleStreamResponse = async () => {
    setIsLoading(true);
    setResponseContent(''); // 清空之前的响应内容

    try {
      const response = await fetch('/api/chat', { // 调用前面定义的API路由
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
        },
        // 示例消息，实际应用中可以从用户输入获取
        body: JSON.stringify({ messages: [{ role: "user", content: "请讲一个关于勇敢骑士的故事。" }] }),
      });

      if (!response.ok || !response.body) {
        throw new Error(`HTTP 错误! 状态: ${response.status}`);
      }

      // 获取响应体的 ReadableStreamReader
      const reader = response.body.getReader();
      const decoder = new TextDecoder('utf-8'); // 用于解码 Uint8Array 到字符串

      let accumulatedChunk = ''; // 用于累积不完整的行

      // 循环读取流中的数据块
      while (true) {
        const { value, done } = await reader.read(); // 读取下一个数据块
        if (done) {
          // 流已结束，处理剩余的累积块（如果存在）
          if (accumulatedChunk.trim() !== '') {
            try {
              const parsed = JSON.parse(accumulatedChunk);
              if (parsed.type === "chunk") {
                setResponseContent((prev) => prev + parsed.content);
              } else if (parsed.type === "done") {
                console.log("完整内容已接收:", parsed.full_content);
              }
            } catch (parseError) {
              console.error("解析 JSON 块失败 (剩余部分):", accumulatedChunk, parseError);
            }
          }
          break;
        }

        // 解码当前数据块，并追加到累积字符串
        accumulatedChunk += decoder.decode(value, { stream: true });

        // 按行分割累积的字符串，处理完整的行
        const lines = accumulatedChunk.split('\n');
        // 保留最后可能不完整的一行，留待下一次读取
        accumulatedChunk = lines.pop() || '';

        for (const line of lines) {
          if (line.trim() === '') continue; // 跳过空行
          try {
            const parsed = JSON.parse(line);
            if (parsed.type === "chunk") {
              // 实时更新UI，追加内容
              setResponseContent((prev) => prev + parsed.content);
            } else if (parsed.type === "done") {
              // 处理流结束时的最终消息
              console.log("完整内容已接收:", parsed.full_content);
            }
          } catch (parseError) {
            console.error("解析 JSON 块失败:", line, parseError);
          }
        }
      }
    } catch (error) {
      console.error("获取流时发生错误:", error);
      setResponseContent("错误: " + (error as Error).message);
    } finally {
      setIsLoading(false);
    }
  };

  return (
    <div>
      <button onClick={handleStreamResponse} disabled={isLoading}>
        {isLoading ? '生成中...' : '生成流式响应'}
      </button>
      <div style={{ whiteSpace: 'pre-wrap', marginTop: '20px', border: '1px solid #ccc', padding: '10px' }}>
        {responseContent || '点击 "生成流式响应" 开始。'}
      </div>
    </div>
  );
}