Next.js 13+（App Router）中Langchain流式传输功能无法实现的问题

1. 实现API路由（App Router） #

在app/api/chat/route.ts中编写流式响应逻辑，用Next.js的ReadableStream构建符合SSE规范的响应流，替代传统的res.write方式：

import { NextResponse } from 'next/server';
import { ConversationalRetrievalQAChain } from 'langchain/chains';
import { ChatOpenAI } from 'langchain/chat_models/openai';
import { MemoryVectorStore } from 'langchain/vectorstores/memory';
import { OpenAIEmbeddings } from 'langchain/embeddings/openai';

export async function POST(request: Request) {
  const { question, chatHistory } = await request.json();

  // 初始化流式LLM模型
  const model = new ChatOpenAI({
    streaming: true,
    temperature: 0,
  });

  // 初始化向量存储（替换为你的实际向量源）
  const vectorStore = new MemoryVectorStore(new OpenAIEmbeddings());

  // 创建对话检索QA链
  const chain = ConversationalRetrievalQAChain.fromLLM(
    model,
    vectorStore.asRetriever(),
    {
      returnSourceDocuments: false,
      memory: { chatHistory },
    }
  );

  // 构建可读流，将Langchain输出转换为SSE格式
  const stream = new ReadableStream({
    async start(controller) {
      try {
        // 获取Langchain的流式结果
        const streamResult = await chain.stream({
          question,
          chat_history: chatHistory,
        });

        // 遍历流式chunk，包装为SSE消息
        for await (const chunk of streamResult) {
          const sseMsg = `event: message\ndata: ${JSON.stringify({ content: chunk })}\n\n`;
          controller.enqueue(new TextEncoder().encode(sseMsg));
        }

        // 流结束时发送close事件
        const closeMsg = `event: close\ndata: close\n\n`;
        controller.enqueue(new TextEncoder().encode(closeMsg));
        controller.close();
      } catch (err) {
        controller.error(err);
      }
    },
  });

  // 返回带正确头信息的响应
  return new NextResponse(stream, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
      'Transfer-Encoding': 'chunked',
    },
  });
}

2. 前端接收流式响应 #

方式一：使用EventSource #

const sendChatRequest = (question, chatHistory) => {
  const eventSource = new EventSource(
    `/api/chat?question=${encodeURIComponent(question)}`
  );

  eventSource.addEventListener('message', (e) => {
    const data = JSON.parse(e.data);
    // 追加流式内容到聊天界面
    console.log(data.content);
  });

  eventSource.addEventListener('close', () => {
    eventSource.close();
    console.log('对话流已结束');
  });

  eventSource.addEventListener('error', (err) => {
    console.error('流式请求出错:', err);
    eventSource.close();
  });
};

方式二：使用Fetch API（支持POST传参） #

const sendChatRequest = async (question, chatHistory) => {
  const res = await fetch('/api/chat', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ question, chatHistory }),
  });

  if (!res.body) return;

  const reader = res.body.getReader();
  const decoder = new TextDecoder();

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    const chunk = decoder.decode(value, { stream: true });
    // 拆分并解析SSE消息
    const messages = chunk.split('\n\n').filter(Boolean);
    
    messages.forEach(msg => {
      if (msg.startsWith('event: message')) {
        const data = JSON.parse(msg.split('data: ')[1]);
        console.log(data.content);
      } else if (msg.startsWith('event: close')) {
        console.log('对话流已结束');
      }
    });
  }
};

关键注意事项 #

• 必须确保Langchain的LLM模型开启streaming: true配置，否则无法生成流式响应
• App Router的Route Handler只能通过ReadableStream返回流式内容，不支持传统的res.write
• SSE响应头必须严格配置，避免浏览器缓存或提前中断连接
• 前端需正确解析SSE格式的消息，每个消息以\n\n作为分隔符

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