AI 对话"打字机"是怎么实现的？从 SSE 流式、Markdown 组件到 Nginx 防粘连

0先看整体：拆成哪七层

我们在做 AI 应用的时候，免不了实现对话问答，甚至更复杂的代码高亮、表格、图表。做了这么多的问答应用，整理下问答流程以及详细的渲染。

做 AI 问答应用，往往卡在几个点上：流式数据怎么收？协议怎么定？Markdown 怎么渲染？AI 输出的表格 / 图表怎么变成好看的组件？还有——为什么有时候字是一卡一卡"蹦"出来的？连接断了又该怎么处理？

一个 AI 对话界面，前端要做的事可以拆成七层：

① SSE 流式传输     —— 数据是怎么一波波到前端的
② 问答协议规范     —— 前后端怎么约定"增量/结束/错误"
③ 流式接收与拼接   —— 收到碎片后怎么拼成完整文本
④ Markdown 渲染    —— 怎么把文本变成带格式的界面
⑤ 自定义节点       —— 怎么插入表格、卡片、图表等富组件
⑥ 断线重连         —— 连接断了怎么处理、怎么重连
⑦ Nginx 防粘连     —— 为什么经过代理会"蹦字"，怎么修

1第一层：SSE —— 让数据像打字机一样"流"过来

1.1 为什么要流式？

传统请求是"等全部结果返回再展示"，但 AI 生成一句话要几秒到几十秒，让用户干等很难受。所以有了流式（Streaming）：数据边生成边传输，前端边收边显示，形成"打字机"效果——既让用户觉得快，又能提前看到内容。

1.2 SSE 是什么？和 WebSocket 有啥区别？

SSE（Server-Sent Events，服务器推送事件）：一种让服务器主动、持续向浏览器推送文本数据的协议，非常适合"AI 一个字一个字往外吐"。

2第二层：SSE 问答协议规范 —— 前后端怎么约定一次问答

流式问答的难点不在传输本身，而在前后端对"消息格式"的约定。约定不清楚，前端就不知道该拼哪段、什么时候结束、出错怎么处理。

2.1 SSE 报文长什么样

SSE 的本质是：服务器把一个文本流以特定格式"喂"给浏览器。一条消息的规范格式是：

event: 事件名(可省略)
data: 数据内容
（每条消息以空行 \n\n 结尾，服务器可以连续推送很多条）

2.2 推荐的一种问答规范（实践常用）

为了让前端能区分"正在输出 / 结束 / 出错"，业界常用统一在 data 里放 JSON，再用 type 字段区分事件类型：

// 正在生成：delta，携带增量内容
event: message
data: {"type":"delta","content":"你"}

// 生成结束：done
event: message
data: {"type":"done","messageId":"msg_123"}

// 出错：error
event: error
data: {"code":"5001","message":"服务繁忙，请稍后重试"}

这个规范的好处：前端只需要在 onmessage 里 switch 一下 type，就知道该"追加文本"还是"收尾"还是"报错"。你要做的就是在项目里定好这套 type 枚举，前后端对齐。

2.3 为什么用"增量 delta"而不是"整段全量"？

3第三层：传输层实战 —— 为什么用 @microsoft/fetch-event-source

原生 EventSource 太简单，遇到真实项目（要带 token、要 POST、要取消、要控制重连）就抓瞎。所以很多人选 @microsoft/fetch-event-source。

3.1 它比原生 EventSource 强在哪

它是"能自定义 headers + 能取消 + 能控制重连的 EventSource"，是生产级流式问答的标配。

3.2 核心用法

import { fetchEventSource } from '@microsoft/fetch-event-source';

const ctrl = new AbortController(); // 用于取消本次问答

async function chat(question, { onDelta, onDone }) {
  await fetchEventSource('/api/chat', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Authorization': `Bearer ${token}`   // 能带 token，这是关键
    },
    body: JSON.stringify({ question, stream: true }),
    signal: ctrl.signal,          // 支持取消
    openWhenHidden: true,         // 切后台也保持连接

    onopen: async (res) => {
      if (!res.ok) throw new Error(`连接失败: ${res.status}`);
    },

    onmessage: (ev) => {
      const data = JSON.parse(ev.data);
      if (data.type === 'delta') onDelta(data.content);  // 追加增量
      if (data.type === 'done')  onDone(data.messageId); // 收尾
    },

    onclose: () => console.log('连接关闭'),
  });
}

4第四层：断线重连 —— onerror 到底怎么控制"断了重连"

这是最容易翻车、也最容易被讲错的一块。先记住结论：onerror 的返回值就决定一件事——"多久后重连"。

4.1 onerror 返回值的正确语义

4.2 什么时候会触发 onerror？（哪些情况算"断"）

4.3 两个容易误解的点

• 主动取消不会重连：用户点"停止"调 ctrl.abort() 走的是正常结束分支，不会触发 onerror 重连，放心用；
• 库没有内置指数退避：默认固定 1000ms 重试。想要"1s、2s、4s"的退避效果，得自己在 onerror 里做（下面示例）。

4.4 实战：设计一套健壮的重连策略

工程上要解决四件事：限次数、做退避、区分可重试 / 不可重试错误、收到数据就重置。

import { fetchEventSource } from '@microsoft/fetch-event-source';

const MAX_RETRIES = 3;      // 最多重试 3 次
const MAX_INTERVAL = 30000; // 退避上限 30s

function createChat(onEvent) {
  let retryCount = 0;
  let interval = 1000;
  const ctrl = new AbortController();

  async function connect(question) {
    await fetchEventSource('/api/chat', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ question, stream: true }),
      signal: ctrl.signal,
      openWhenHidden: true,

      onopen: async (res) => {
        // 在 onopen 里主动判定"是否该继续"
        if (res.status === 401) throw new Error('UNAUTHORIZED'); // 不可重试
        if (res.status !== 200) throw new Error(`HTTP_${res.status}`);
      },

      onmessage: (ev) => {
        // 收到数据 → 连接健康 → 重置重试状态
        retryCount = 0;
        interval = 1000;
        const data = JSON.parse(ev.data);
        onEvent(data);
      },

      onerror: (err) => {
        // ① 不可重试的错误：throw，放弃
        if (err.message === 'UNAUTHORIZED') throw err;
        // ② 超过重试上限：throw，放弃
        if (retryCount >= MAX_RETRIES) {
          console.error('重试次数已用尽', err);
          throw err;
        }
        // ③ 否则：手动指数退避，返回间隔 → 触发重连
        retryCount += 1;
        interval = Math.min(interval * 2, MAX_INTERVAL);
        console.log(`第 ${retryCount} 次重连，${interval}ms 后重试`);
        return interval;   // 返回数字 = 按这个间隔重连
      },
    });
  }

  return {
    start: (q) => connect(q),
    stop: () => ctrl.abort(), // 主动取消，不触发重连
  };
}

4.5 配一个状态机，让 UI 知道"现在在干嘛"

// state: 'idle' | 'connecting' | 'streaming' | 'reconnecting' | 'closed'
const [phase, setPhase] = useState('idle');

onopen:    () => setPhase('connecting'),
onmessage: () => setPhase('streaming'),
onerror:   () => setPhase('reconnecting'),
onclose:   () => setPhase('closed'),

5第五层：数据处理 —— 把增量拼成完整 Markdown

收到 delta 后，需要维护一份"累加文本"，同时用节流控制渲染频率（否则每个字符都重渲染，卡）：

const [streamText, setStreamText] = useState('');

// 用 ref 存最新文本，配合节流更新视图
const textRef = useRef('');
const timerRef = useRef(null);

const onDelta = (delta) => {
  textRef.current += delta;
  if (timerRef.current) return;            // 节流：已有定时器就不再触发
  timerRef.current = setTimeout(() => {
    setStreamText(textRef.current);        // 批量更新一次界面
    timerRef.current = null;
  }, 60);                                  // 每 60ms 刷新一次
};

6第六层：MD 组件封装 —— 展示 / 自定义节点 / 表格 / 图表

这里就是很多团队自己封装的那些 MD 开头组件。核心思想只有一条，但很关键：

先把 Markdown 解析成 AST（语法树），再用一套"组件映射"把每种 AST 节点渲染成 React 组件。所谓"自定义节点"，就是往这套映射里塞你自己的组件。

6.1 一个 MD 组件的基本骨架

function Markdown({ content, components }) {
  return (
    <ReactMarkdown
      components={{
        h1: H1Node,
        table: TableNode,   // 表格节点
        code: CodeNode,     // 代码/自定义块节点
        a: LinkNode,
        ...components,      // 外部可继续扩展
      }}
    >
      {content}
    </ReactMarkdown>
  );
}

6.2 表格节点：把 Markdown 表格变"组件表格"

Markdown 的表格长这样，react-markdown 默认能渲染，但要自定义样式 / 交互，就接管 table 节点：

| 名称 | 状态 |
| ---- | ---- |
| 退款 | 待处理 |

function TableNode({ children }) {
  return <table className="md-table">{children}</table>;
  // 也可以解析成二维数组，用 antd/自研 Table 渲染
}

6.3 自定义节点：CodeNode 就是一个"分发器"

这是 MD 组件最灵活、也最核心的一环。要理解它，先记住一个关键点：

CodeNode 不是一个"代码渲染器"，而是一个"分发器（dispatcher）"——它拿到一段代码块，先看它的 language 标记，然后把渲染这件事"分派"给对应的组件。分发给谁，由你来定，这就是自定义能力的来源。

```xxx { ... } ```
    ↓ 进入 CodeNode（分发器）
    ↓ 读取 language 标记
    ├─ language = "card"   → 分发给 <Card> 组件
    ├─ language = "chart"  → 分发给 <Chart> 组件
    ├─ language = "js/py"  → 分发给 <SyntaxHighlighter> 代码高亮
    └─ 其他语言            → 走默认高亮 / 朴素显示

代码层面，CodeNode 就是一层 switch：

function CodeNode({ className, children }) {
  // 从 className 里取出语言标记，比如 language-chart → "chart"
  const lang = (className || '').replace('language-', '');

  // 分发器：按 language 决定"这段代码怎么展示"
  switch (lang) {
    case 'card':                       // 自定义卡片
      return <Card {...JSON.parse(String(children))} />;
    case 'chart':                      // 自定义图表
      return <Chart option={JSON.parse(String(children))} />;
    case 'button':                     // 自定义按钮
      return <ActionButton {...JSON.parse(String(children))} />;
    default:                           // 其余当普通代码，走代码高亮
      return <SyntaxHighlighter language={lang}>{String(children)}</SyntaxHighlighter>;
  }
}

6.4 图表展示：把 AI 的数据画成 ECharts

AI 对话里常出现"分析数据"的场景，纯文字表述不清楚，图表一眼看懂。思路和自定义节点一样：约定一种结构化格式，比如让 AI 输出一个特殊的 ECharts JSON 代码块：

```chart
{"type":"bar","data":{"x":["1月","2月","3月"],"y":[100,120,90]}}
```

前端识别 chart 代码块 → 解析 JSON → 动态渲染 ECharts 组件：

import ReactECharts from 'echarts-for-react';

function ChartNode({ config }) {
  const option = {
    xAxis: { data: config.data.x },
    yAxis: {},
    series: [{ type: config.type, data: config.data.y }]
  };
  return <ReactECharts option={option} />;
}

6.5 为什么用 AST 而不是正则？

正则只能处理"样子"匹配，遇到嵌套、转义、流式半截内容就崩；AST 是结构化解析，能准确区分"这是表格标题还是正文"，也方便精准接管某个节点。生产级渲染，选 AST。

7第七层：Nginx 配置 —— 让流式"不粘连"

很多人会遇到：直连后端逐字流畅，一经过 Nginx 就"2-3 秒蹦 20 个字"。这不是前端 bug，是 Nginx 默认行为跟 SSE 打架。

7.1 为什么会粘连（两层延迟）

• Nginx 缓冲：默认会把后端输出"攒到 4k/8k 才转发"，小数据一直被扣着；
• TCP 小包合并：即便转发，TCP 也会等约 200ms 看有没有后续小包一起发，形成"10 字 + 200ms"的卡顿节奏。

7.2 三件套配置

location /api/chat {
    proxy_pass http://127.0.0.1:8080;   # 你的后端 SSE 服务

    # 核心三件套
    proxy_http_version 1.1;   # 强制 HTTP/1.1，支持长连接 + 分块传输
    proxy_cache off;          # 禁用代理缓存，防旧数据污染
    proxy_buffering off;      # 【核心】关闭缓冲，逐字节实时透传

    # 补充配置
    proxy_read_timeout 3600s; # 长连接超时拉长
    gzip off;                 # 禁用压缩（压缩会攒包）
    proxy_set_header Connection "";
}

• proxy_http_version 1.1：解决"协议支持"，SSE 依赖 keep-alive 和 chunked；
• proxy_buffering off：解决粘连的核心，数据不攒、实时透传；
• proxy_cache off：解决"缓存污染"，保证每次都是实时数据。

8完整链路 + 踩坑汇总

POST /api/chat（fetch-event-source，带 token）
  → Nginx 不缓冲、实时透传
  → 后端按 SSE 协议吐 {type:delta}
  → 前端 onmessage 拼文本 + 节流
  → MD 组件 AST 渲染：普通/表格/代码/自定义卡片/图表
  → 断了？onerror 按退避策略重连，超限则放弃

高频坑清单

9选型建议：这几层能不能一起用？

10总结 & 术语表 & 附录

10.1 总结

SSE 协议定好"增量 / 结束 / 错误"的规范，fetch-event-source 负责带 token、可取消地把流接过来，onerror 的返回值控制"断了多久重连、要不要放弃"，MD 组件用 AST 把 Markdown 渲染成"普通文本 + 表格 + 自定义节点 + 图表"，而 Nginx 的 proxy_buffering off 三件套，是让这一切"逐字流畅、不粘连"。

术语表

附录 A：完整链路可运行骨架（fetch-event-source + 重连 + 状态机）

// ========== 可运行骨架：AI 对话流式客户端 ==========
import { fetchEventSource } from '@microsoft/fetch-event-source';
import { useState } from 'react';

const MAX_RETRIES = 3;
const MAX_INTERVAL = 30000;

export function useChat() {
  const [phase, setPhase] = useState('idle');   // idle|connecting|streaming|reconnecting|closed
  const [streamText, setStreamText] = useState('');

  const textRef = useRef('');
  const timerRef = useRef(null);
  let retryCount = 0;
  let interval = 1000;
  const ctrl = new AbortController();

  const pushDelta = (delta) => {
    textRef.current += delta;
    if (timerRef.current) return;
    timerRef.current = setTimeout(() => {
      setStreamText(textRef.current);   // 60ms 节流批量更新
      timerRef.current = null;
    }, 60);
  };

  async function ask(question) {
    setPhase('connecting');
    await fetchEventSource('/api/chat', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${token}`
      },
      body: JSON.stringify({ question, stream: true }),
      signal: ctrl.signal,
      openWhenHidden: true,

      onopen: async (res) => {
        if (res.status === 401) throw new Error('UNAUTHORIZED');
        if (!res.ok) throw new Error(`HTTP_${res.status}`);
      },

      onmessage: (ev) => {
        retryCount = 0; interval = 1000;
        setPhase('streaming');
        const data = JSON.parse(ev.data);
        if (data.type === 'delta') pushDelta(data.content);
        if (data.type === 'done')  { setPhase('closed'); ctrl.abort(); }
      },

      onerror: (err) => {
        if (err.message === 'UNAUTHORIZED' || retryCount >= MAX_RETRIES) {
          setPhase('closed'); throw err;
        }
        retryCount += 1;
        interval = Math.min(interval * 2, MAX_INTERVAL);
        setPhase('reconnecting');
        return interval;
      },
    });
  }

  const stop = () => ctrl.abort();   // 主动停止，不触发重连

  return { phase, streamText, ask, stop };
}

附录 B：Nginx 防粘连完整配置

# ========== Nginx SSE 防粘连配置 ==========
server {
    listen 80;

    location /api/chat {
        proxy_pass http://127.0.0.1:8080;

        proxy_http_version 1.1;   # 长连接 + chunked
        proxy_cache off;          # 禁缓存
        proxy_buffering off;      # 【核心】逐字节透传
        proxy_read_timeout 3600s; # 长连接不超时
        gzip off;                 # 禁压缩
        proxy_set_header Connection "";
    }
}