如何在AWS Lambda中使用Nest.js SSE？本地正常部署后失效解决方案

核心问题说明 #

Nest.js原生的@Sse()装饰器依赖传统HTTP服务器的持续连接模型，但AWS Lambda的无状态执行环境需要适配Lambda的响应流式传输机制，同时配合API Gateway（或Lambda Function URL）的流式配置才能正常工作。

解决方案步骤 #

1. 启用Lambda的响应流式功能 #

• 在AWS控制台的Lambda函数配置页，进入“配置”>“函数概述”>“编辑”，开启响应流式传输。
• 若用Serverless Framework，在serverless.yml中添加配置：

  functions:
    conversationFunc:
      handler: dist/main.handler
      events:
        - httpApi:
            path: /conversation
            method: post
            payloadFormatVersion: '2.0'
            responseStreaming: true
  

• 若用AWS CDK，在Lambda函数构造中设置responseStreaming: true参数。

2. 修改Nest.js代码适配Lambda流式 #

放弃原生@Sse()装饰器，直接返回AsyncIterable类型响应，并手动配置SSE格式的响应头。

服务层代码修改

将原Observable逻辑改为返回AsyncIterable<string>，直接复用OpenAI的流式迭代器：

async *sendMessage(dto: ChatDto): AsyncIterable<string> {
  const configuration = {
    apiKey: process.env.OPENAI_API_KEY
  };
  const openai = new OpenAI(configuration);

  try {
    const responseStream = await openai.chat.completions.create({
      model: dto.model,
      messages: dto.messages as any,
      stream: true,
    });

    for await (const chunk of responseStream) {
      const content = chunk.choices[0].delta.content;
      if (content) { // 过滤空内容块
        // 按SSE规范格式返回数据
        yield `data: ${JSON.stringify({ data: content })}\n\n`;
      }
    }
  } catch (error) {
    yield `event: error\ndata: ${JSON.stringify({ message: error.message })}\n\n`;
  }
}

控制器层代码修改

移除@Sse()装饰器，使用普通@Post()，手动设置响应头并逐块写入流式数据：

import { Controller, Post, Body, UseGuards, Res } from '@nestjs/common';
import { Response } from 'express';

@UseGuards(JwtGuard, ApiKeyGuard)
@Post("conversation")
async sendMessage(
  @Body() dto: ChatDto,
  @Res() res: Response
) {
  // 配置SSE响应头
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');

  const stream = this.service.sendMessage(dto);
  // 逐块写入响应并强制刷新缓冲区（Lambda环境需确保数据及时发送）
  for await (const chunk of stream) {
    res.write(chunk);
    await new Promise(resolve => res.flushHeaders(resolve));
  }

  res.end();
}

3. 关键注意事项 #

• Lambda流式响应要求返回的AsyncIterable元素为字符串或Uint8Array，直接返回SSE格式字符串即可。
• 确保API Gateway的payload格式版本为2.0，否则无法正确传递流式响应。
• 若使用Lambda Function URL，只需在配置中启用响应流式，无需额外配置API Gateway。

测试验证 #

部署修改后的代码到Lambda，调用接口时客户端会持续接收SSE格式的流式数据，行为与本地环境一致。

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