前端 SSE 实现详解：原理及 AI 流式对话应用

当需要向服务器发送 POST 数据时（比如发送用户输入到 AI 模型），可以使用 fetch + ReadableStream 来模拟 SSE：

const response = await fetch('/api/stream-chat', {
    method: 'POST',
    headers: {
        'Content-Type': 'application/json',
    },
    body: JSON.stringify({ userInput: inputText })
});
if (!response.ok) throw new Error(`HTTP 错误：${response.status}`);
if (!response.body) throw new Error("响应体不可用");
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 });
    const lines = buffer.split('\n\n');
    buffer = lines.pop() || '';
    for (const line of lines) {
        if (!line.startsWith('data: ')) continue;
        const dataStr = line.slice(6);
        const data = JSON.parse(dataStr);
        // 处理数据...
    }
}

四、实战：AI 聊天的流式输出实现

1. 后端实现（Express + LangChain）

app.post('/api/stream-chat', async (req, res) => {
    try {
        const { userInput } = req.body;
        if (!userInput) return res.status(400).json({ error: "用户输入不能为空" });

        // 设置 SSE 响应头
        res.setHeader('Content-Type', 'text/event-stream');
        res.setHeader('Cache-Control', 'no-cache');
        res.setHeader('Connection', 'keep-alive');
        // 立即发送响应头
        res.flushHeaders();

        // 调用 AI 模型生成回复
        const stream = await model.stream(`用户提问：${userInput}，请用简洁的语言回复`);

        // 逐块发送 AI 输出
        for await (const chunk of stream) {
            res.write(`data: ${JSON.stringify({ content: chunk?.content || chunk })}\n\n`);
        }

        // 发送结束标识
        res.write(`data: ${JSON.stringify({ done: true })}\n\n`);
        res.end();
    } catch (err) {
        console.error('Error in stream-chat:', err);
        res.write(`data: ${JSON.stringify({ error: err?.message || '服务器内部错误' })}\n\n`);
        res.end();
    }
});

2. 前端实现（fetch + ReadableStream）

// 发送请求
const response = await fetch('http://localhost:8000/api/stream-chat', {
    method: 'POST',
    headers: {
        'Content-Type': 'application/json',
    },
    body: JSON.stringify({ userInput: inputText })
});

// 检查响应
if (!response.ok) throw new Error(`HTTP 错误：${response.status}`);
if (!response.body) throw new Error("响应体不可用");

// 创建读取器和解码器
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = '';

// 处理流式数据
while (isStreaming) {
    const { done, value } = await reader.read();
    if (done) break;

    // 解码并处理 SSE 格式数据
    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split('\n\n');
    buffer = lines.pop() || '';

    // 处理每一条消息
    for (const line of lines) {
        if (!line.startsWith('data: ')) continue;
        const dataStr = line.slice(6);
        const data = JSON.parse(dataStr);

        // 处理错误信息
        if (data.error) {
            aiMsgElement.textContent = `错误：${data.error}`;
            isStreaming = false;
            break;
        }

        // 处理结束标识
        if (data.done) {
            isStreaming = false;
            break;
        }

        // 逐字显示 AI 回复
        aiMsgElement.textContent += data.content;
    }
}

五、SSE 实现的关键技术点

1. 数据格式处理

SSE 使用简单的文本格式，每个消息以 data: 开头，以 \n\n 结束。前端需要：

• 正确解析这种格式
• 处理可能分块到达的数据（使用缓冲区）
• 处理不完整的消息（保留到缓冲区）

2. 流式数据读取

使用 ReadableStream API 读取流式数据：

• getReader()：创建读取器
• read()：异步读取数据块
• done：标识流是否结束
• value：当前数据块（二进制）

3. 文本解码

使用 TextDecoder API 将二进制数据转换为字符串：

• decode(value, { stream: true })：流式解码，支持分块处理

4. 错误处理

需要处理多种错误情况：

• HTTP 错误（响应状态码非 200）
• 响应体不可用
• JSON 解析错误
• 网络中断

5. 状态管理

需要管理流式处理的状态：

• isStreaming：控制是否继续处理数据
• 停止按钮：允许用户手动中断流式传输
• 状态重置：流式结束后恢复 UI 状态

六、SSE vs WebSocket

七、SSE 的优缺点

优点：

1. 实现简单：基于 HTTP，不需要特殊的服务器支持
2. 自动重连：浏览器会自动处理重连
3. 轻量级：开销小，适合简单的实时场景
4. 兼容性好：支持所有现代浏览器
5. 易于调试：使用标准 HTTP 工具即可调试

缺点：

1. 单向通信：只能服务器向客户端推送
2. 数据格式限制：只能传输文本数据
3. 连接数限制：浏览器对同一域名的连接数有限制
4. 不支持二进制数据：需要转换为文本格式

八、适用场景

SSE 特别适合以下场景：

1. AI 聊天的流式输出：实时显示 AI 生成的回复
2. 实时通知：系统通知、消息提醒
3. 实时数据更新：股票行情、天气数据
4. 日志流：实时查看服务器日志
5. 监控数据：系统状态、性能指标

九、代码优化建议

1. 错误处理增强

try {
    // 现有代码...
} catch (error) {
    // 显示错误信息
    aiMsgElement.textContent += `\n（出错：${error.message}）`;
    // 在控制台输出错误
    console.error("流式接收错误：", error);
    // 重置状态
    isStreaming = false;
    sendBtn.disabled = false;
    stopBtn.style.display = 'none';
}

2. 性能优化

// 对于大型消息，使用 DocumentFragment 减少 DOM 操作
const fragment = document.createDocumentFragment();
const tempElement = document.createElement('div');

// 处理数据时先更新临时元素
tempElement.textContent += data.content;

// 定期更新 DOM（比如每 100ms）
if (Date.now() - lastUpdateTime > 100) {
    aiMsgElement.textContent = tempElement.textContent;
    lastUpdateTime = Date.now();
}

3. 用户体验优化

// 添加加载动画
aiMsgElement.innerHTML = '<div class="loading">生成中...</div>';

// 流式结束后移除加载动画
if (data.done) {
    aiMsgElement.innerHTML = aiMsgElement.textContent;
    isStreaming = false;
    break;
}

十、总结

SSE 是一种简单高效的服务器向客户端推送数据的技术，特别适合实时流式输出场景。通过本文的介绍，你应该已经了解了：

1. SSE 的基本概念和工作原理
2. 前端实现 SSE 的两种方式
3. 如何实现 AI 聊天的流式输出
4. SSE 的优缺点和适用场景
5. 代码优化的建议

SSE 虽然简单，但功能强大，是实时 Web 应用的重要工具之一。在实际开发中，根据具体需求选择合适的实时通信方案，才能达到最佳效果。