2259 words

11 minutes

SSE (Server-Sent Events) 在 AI-Interview 项目中的实战应用

2026-04-07

AI-Interview

RAG

/

SSE

/

FastAPI

SSE (Server-Sent Events) 在 AI-Interview 项目中的实战应用#

引言#

在 AI-Interview 项目中，我们需要在模拟面试场景下实现实时流式输出，让用户能够看到 AI 面试官逐字逐句地生成回答。这一需求促使我们选择了 SSE (Server-Sent Events) 作为核心技术方案。本文将详细记录 SSE 的技术原理、项目中的具体实现架构以及在实际应用中的经验总结。

SSE 基本概念#

SSE (Server-Sent Events) 是一种服务端推送技术，允许服务器通过 HTTP 连接向客户端自动发送实时更新。与需要额外协议的全双工 WebSocket 不同，SSE 基于标准的 HTTP/1.1 协议工作，实现更为简洁。

核心特点#

SSE 具有以下显著特性：

单向通信：仅支持服务端向客户端推送数据，客户端无法通过同一连接发送数据
基于 HTTP 长连接：利用 HTTP/1.1 的 keep-alive 机制建立持久连接
原生自动重连：浏览器内置 EventSource API 会自动处理断线重连
轻量级：相比 WebSocket 协议开销更小，数据格式更为简洁
实现简单：在 FastAPI 中仅需使用 StreamingResponse 即可实现

数据格式#

SSE 传输使用文本协议，每条消息由多行组成，以两个连续换行符作为消息结束标志：

1
event: message
2
data: {"content": "你好"}
3

4
event: progress
5
data: {"stage": "检索中"}
6
id: 42
7
retry: 5000

关键字段说明：

字段	用途
`event:`	定义事件类型，客户端可据此注册不同处理器
`data:`	消息的实际数据内容
`id:`	事件ID，浏览器自动记录，断线时发送 Last-Event-ID 头用于恢复
`retry:`	指定断连后自动重连的时间间隔（毫秒）
`\n\n`	两个换行符标志单个事件结束

SSE 与 WebSocket 对比#

在实时通信技术选型时，我们首先对比了 SSE 和 WebSocket 两种方案的优劣：

特性	SSE	WebSocket
通信方向	单向（服务端到客户端）	双向（全双工）
协议	标准 HTTP/1.1	WebSocket 特有协议 (ws://)
实现复杂度	低	中
自动重连	原生支持	需手动实现
浏览器兼容性	IE 不支持	所有现代浏览器
二进制数据支持	仅支持文本	支持二进制帧
调试便利性	可用 curl 直接测试	需 WebSocket 客户端

对于 AI-Interview 项目的 RAG 问答场景，我们最终选择了 SSE。原因是该场景本质上是请求-响应模式，用户的每次提问对应一次独立的回答流程。即使需要多轮交互，也可以通过发起新的请求来实现。SSE 方案不仅实现成本低，而且生产环境的调试和排障都更为简便。

加密机制#

需要特别说明的是，SSE 本身并不负责数据加密，加密功能由下层协议提供。当使用 HTTPS 部署时，SSE 传输全程受 TLS 保护：

1
OSI 七层模型中的位置：
2
┌─────────────────────────────┐
3
│ 7. 应用层 (SSE EventSource) │
4
├─────────────────────────────┤
5
│ 7. 应用层 (HTTP/1.1)        │
6
├─────────────────────────────┤
7
│ 6. 表示层 (TLS 1.2/1.3)      │ ← 真正负责加密
8
├─────────────────────────────┤
9
│ 5. 会话层 (TCP)              │
10
└─────────────────────────────┘

因此，确保生产环境使用 HTTPS 部署即可保证 SSE 通信的安全性。

项目架构实现#

AI-Interview 项目采用 FastAPI + sse-starlette 框架实现 SSE 功能，共设计了三个核心端点满足不同的业务场景。

端点一：获取面试问题#

GET /interview/question?stream=true 端点负责以流式方式返回面试问题，支持逐字输出效果，显著提升用户体验。

通信流程如下：

1
前端                               后端
2
  │  GET /interview/question ───►   │  发起 HTTP GET 请求
3
  │                                  │
4
  │◄── 200 OK + 长连接 ◄──────────  │  返回 StreamingResponse
5
  │◄── event: question_start ──────  │  { question_id, series, number }
6
  │◄── event: token\ndata: "请" ───  │  逐 token 流式输出
7
  │◄── event: token\ndata: "介" ───  │
8
  │◄── event: question_end ─────────  │
9
  │◄── event: end ──────────────────  │  连接关闭

端点二：提交回答并获取追问#

POST /interview/answer 是项目中最复杂的 SSE 端点，它同时承担着接收用户回答、调用 AI 模型进行评估、以及流式返回追问或下一题的职责。

完整的通信流程：

1
前端                               后端
2
  │  POST /interview/answer ──────► │  提交用户回答
3
  │◄── 200 OK + 长连接 ◄──────────   │
4
  │◄── event: evaluation ──────────  │  评估结果
5
  │◄── event: feedback ────────────  │  待发送的反馈
6
  │◄── event: question_start ──────  │  追问/下个问题的元数据
7
  │◄── event: token\ndata: "能" ───  │  逐 token 流式输出问题
8
  │◄── event: question_end ─────────  │
9
  │◄── event: end ──────────────────  │

后端实现详解#

SSE 事件生成器#

后端使用 sse_starlette 库提供的 EventSourceResponse 类来构建 SSE 响应。核心实现是异步生成器函数 event_generator：

1
from sse_starlette.sse import EventSourceResponse
2

3
async def event_generator():
4
    # 首先发送问题开始事件，包含元数据
5
    yield {"event": "question_start", "data": json.dumps({...})}
6

7
    # 然后逐 token 流式输出
8
    async for token in llm.stream():
9
        yield {"event": "token", "data": json.dumps({"content": token})}
10

11
    # 问题输出完成
12
    yield {"event": "question_end", "data": json.dumps({})}
13

14
    # 发送结束事件
15
    yield {"event": "end", "data": json.dumps({})}
16

17
return EventSourceResponse(event_generator())

生成器按顺序 yield 不同类型的 SSE 事件，每个事件包含 event 字段标识类型和 data 字段承载数据。EventSourceResponse 会将这些异步生成的事件自动转化为 HTTP 流式响应。

事件类型设计#

项目定义了以下几种 SSE 事件类型：

1
# 事件类型枚举
2
QUESTION_START = "question_start"   # 问题开始，携带问题元数据
3
TOKEN = "token"                      # 单个 token，用于流式输出
4
QUESTION_END = "question_end"        # 问题输出完毕
5
FEEDBACK = "feedback"                # AI 反馈内容
6
EVALUATION = "evaluation"             # 回答评估结果
7
END = "end"                           # SSE 连接结束

这种事件类型设计使得前端可以针对不同事件编写独立的处理逻辑，实现精细化的 UI 更新控制。

前端实现详解#

SSE 消息解析#

项目前端使用 Fetch API 配合 ReadableStream 来手动解析 SSE 数据，而非原生的 EventSource API。这样做可以获得更大的控制权，便于处理复杂的业务逻辑：

1
async function streamQuestion(apiUrl) {
2
    const response = await fetch(`${API_BASE}${apiUrl}`);
3
    const reader = response.body.getReader();
4
    const decoder = new TextDecoder();
5

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

10
        // 解码数据块，保持 UTF-8 正确性
11
        const chunk = decoder.decode(value, { stream: true });
12
        const lines = chunk.split('\n');
13

14
        // 解析 event: 和 data: 行
15
        for (const line of lines) {
16
            if (line.startsWith('event: ')) {
17
                const eventType = line.slice(7);
18
                // 根据事件类型分发处理
19
            } else if (line.startsWith('data: ')) {
20
                const data = JSON.parse(line.slice(6));
21
                // 处理数据
22
            }
23
        }
24
    }
25
}

事件分发处理#

针对不同类型的 SSE 事件，前端维护一个事件处理器映射：

1
const eventHandlers = {
2
    'question_start': (data) => {
3
        updateQuestionMeta(data);
4
        startTypingIndicator();
5
    },
6
    'token': (data) => {
7
        appendToken(data.content);
8
    },
9
    'question_end': () => {
10
        stopTypingIndicator();
11
    },
12
    'feedback': (data) => {
13
        displayFeedback(data);
14
    },
15
    'evaluation': (data) => {
16
        showEvaluationResult(data);
17
    },
18
    'end': () => {
19
        cleanup();
20
    }
21
};