當(dāng)前位置：首頁(yè) >

从文本到图像：SSE 如何助力 AI 内容实时呈现？（Typescript篇）

發(fā)布時(shí)間：2025/3/8 32 如意码农

生活随笔收集整理的這篇文章主要介紹了从文本到图像：SSE 如何助力 AI 内容实时呈现？（Typescript篇）小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

#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 內(nèi)容實(shí)時(shí)呈現(xiàn)？（Typescript篇）

前言

在這個(gè)人工智能大模型日益普及的時(shí)代，AI 的能力從最初的簡(jiǎn)單文本回復(fù)，發(fā)展到了生成圖像，甚至可以實(shí)時(shí)輸出思考過(guò)程。那么，問(wèn)題來(lái)了：這些多樣化的數(shù)據(jù)是如何高效地從后端傳遞到前端的呢？今天，我們就來(lái)聊聊一種輕量級(jí)、簡(jiǎn)單又實(shí)用的技術(shù)——SSE（Server-Sent Events）。

SSE(server-sent events)

一句話概括: SSE（Server-Sent Events）是一種基于 HTTP 的輕量級(jí)協(xié)議，允許服務(wù)端通過(guò)長(zhǎng)連接向客戶端單向?qū)崟r(shí)推送結(jié)構(gòu)化文本數(shù)據(jù)流。

它有哪些特點(diǎn)？

簡(jiǎn)單易用：前端和后端代碼實(shí)現(xiàn)起來(lái)非常簡(jiǎn)單。
長(zhǎng)連接：使用 HTTP 持久連接，適合持續(xù)推送數(shù)據(jù)。
單向通信：服務(wù)端推送，前端接收，不支持前端主動(dòng)發(fā)消息。
輕量高效：相比 WebSocket 更加輕量。

JSON返回 vs SSE vs WebSocket 有什么區(qū)別

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`
處理方式	全量讀取，自動(dòng) JSON 解析	按塊（chunk）逐步讀取響應(yīng)體，手動(dòng)處理	雙向通信：可持續(xù)接收和發(fā)送消息
內(nèi)存占用	可能較高	較低	取決于消息頻率和大小，但通常開銷較低
復(fù)雜性	簡(jiǎn)單	相對(duì)復(fù)雜	需要手動(dòng)處理連接、消息事件、錯(cuò)誤等
適用場(chǎng)景	小到中等大小 JSON 響應(yīng)	大型文件、實(shí)時(shí)數(shù)據(jù)、非 JSON 數(shù)據(jù)	實(shí)時(shí)雙向通信場(chǎng)景，例如聊天應(yīng)用、在線游戲等
實(shí)時(shí)性	無(wú)法實(shí)時(shí)	可以通過(guò)流式返回實(shí)現(xiàn)接近實(shí)時(shí)	原生支持實(shí)時(shí)通信，延遲低
協(xié)議	HTTP	HTTP	WebSocket（基于 HTTP 升級(jí)的全雙工協(xié)議）
連接狀態(tài)	每次請(qǐng)求獨(dú)立連接	每次請(qǐng)求獨(dú)立連接	長(zhǎng)連接：連接建立后可持續(xù)使用
服務(wù)端推送	不支持	不支持	原生支持：服務(wù)端主動(dòng)推送消息到客戶端

淺入淺出

我們通過(guò)一個(gè)簡(jiǎn)單的例子來(lái)了解服務(wù)端如何通過(guò) SSE 向前端推送數(shù)據(jù)。

后端代碼:

let cursor = 0;

while (cursor < text.content.length) {

  const randomLength = Math.floor(Math.random() * 10) + 1;

  // 從當(dāng)前光標(biāo)位置切片文本，生成一個(gè)塊

  const chunk = text.content.slice(cursor, cursor + randomLength);

  cursor += randomLength;

  // 將數(shù)據(jù)塊以 SSE 格式發(fā)送到客戶端

  res.write(`data: ${chunk}\n\n`);

  await sleep(100);

}

// 當(dāng)所有數(shù)據(jù)發(fā)送完成時(shí)，發(fā)送一個(gè)特殊的結(jié)束標(biāo)記

res.write('data: [DONE]\n\n');

res.end();

核心邏輯：

通過(guò) res.write 向客戶端發(fā)送數(shù)據(jù)塊（以 data: 開頭，符合 SSE 格式）。
每次發(fā)送后稍作延遲（模擬數(shù)據(jù)生成的過(guò)程）。
發(fā)送完所有數(shù)據(jù)后，用 [DONE] 標(biāo)記結(jié)束。

前端代碼：

const response = await fetch('/api/sse', {

  method: 'POST',

});

if (!response.ok) return;

const reader = response.body?.getReader();

if (!reader) return;

// 初始化一個(gè)緩沖區(qū)，用于存儲(chǔ)未處理的流數(shù)據(jù)

let buffer = '';

// 創(chuàng)建一個(gè) TextDecoder，用于將流數(shù)據(jù)解碼為字符串

const decoder = new TextDecoder();

while (true) {

  // 從流中讀取下一個(gè)塊（chunk）

  const { value, done } = await reader.read();

  // 如果流讀取完成（done 為 true），退出循環(huán)

  if (done) {

    break;

  }

  if (value) {

    const chunk = decoder.decode(value, { stream: true });

    buffer += chunk;

    // 按照雙換行符（\n\n）將緩沖區(qū)拆分為多行

    let lines = buffer.split('\n\n');

    // 將最后一行（可能是不完整的行）存回緩沖區(qū)，等待下一次讀取補(bǔ)全

    buffer = lines.pop() || '';

    for (const line of lines) {

      // 檢查行是否以 'data: ' 開頭，這是 SSE (Server-Sent Events) 的格式

      if (line.startsWith('data: ')) {

        const data = line.slice(6);

        // 如果接收到的是特殊標(biāo)記 '[DONE]'，說(shuō)明數(shù)據(jù)流結(jié)束，直接返回

        if (data === '[DONE]') {

          return;

        }

        setMessage((prev) => {

          return (prev += data);

        });

      }

    }

  }

}

核心邏輯：

通過(guò)流式讀取服務(wù)端返回的數(shù)據(jù)
流數(shù)據(jù)解碼為字符串并解析 SSE 數(shù)據(jù)格式
接收到結(jié)束標(biāo)記 [DONE] 結(jié)束

有了基礎(chǔ)實(shí)現(xiàn)之后，接下來(lái)我們看看一些稍微復(fù)雜一點(diǎn)的場(chǎng)景，比如：

如何處理錯(cuò)誤？
如何控制 SSE 請(qǐng)求的中斷？
如何支持更復(fù)雜的數(shù)據(jù)結(jié)構(gòu)，比如 JSON 格式？圖片？

進(jìn)階

將 SSE 返回的數(shù)據(jù)結(jié)構(gòu)需改為 JSON 格式

{ "t": "返回類型", "r": "返回內(nèi)容" }

前端使用 AbortController 來(lái)控制是否結(jié)束當(dāng)前請(qǐng)求（但是在實(shí)際使用過(guò)程中可能需要其他方案）

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: '發(fā)生錯(cuò)誤！' });

    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);

    }

  }

}

實(shí)戰(zhàn)：接入Deepseek大模型

源代碼地址: Github

總結(jié)

SSE 是一種簡(jiǎn)單而有效的技術(shù)，特別適用于需要從服務(wù)器向客戶端實(shí)時(shí)推送數(shù)據(jù)的場(chǎng)景。相對(duì)于 WebSocket，它更加輕量，實(shí)現(xiàn)也更簡(jiǎn)單。文章通過(guò)示例代碼和視頻演示，清晰地展示了 SSE 的基本原理和進(jìn)階用法，以及在實(shí)際項(xiàng)目中的應(yīng)用。

支持我們！

本文來(lái)自 Sdcb Chats 部分代碼，如果您覺得有幫助請(qǐng)?jiān)?GitHub 上 Star 我們！您的支持是我們前進(jìn)的動(dòng)力。

再次感謝您的支持，期待未來(lái)為您帶來(lái)更多驚喜！

總結(jié)

以上是生活随笔為你收集整理的从文本到图像：SSE 如何助力 AI 内容实时呈现？（Typescript篇）的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。

上一篇： Docker 部署ELK 日志分析
下一篇：性能监测与优化命令free