AWS Lambda实现SSE：后续调用无日志触发内部服务器错误

核心问题 #

你的代码存在几个关键问题，导致了无日志的内部服务器错误：

• requestStream未初始化：代码中直接使用了requestStream变量，但从未创建Readable流实例。首次调用可能因Lambda临时执行环境侥幸运行，后续调用会因变量未定义触发异步错误，这类错误不在try/catch捕获范围内，因此无日志输出。
• 未监听OpenAI流的error事件：OpenAI响应流若发生网络中断等错误，会触发error事件，未监听的话会导致未捕获异常，直接终止函数且无日志。
• 流结束逻辑重复：在end事件中同时调用requestStream.push(null)和responseStream.end()，但pipeline会在源流结束时自动处理目标流的结束，重复调用会引发异常。
• 拼写错误：No promt provided中的promt应为prompt，可能导致错误提示不准确。

修复后的代码 #

import util from "util";
import stream from "stream";
import { Configuration, OpenAIApi } from "openai";

const { Readable } = stream;
const pipeline = util.promisify(stream.pipeline);
// 假设system是你定义的系统消息数组，此处补充示例定义
const system = [{ role: "system", content: "You are a helpful assistant." }];

// 初始化OpenAI配置（确保环境变量已设置OPENAI_API_KEY）
const configuration = new Configuration({ apiKey: process.env.OPENAI_API_KEY });
const openai = new OpenAIApi(configuration);

export const handler = awslambda.streamifyResponse(
  async (event, responseStream, _context) => {
    let requestStream = null;
    try {
      const { prompt } = JSON.parse(event.body ?? "{}"); // 解析为对象而非数组

      if (!prompt) throw new Error("No prompt provided");
      if (!Array.isArray(prompt)) throw new Error("Prompt should be an array");

      const messages = [...system, ...prompt];

      const completion = await openai.createChatCompletion(
        {
          model: "gpt-4",
          temperature: 0.3,
          messages: messages,
          stream: true,
        },
        { responseType: "stream" }
      );

      // 初始化自定义可读流
      requestStream = new Readable({
        read() {} // 手动控制push，read方法留空
      });

      // 监听OpenAI流的data事件
      completion.data.on("data", (data) => {
        if (!requestStream.destroyed) { // 用destroyed判断流状态更准确
          const dataStr = data.toString();
          console.log("Data as string: ", dataStr);
          requestStream.push(dataStr);
        }
      });

      // 监听OpenAI流的end事件
      completion.data.on("end", () => {
        console.log("On End");
        if (!requestStream.destroyed) {
          requestStream.push(null); // 结束源流，pipeline会自动结束responseStream
        }
      });

      // 监听OpenAI流的error事件，捕获异步错误
      completion.data.on("error", (err) => {
        console.error("OpenAI stream error: ", err);
        if (!requestStream.destroyed) {
          requestStream.destroy(err); // 销毁流并传递错误，pipeline会捕获
        }
      });

      // 设置SSE响应头
      responseStream.setContentType("text/event-stream");
      responseStream.setHeader("Cache-Control", "no-cache");
      responseStream.setHeader("Connection", "keep-alive");

      // 执行流管道
      await pipeline(requestStream, responseStream);
      console.log("FINISHED!!!!!");
    } catch (error) {
      console.error("Handler error: ", error);
      // 处理初始化阶段的错误
      if (requestStream && !requestStream.destroyed) {
        requestStream.push(
          JSON.stringify({ error: error.message, details: error.response?.data })
        );
        requestStream.push(null);
      }
      // 确保响应流正确结束
      if (!responseStream.closed) {
        responseStream.setContentType("application/json");
        responseStream.write(JSON.stringify({ error: error.message }));
        responseStream.end();
      }
    }
  }
);

调试建议 #

• 捕获所有异步错误：给所有流（OpenAI响应流、自定义requestStream）添加error事件监听，确保异步错误能被记录到CloudWatch日志。
• 启用Lambda详细日志：在Lambda函数配置中将日志级别设为DEBUG，确保日志组保留策略正确，避免日志丢失。
• 检查执行环境复用问题：Lambda会复用执行环境，若流资源未正确清理，后续调用会受影响。函数结束时需确保所有流都被销毁或结束。
• 模拟异常场景测试：手动模拟OpenAI API超时、网络中断等场景，观察日志是否能捕获到错误。
• 用CloudWatch Insights排查日志：通过CloudWatch Insights查询特定请求ID的日志，过滤ERROR级别日志，定位无日志错误的根源。

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