跳到主要内容

流式渲染与 SSE

问题​

AI 应用中的流式输出是如何实现的？前端如何处理和渲染流式文本？

答案​

一、为什么需要流式输出​

LLM 生成是逐 Token 的过程，一次完整生成可能需要 5-30 秒。流式输出让用户边生成边看，大幅提升体验：

方式等待感受TTFT
非流式等 10 秒后一次性显示全部10s
流式0.5 秒开始逐字显示0.5s
TTFT（Time to First Token）

TTFT 是衡量 AI 应用响应速度的关键指标——用户看到第一个字的时间。

二、SSE（Server-Sent Events）​

SSE 是 AI 流式输出最常用的协议：

SSE 数据格式：每条消息以 data: 前缀，以 \n\n 分隔：

data: {"choices":[{"delta":{"content":"你"}}]}

data: {"choices":[{"delta":{"content":"好"}}]}

data: [DONE]

三、后端实现​

app/api/chat/route.ts
import OpenAI from "openai";

const openai = new OpenAI();

export async function POST(req: Request) {
  const { messages } = await req.json();

  // 调用 OpenAI 流式 API
  const stream = await openai.chat.completions.create({
    model: "gpt-4o",
    messages,
    stream: true,
  });

  // 转换为 ReadableStream
  const encoder = new TextEncoder();
  const readableStream = new ReadableStream({
    async start(controller) {
      for await (const chunk of stream) {
        const text = chunk.choices[0]?.delta?.content || "";
        if (text) {
          // SSE 格式：data: ...\n\n
          controller.enqueue(encoder.encode(`data: ${JSON.stringify({ text })}\n\n`));
        }
      }
      controller.enqueue(encoder.encode("data: [DONE]\n\n"));
      controller.close();
    },
  });

  return new Response(readableStream, {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      Connection: "keep-alive",
    },
  });
}

四、前端消费​

方式一：EventSource（简单但只支持 GET）​

const source = new EventSource("/api/chat");
source.onmessage = (event) => {
  if (event.data === "[DONE]") {
    source.close();
    return;
  }
  const { text } = JSON.parse(event.data);
  appendToChat(text);
};

方式二：fetch + ReadableStream（推荐，支持 POST）​

async function streamChat(messages: Message[]) {
  const response = await fetch("/api/chat", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ messages }),
  });

  const reader = response.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

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

    buffer += decoder.decode(value, { stream: true });

    // 按 SSE 分隔符解析
    const lines = buffer.split("\n\n");
    buffer = lines.pop()!; // 保留未完成的部分

    for (const line of lines) {
      const data = line.replace("data: ", "");
      if (data === "[DONE]") return;
      const { text } = JSON.parse(data);
      appendToChat(text);
    }
  }
}

方式三：Vercel AI SDK（最简）​

"use client";
import { useChat } from "@ai-sdk/react";

export default function Chat() {
  // useChat 自动处理流式请求、解析、状态管理
  const { messages, input, handleInputChange, handleSubmit } = useChat();

  return (
    <div>
      {messages.map((m) => (
        <div key={m.id}>{m.content}</div>
      ))}
      <form onSubmit={handleSubmit}>
        <input value={input} onChange={handleInputChange} />
      </form>
    </div>
  );
}

五、流式 Markdown 渲染​

AI 输出通常是 Markdown 格式，流式渲染需要处理未完成的 Markdown：

import ReactMarkdown from "react-markdown";
import remarkGfm from "remark-gfm";
import { Prism as SyntaxHighlighter } from "react-syntax-highlighter";

function StreamingMessage({ content }: { content: string }) {
  return (
    <ReactMarkdown
      remarkPlugins={[remarkGfm]}
      components={{
        code({ className, children, ...props }) {
          const match = /language-(\w+)/.exec(className || "");
          return match ? (
            <SyntaxHighlighter language={match[1]}>
              {String(children)}
            </SyntaxHighlighter>
          ) : (
            <code className={className} {...props}>{children}</code>
          );
        },
      }}
    >
      {content}
    </ReactMarkdown>
  );
}
流式 Markdown 的挑战
  • 未闭合的代码块（``` 只出现了开头）
  • 未完成的表格行
  • 部分链接语法 [text](url
  • 建议在渲染前做简单的语法补全处理

常见面试问题​

Q1: SSE 和 WebSocket 做 AI 流式输出哪个更好？​

答案：

  • SSE：单向流（Server→Client），基于 HTTP，简单可靠，AI 流式输出首选
  • WebSocket：双向通信，适合需要实时交互的场景（如协同编辑、聊天室）
  • AI 对话场景是请求-流式响应模式，SSE 完全足够且更简单

Q2: 如何处理流式输出中的错误？​

答案：

  1. HTTP 错误：在 fetch 后检查 response.ok
  2. 流中断：reader.read() 返回 done: true，或连接断开
  3. LLM 错误：在 SSE 数据中携带 error 字段
  4. 超时：设置 AbortController，超时后取消请求
const controller = new AbortController();
const timeout = setTimeout(() => controller.abort(), 60000);

const response = await fetch("/api/chat", {
  signal: controller.signal,
  // ...
});

相关链接​