从文本到图像：SSE 如何助力 AI 内容实时呈现？（Typescript篇）
#div_digg { float: right; font-size: 12px; margin: 10px; text-align: center; width: 120px; position: fixed; right: 0; bottom: 0; z-index: 10; background-color: rgba(255, 255, 255, 1); padding: 10px; border: 1px solid rgba(204, 204, 204, 1) }
#cnblogs_post_body pre code span { font-family: Consolas, monospace }
#blogTitle>h2 { font-family: Consolas, monospace }
#blog-news { font-family: Consolas, monospace }
#topics .postTitle a { font-family: Georgia, Times New Roman, Times, sans-serif, monospace; font-weight: bold }
#cnblogs_post_body p { margin: 18px auto; color: rgba(0, 0, 0, 1); font-family: Georgia, Times New Roman, Times, sans-serif, monospace; font-size: 16px; text-indent: 0 }
#cnblogs_post_body h1 { font-family: Georgia, Times New Roman, Times, sans-serif, monospace; font-size: 32px; font-weight: bold; line-height: 1.5; margin: 10px 0 }
#cnblogs_post_body h2 { font-family: Consolas, "Microsoft YaHei", monospace; font-size: 26px; font-weight: bold; line-height: 1.5; margin: 20px 0 }
#cnblogs_post_body h3 { font-family: Georgia, Times New Roman, Times, sans-serif, monospace; font-size: 20px; font-weight: bold; line-height: 1.5; margin: 10px 0 }
#cnblogs_post_body h4 { font-family: Georgia, Times New Roman, Times, sans-serif, monospace; font-size: 18px; font-weight: bold; margin: 10px 0 }
em { font-style: normal; color: rgba(0, 0, 0, 1) }
#cnblogs_post_body ul li { font-family: Georgia, Times New Roman, Times, sans-serif, monospace; color: rgba(0, 0, 0, 1); font-size: 16px; list-style-type: disc }
#cnblogs_post_body ol li { font-family: Georgia, Times New Roman, Times, sans-serif, monospace; color: rgba(0, 0, 0, 1); font-size: 16px; list-style-type: decimal }
#cnblogs_post_body a:link { text-decoration: none; color: rgba(0, 44, 153, 1) }
#topics .postBody blockquote { background: rgba(255, 243, 212, 1); border-top: none; border-right: none; border-bottom: none; border-left: 5px solid rgba(246, 183, 60, 1); margin: 0; padding-left: 10px }
.cnblogs-markdown code { font-family: Consolas, "Microsoft YaHei", monospace !important; font-size: 16px !important; line-height: 1.8; background-color: rgba(245, 245, 245, 1) !important; border: none !important; padding: 0 5px !important; border-radius: 3px !important; margin: 1px 5px; vertical-align: middle; display: inline-block }
.cnblogs-markdown .hljs { font-family: Consolas, "Microsoft YaHei", monospace !important; font-size: 16px !important; line-height: 1.5 !important; padding: 5px !important }
#cnblogs_post_body h1 code, #cnblogs_post_body h2 code { font-size: inherit !important; border: none !important }
從文本到圖像：SSE 如何助力 AI 內容實時呈現？（Typescript篇）
前言
在這個人工智能大模型日益普及的時代，AI 的能力從最初的簡單文本回復，發展到了生成圖像，甚至可以實時輸出思考過程。那么，問題來了：這些多樣化的數據是如何高效地從后端傳遞到前端的呢？今天，我們就來聊聊一種輕量級、簡單又實用的技術——SSE（Server-Sent Events）。
SSE(server-sent events)
一句話概括: SSE（Server-Sent Events）是一種基于 HTTP 的輕量級協議，允許服務端通過長連接向客戶端單向實時推送結構化文本數據流。
它有哪些特點？
- 簡單易用：前端和后端代碼實現起來非常簡單。
- 長連接：使用 HTTP 持久連接，適合持續推送數據。
- 單向通信：服務端推送，前端接收，不支持前端主動發消息。
- 輕量高效：相比 WebSocket 更加輕量。
JSON返回 vs SSE vs WebSocket 有什么區別
JSON 返回：
const response = await fetch('https://');
await response.json();流式返回：
const response = await fetch('https://');
const reader = response.body?.getReader();
while (true) {
const { value, done } = await reader.read();
}WebSocket：
const socket = new WebSocket('ws://');
socket.onopen = () => {};
socket.onmessage = () => {};| 特性 | response.json() |
ReadableStream |
WebSocket |
|---|---|---|---|
| 處理方式 | 全量讀取，自動 JSON 解析 | 按塊（chunk）逐步讀取響應體，手動處理 | 雙向通信：可持續接收和發送消息 |
| 內存占用 | 可能較高 | 較低 | 取決于消息頻率和大小，但通常開銷較低 |
| 復雜性 | 簡單 | 相對復雜 | 需要手動處理連接、消息事件、錯誤等 |
| 適用場景 | 小到中等大小 JSON 響應 | 大型文件、實時數據、非 JSON 數據 | 實時雙向通信場景，例如聊天應用、在線游戲等 |
| 實時性 | 無法實時 | 可以通過流式返回實現接近實時 | 原生支持實時通信，延遲低 |
| 協議 | HTTP | HTTP | WebSocket（基于 HTTP 升級的全雙工協議） |
| 連接狀態 | 每次請求獨立連接 | 每次請求獨立連接 | 長連接：連接建立后可持續使用 |
| 服務端推送 | 不支持 | 不支持 | 原生支持：服務端主動推送消息到客戶端 |
淺入淺出
我們通過一個簡單的例子來了解服務端如何通過 SSE 向前端推送數據。
后端代碼:
let cursor = 0;
while (cursor < text.content.length) {
const randomLength = Math.floor(Math.random() * 10) + 1;
// 從當前光標位置切片文本，生成一個塊
const chunk = text.content.slice(cursor, cursor + randomLength);
cursor += randomLength;
// 將數據塊以 SSE 格式發送到客戶端
res.write(`data: ${chunk}\n\n`);
await sleep(100);
}
// 當所有數據發送完成時，發送一個特殊的結束標記
res.write('data: [DONE]\n\n');
res.end();核心邏輯：
- 通過 res.write 向客戶端發送數據塊（以 data: 開頭，符合 SSE 格式）。
- 每次發送后稍作延遲（模擬數據生成的過程）。
- 發送完所有數據后，用 [DONE] 標記結束。
前端代碼：
const response = await fetch('/api/sse', {
method: 'POST',
});
if (!response.ok) return;
const reader = response.body?.getReader();
if (!reader) return;
// 初始化一個緩沖區，用于存儲未處理的流數據
let buffer = '';
// 創建一個 TextDecoder，用于將流數據解碼為字符串
const decoder = new TextDecoder();
while (true) {
// 從流中讀取下一個塊（chunk）
const { value, done } = await reader.read();
// 如果流讀取完成（done 為 true），退出循環
if (done) {
break;
}
if (value) {
const chunk = decoder.decode(value, { stream: true });
buffer += chunk;
// 按照雙換行符（\n\n）將緩沖區拆分為多行
let lines = buffer.split('\n\n');
// 將最后一行（可能是不完整的行）存回緩沖區，等待下一次讀取補全
buffer = lines.pop() || '';
for (const line of lines) {
// 檢查行是否以 'data: ' 開頭，這是 SSE (Server-Sent Events) 的格式
if (line.startsWith('data: ')) {
const data = line.slice(6);
// 如果接收到的是特殊標記 '[DONE]'，說明數據流結束，直接返回
if (data === '[DONE]') {
return;
}
setMessage((prev) => {
return (prev += data);
});
}
}
}
}核心邏輯：
- 通過流式讀取服務端返回的數據
- 流數據解碼為字符串并解析 SSE 數據格式
- 接收到結束標記 [DONE] 結束
有了基礎實現之后，接下來我們看看一些稍微復雜一點的場景，比如：
- 如何處理錯誤？
- 如何控制 SSE 請求的中斷？
- 如何支持更復雜的數據結構，比如 JSON 格式？圖片？
進階
- 將 SSE 返回的數據結構需改為 JSON 格式
{ "t": "返回類型", "r": "返回內容" }- 前端使用 AbortController 來控制是否結束當前請求（但是在實際使用過程中可能需要其他方案）
const response = await fetch('/api/sse', {
signal: abortController.signal,
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({ ...reqBody }),
});后端代碼：
let cursor = 0;
writeBySSE(res, { t: SSEResultType.Image, r: data.imageUrl });
while (cursor < data.think.length) {
const randomLength = Math.floor(Math.random() * 10) + 1;
const chunk = data.think.slice(cursor, cursor + randomLength);
cursor += randomLength;
if (showSSEError && cursor > showErrorCount) {
writeBySSE(res, { t: SSEResultType.Error, r: '發生錯誤！' });
res.end();
}
writeBySSE(res, { t: SSEResultType.Think, r: chunk });
await sleep(50);
}前端代碼：
for (const line of lines) {
if (line.startsWith('data: ')) {
const l = line.slice(6);
const data: SseResponseLine = JSON.parse(l);
if (data.t === SSEResultType.Image) {
setMessage((prev) => {
return { ...prev, image: data.r };
});
} else if (data.t === SSEResultType.Think) {
setMessage((prev) => {
const newThink = prev.think + data.r;
if (prev.think === newThink) return prev;
return { ...prev, think: newThink };
});
} else if (data.t === SSEResultType.Text) {
setMessage((prev) => {
const newContent = prev.content + data.r;
if (prev.content === newContent) return prev;
return { ...prev, content: newContent };
});
} else if (data.t === SSEResultType.Cancelled) {
setMessage((prev) => {
return { ...prev, isCancelled: true };
});
setIsSending(false);
} else if (data.t === SSEResultType.End) {
setIsSending(false);
} else if (data.t === SSEResultType.Error) {
setMessage((prev) => {
return { ...prev, errorMsg: data.r };
});
setIsSending(false);
}
}
}實戰：接入Deepseek大模型
源代碼地址: Github
總結
SSE 是一種簡單而有效的技術，特別適用于需要從服務器向客戶端實時推送數據的場景。相對于 WebSocket，它更加輕量，實現也更簡單。文章通過示例代碼和視頻演示，清晰地展示了 SSE 的基本原理和進階用法，以及在實際項目中的應用。
支持我們！
本文來自 Sdcb Chats 部分代碼，如果您覺得有幫助請在 GitHub 上 Star 我們！您的支持是我們前進的動力。
再次感謝您的支持，期待未來為您帶來更多驚喜！
總結
以上是生活随笔為你收集整理的从文本到图像：SSE 如何助力 AI 内容实时呈现？（Typescript篇）的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： Docker 部署ELK 日志分析
- 下一篇： 性能监测与优化命令free