如何在Vercel部署的Next.js Serverless函数中使用EventSource

问题背景 #

需要将基于Edge函数实现的OpenAI流式响应代码，适配为可在Vercel部署的Next.js Serverless函数运行。由于依赖Firebase等需要Node.js环境的工具，无法继续使用Edge函数，同时不清楚如何在Serverless函数中实现类似流式响应效果。

解决方案 #

1. 修改 /api/generate.ts #

替换Edge函数的Request/Response模型为Next.js Serverless函数的NextApiRequest/NextApiResponse，调整响应处理逻辑：

import type { NextApiRequest, NextApiResponse } from "next";
import { OpenAIStream, OpenAIStreamPayload } from "../../utils/OpenAIStream";

if (!process.env.OPENAI_API_KEY) {
  throw new Error("缺少OpenAI API密钥环境变量");
}

export const config = {
  runtime: "nodejs", // 切换为Serverless运行时
};

const handler = async (req: NextApiRequest, res: NextApiResponse) => {
  const { prompt } = req.body as { prompt?: string };

  if (!prompt) {
    return res.status(400).json({ error: "请求中缺少prompt参数" });
  }

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

  // 设置流式响应头，符合SSE规范
  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache");
  res.setHeader("Connection", "keep-alive");
  res.setHeader("Access-Control-Allow-Origin", "*");

  try {
    await OpenAIStream(payload, res);
  } catch (error) {
    res.status(500).json({ error: "流式响应处理失败" });
  }
};

export default handler;

2. 修改 utils/OpenAIStream.ts #

适配Node.js环境的流处理逻辑，移除浏览器专属API，直接向NextApiResponse写入流式内容：

import { createParser, ParsedEvent, ReconnectInterval } from "eventsource-parser";
import type { NextApiResponse } from "next";

export type ChatGPTAgent = "user" | "system";

export interface ChatGPTMessage {
  role: ChatGPTAgent;
  content: string;
}

export interface OpenAIStreamPayload {
  model: string;
  messages: ChatGPTMessage[];
  temperature: number;
  top_p: number;
  frequency_penalty: number;
  presence_penalty: number;
  max_tokens: number;
  stream: boolean;
  n: number;
}

export async function OpenAIStream(payload: OpenAIStreamPayload, res: NextApiResponse) {
  const decoder = new TextDecoder();
  let counter = 0;

  const openAIRes = 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),
  });

  // 解析SSE格式的流式响应
  const parser = createParser((event: ParsedEvent | ReconnectInterval) => {
    if (event.type === "event") {
      const data = event.data;
      if (data === "[DONE]") {
        res.write("data: [DONE]\n\n");
        res.end();
        return;
      }
      try {
        const json = JSON.parse(data);
        const text = json.choices[0].delta?.content || "";
        // 跳过初始换行符
        if (counter < 2 && (text.match(/\n/) || []).length) {
          counter++;
          return;
        }
        // 以SSE格式写入响应内容
        res.write(`data: ${JSON.stringify({ content: text })}\n\n`);
        counter++;
      } catch (e) {
        res.status(500).end("解析响应出错");
      }
    }
  });

  if (!openAIRes.body) {
    res.status(500).end("OpenAI响应为空");
    return;
  }

  // 逐块读取OpenAI返回的流数据
  const reader = openAIRes.body.getReader();
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    parser.feed(decoder.decode(value));
  }
}

关键改动说明 #

• 运行时切换：将runtime从edge改为nodejs，确保支持Firebase等Node.js环境依赖。
• 响应头配置：设置符合SSE规范的响应头，保证客户端能持续接收流式内容。
• 流处理适配：移除浏览器专属的ReadableStream，改用Node.js原生流读取API，直接向响应对象写入解析后的内容。
• 格式兼容：保持和原Edge函数一致的SSE输出格式，前端无需额外修改即可兼容。

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