1. 被誤讀的 SSE它從來就不只是流式輸出很多人第一次接觸 SSE是在做 AI 對(duì)話界面的時(shí)候。前端要一個(gè)字一個(gè)字往外蹦后端接口不能一次性返回于是搜索流式輸出搜到了EventSource抄了一段代碼跑通了然后就把 SSE 歸類成AI 打字機(jī)效果的實(shí)現(xiàn)方式。這個(gè)認(rèn)知不能說錯(cuò)但實(shí)在太窄了。SSE 的全稱是Server-Sent Events它是 HTTP 協(xié)議體系里一個(gè)正式的、有規(guī)范定義的推送機(jī)制。瀏覽器端用EventSource接口去消費(fèi)服務(wù)端按照text/event-stream的格式持續(xù)寫數(shù)據(jù)。它解決的問題本質(zhì)上是在一條長(zhǎng)連接上服務(wù)端如何主動(dòng)、持續(xù)、有序地把事件推給客戶端。流式輸出只是它最顯眼的一種應(yīng)用形態(tài)而不是它的全部。我見過太多項(xiàng)目把 SSE 用成了偽流式——后端攢一批數(shù)據(jù)發(fā)一次前端收到后整段替換看起來在動(dòng)實(shí)際上延遲很高、體驗(yàn)很糙。也見過有人把 SSE 和 WebSocket 混為一談?dòng)X得能推消息就是一樣的。這兩者在連接模型、協(xié)議開銷、重連機(jī)制、適用場(chǎng)景上差別很大選錯(cuò)了會(huì)在后期付出代價(jià)。這篇文章想做的事情很明確把 SSE 從協(xié)議層到應(yīng)用層講透把EventSource的使用邊界講清楚再落到 LangChain 這類框架的流式輸出上看看工業(yè)級(jí)的實(shí)時(shí)推送到底是怎么搭起來的。中間會(huì)穿插 Node 環(huán)境相關(guān)的實(shí)操細(xì)節(jié)因?yàn)楹芏嗫悠鋵?shí)出在環(huán)境和工具鏈上而不是 SSE 本身。適合已經(jīng)寫過一點(diǎn)前端、想認(rèn)真搞懂實(shí)時(shí)推送的開發(fā)者也適合正在做 AI 應(yīng)用、被流式輸出折磨過的同學(xué)。2. SSE 的協(xié)議本質(zhì)一條被低估的長(zhǎng)連接2.1 它到底長(zhǎng)什么樣SSE 的報(bào)文格式簡(jiǎn)單到有點(diǎn)樸素。服務(wù)端返回的 Content-Type 是text/event-stream然后按照固定格式往連接里寫文本event: message data: {content: 你} data: {content: 好} id: 1 retry: 3000幾個(gè)關(guān)鍵字段需要記牢data消息體可以多行多行會(huì)被拼接成一個(gè)字符串行與行之間用換行符連接。event事件類型名前端可以用addEventListener監(jiān)聽自定義事件不寫就是默認(rèn)的message。id事件 ID瀏覽器會(huì)自動(dòng)記錄最后收到的 ID斷線重連時(shí)會(huì)通過Last-Event-ID請(qǐng)求頭帶回去。retry重連間隔毫秒數(shù)告訴瀏覽器斷線后多久重試。以:開頭的行是注釋常用來做心跳?；?。這里有個(gè)容易被忽略的點(diǎn)每條消息之間必須用空行分隔。我見過有人寫服務(wù)端代碼時(shí)忘了這個(gè)空行結(jié)果前端一直收不到消息排查半天以為是跨域問題。實(shí)際上 SSE 的解析器是靠空行來判定一條消息結(jié)束的沒有空行數(shù)據(jù)就一直掛在緩沖區(qū)里。2.2 為什么它比輪詢和 WebSocket 更適合某些場(chǎng)景先做個(gè)對(duì)比把三種常見方案擺在一起看維度短輪詢WebSocketSSE連接方向客戶端反復(fù)請(qǐng)求全雙工服務(wù)端單向推送協(xié)議HTTP獨(dú)立協(xié)議握手用 HTTPHTTP自動(dòng)重連無需自己實(shí)現(xiàn)瀏覽器內(nèi)置斷點(diǎn)續(xù)傳無需自己實(shí)現(xiàn)內(nèi)置 Last-Event-ID代理/網(wǎng)關(guān)兼容好一般需配置升級(jí)好就是普通 HTTP數(shù)據(jù)格式任意二進(jìn)制/文本文本適用場(chǎng)景低頻狀態(tài)查詢雙向高頻交互服務(wù)端單向推送SSE 最大的優(yōu)勢(shì)在于它就是普通 HTTP。這意味著它天然穿過大多數(shù)代理、網(wǎng)關(guān)、負(fù)載均衡器不需要額外的協(xié)議升級(jí)配置。而 WebSocket 在很多企業(yè)網(wǎng)絡(luò)環(huán)境里會(huì)被中間設(shè)備攔截或降級(jí)配置起來麻煩得多。另一個(gè)被低估的能力是自動(dòng)重連和斷點(diǎn)續(xù)傳。瀏覽器發(fā)現(xiàn)連接斷了會(huì)按照retry指定的間隔自動(dòng)重連并且把最后收到的id通過Last-Event-ID頭帶回去。服務(wù)端只要根據(jù)這個(gè) ID 把斷線期間的消息補(bǔ)發(fā)就實(shí)現(xiàn)了消息不丟。這個(gè)能力在 WebSocket 里是要自己從零實(shí)現(xiàn)的。但 SSE 的短板也很明顯單向。客戶端不能通過這條連接發(fā)消息給服務(wù)端要發(fā)只能另開一個(gè) HTTP 請(qǐng)求。所以聊天、協(xié)作編輯這類需要雙向高頻通信的場(chǎng)景WebSocket 更合適。而通知推送、日志流、AI 流式輸出、進(jìn)度上報(bào)這類服務(wù)端說、客戶端聽的場(chǎng)景SSE 是更省事的選擇。2.3 連接數(shù)限制這個(gè)老問題瀏覽器對(duì)同域名下的 HTTP/1.1 連接數(shù)有限制通常是 6 個(gè)。SSE 會(huì)長(zhǎng)期占用其中一個(gè)。如果你在一個(gè)頁面里開了多個(gè)EventSource很容易把連接池占滿導(dǎo)致其他請(qǐng)求排隊(duì)。解決辦法有兩個(gè)一是升級(jí)到 HTTP/2多路復(fù)用讓連接數(shù)限制基本消失二是多個(gè)數(shù)據(jù)流復(fù)用同一個(gè) SSE 連接通過event字段區(qū)分不同業(yè)務(wù)類型。第二種做法在后端稍微復(fù)雜一點(diǎn)但前端只需要一個(gè)EventSource實(shí)例管理起來更清爽。提示如果你的應(yīng)用還在 HTTP/1.1 上跑并且頁面里同時(shí)有多個(gè)實(shí)時(shí)數(shù)據(jù)源優(yōu)先考慮合并成一條 SSE 連接用事件類型分流而不是開多個(gè) EventSource。3. EventSource 的實(shí)戰(zhàn)邊界能做什么不能做什么3.1 基礎(chǔ)用法和那些必須知道的限制EventSource的 API 簡(jiǎn)單得讓人放松警惕const es new EventSource(/api/stream); es.onmessage (e) { console.log(收到:, e.data); }; es.addEventListener(progress, (e) { console.log(進(jìn)度事件:, e.data); }); es.onerror (err) { console.error(出錯(cuò)了, err); };但它的限制同樣明顯而且每一條都可能在實(shí)際項(xiàng)目里咬你一口不能自定義請(qǐng)求頭。EventSource構(gòu)造函數(shù)只接受 URL 和可選的withCredentials你沒法加Authorization頭。這意味著基于 Token 的鑒權(quán)方案要么改用 Cookie要么把 Token 放到 URL 參數(shù)里不推薦會(huì)進(jìn)日志要么用 polyfill 庫(kù)。只能發(fā) GET 請(qǐng)求。想傳復(fù)雜參數(shù)只能拼在 URL 上長(zhǎng)度受限語義也差。不能主動(dòng)關(guān)閉后重連。調(diào)用es.close()之后瀏覽器不會(huì)再自動(dòng)重連。如果你希望暫停一下再恢復(fù)得自己重新 new 一個(gè)。錯(cuò)誤處理很粗糙。onerror拿不到 HTTP 狀態(tài)碼你無法區(qū)分是 401 鑒權(quán)失敗還是 500 服務(wù)端錯(cuò)誤只能靠重連行為間接判斷。3.2 用 fetch 替代 EventSource 的取舍因?yàn)樯厦孢@些限制很多項(xiàng)目干脆放棄EventSource改用fetchReadableStream手動(dòng)解析 SSE 格式。這樣能自定義請(qǐng)求頭、能用 POST、能拿到完整的狀態(tài)碼。const resp await fetch(/api/stream, { method: POST, headers: { Content-Type: application/json, Authorization: Bearer ${token} }, body: JSON.stringify({ query: 你好 }) }); const reader resp.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); buffer lines.pop(); for (const line of lines) { if (line.startsWith(data:)) { const data line.slice(5).trim(); if (data) handleMessage(data); } } }代價(jià)是你要自己實(shí)現(xiàn)重連、自己處理 Last-Event-ID、自己解析事件邊界。這段代碼看著不長(zhǎng)但邊界情況很多分片可能把一個(gè)多字節(jié)字符切斷data:后面可能有也可能沒有空格空行判定要準(zhǔn)確。我踩過的坑是decoder.decode沒傳{ stream: true }導(dǎo)致中文被截?cái)喑蓙y碼排查了很久。所以選型邏輯很清楚如果鑒權(quán)簡(jiǎn)單、只需要 GET、能接受瀏覽器默認(rèn)重連行為用 EventSource如果需要自定義頭、POST、精細(xì)控制重連用 fetch 手動(dòng)解析。沒有絕對(duì)優(yōu)劣看場(chǎng)景。3.3 一個(gè)真實(shí)的標(biāo)簽返回不完整排查過程熱詞里有個(gè)標(biāo)簽返回未完整怎么處理這個(gè)我太有共鳴了。做 AI 流式輸出時(shí)模型返回的內(nèi)容里帶結(jié)構(gòu)化標(biāo)簽比如answer.../answer前端按標(biāo)簽解析渲染。結(jié)果經(jīng)常出現(xiàn)標(biāo)簽只收到一半的情況ans就斷了或者/answer缺了最后的。排查鏈路是這樣的第一步先確認(rèn)是傳輸層截?cái)噙€是解析層截?cái)唷Ｔ趏nmessage里把原始e.data打出來如果原始數(shù)據(jù)本身就是半截標(biāo)簽?zāi)鞘欠?wù)端分片的問題如果原始數(shù)據(jù)完整但渲染出來是半截那是前端解析邏輯的問題。第二步如果是傳輸層檢查服務(wù)端的 flush 時(shí)機(jī)。Node 里用res.write()之后如果沒有觸發(fā) flush數(shù)據(jù)會(huì)攢在緩沖區(qū)里。SSE 場(chǎng)景下通常要配合res.flushHeaders()和確保沒有中間件在做壓縮緩沖。第三步如果是解析層問題往往出在按 chunk 邊界解析。流式數(shù)據(jù)到達(dá)的邊界和標(biāo)簽邊界沒有任何關(guān)系一個(gè)標(biāo)簽可能橫跨三個(gè) chunk。正確做法是維護(hù)一個(gè)緩沖區(qū)每次收到新數(shù)據(jù)就追加然后嘗試匹配完整的標(biāo)簽匹配不到就等下一批。let tagBuffer ; function processChunk(chunk) { tagBuffer chunk; const regex /answer([\s\S]*?)\/answer/g; let match; let lastIndex 0; while ((match regex.exec(tagBuffer)) ! null) { renderAnswer(match[1]); lastIndex regex.lastIndex; } tagBuffer tagBuffer.slice(lastIndex); }這個(gè)模式的核心思想是永遠(yuǎn)不要假設(shè)一次收到的數(shù)據(jù)是一個(gè)完整語義單元。流式處理的本質(zhì)就是和邊界不確定性打交道。4. LangChain 流式輸出框架幫你做了什么又藏了什么4.1 從 LLM 到前端的完整鏈路LangChain 的流式輸出不是單一環(huán)節(jié)的事它是一條鏈路模型 API 的流式響應(yīng) → LangChain 的回調(diào)/流式接口 → 你的服務(wù)端 → SSE → 前端渲染。任何一環(huán)沒打通前端都看不到逐字效果。先看模型層。主流模型 API 都支持stream: true返回的是一系列 chunk每個(gè) chunk 包含一小段增量文本。LangChain 把這層封裝成了stream()方法from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, streamingTrue) for chunk in llm.stream(講個(gè)笑話): print(chunk.content, end, flushTrue)注意streamingTrue這個(gè)參數(shù)。有些模型封裝默認(rèn)不開流式你不顯式打開stream()拿到的可能是一次性返回的完整結(jié)果。這個(gè)坑我在項(xiàng)目里遇到過前端一直不蹦字查了半天發(fā)現(xiàn)是模型初始化時(shí)漏了這個(gè)參數(shù)。再往上一層是 Chain 和 Agent。LangChain 提供了astream和astream_events兩個(gè)接口。astream返回的是最終輸出的流astream_events返回的是全鏈路事件流包括 LLM 開始、LLM 結(jié)束、工具調(diào)用、檢索等中間步驟。做工業(yè)智能體的時(shí)候astream_events更有價(jià)值因?yàn)槟隳馨颜跈z索知識(shí)庫(kù)正在調(diào)用工具這些狀態(tài)也推給前端。4.2 服務(wù)端怎么把 LangChain 的流接到 SSE這是整個(gè)鏈路里最容易出問題的地方。Python 側(cè)用 FastAPI 舉例from fastapi import FastAPI from fastapi.responses import StreamingResponse from langchain_openai import ChatOpenAI app FastAPI() llm ChatOpenAI(modelgpt-4o-mini, streamingTrue) async def event_generator(query: str): async for chunk in llm.astream(query): if chunk.content: yield fdata: {chunk.content}\n\n yield data: [DONE]\n\n app.get(/api/stream) async def stream(query: str): return StreamingResponse( event_generator(query), media_typetext/event-stream, headers{ Cache-Control: no-cache, X-Accel-Buffering: no, } )幾個(gè)關(guān)鍵點(diǎn)值得展開X-Accel-Buffering: no這個(gè)頭是給 Nginx 看的告訴它不要緩沖這個(gè)響應(yīng)。沒有這個(gè)頭Nginx 默認(rèn)會(huì)攢夠一定大小才轉(zhuǎn)發(fā)前端就會(huì)看到卡一下蹦一大段的效果。這個(gè)坑非常隱蔽本地開發(fā)直連沒問題一上生產(chǎn)經(jīng)過 Nginx 就出問題。media_typetext/event-stream必須設(shè)置否則瀏覽器不會(huì)按 SSE 解析。[DONE]結(jié)束標(biāo)記是 OpenAI 風(fēng)格的約定前端收到它就主動(dòng)close()避免連接一直掛著。異步生成器要用async for因?yàn)?LangChain 的astream是異步的。如果你用同步的stream配異步框架會(huì)阻塞事件循環(huán)多個(gè)用戶同時(shí)請(qǐng)求時(shí)直接卡死。4.3 前端消費(fèi) LangChain 流的完整寫法前端這塊如果用EventSource因?yàn)椴荒馨l(fā) POST參數(shù)只能拼 URL。更常見的做法是用 fetch 手動(dòng)解析前面已經(jīng)給過代碼。這里補(bǔ)充一個(gè)完整的、帶錯(cuò)誤處理和結(jié)束判定的版本async function chat(query, onChunk, onDone) { const controller new AbortController(); const resp await fetch(/api/stream?query encodeURIComponent(query), { signal: controller.signal }); if (!resp.ok) { throw new Error(HTTP ${resp.status}); } const reader resp.body.getReader(); const decoder new TextDecoder(); let buffer ; try { while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); let idx; while ((idx buffer.indexOf(\n\n)) ! -1) { const rawEvent buffer.slice(0, idx); buffer buffer.slice(idx 2); const dataLines rawEvent .split(\n) .filter(l l.startsWith(data:)) .map(l l.slice(5).trimStart()); const data dataLines.join(\n); if (data [DONE]) { onDone(); return; } onChunk(data); } } } finally { controller.abort(); } }這段代碼里buffer.indexOf(\n\n)是核心——按空行切分事件而不是按 chunk 切分。這是 SSE 解析的正確姿勢(shì)。很多人圖省事直接對(duì)每個(gè) chunk 做split(data:)遇到跨 chunk 的事件就崩了。5. 環(huán)境與工具鏈那些和 SSE 無關(guān)卻總在拖后腿的坑5.1 Node 環(huán)境安裝的常見翻車現(xiàn)場(chǎng)熱詞里 Node 相關(guān)的詞占了很大比例說明很多人在環(huán)境這一步就卡住了。這些坑和 SSE 本身沒關(guān)系但會(huì)直接影響你能不能跑通 demo。Windows 上npm.ps1無法加載。報(bào)錯(cuò)信息是因?yàn)樵诖讼到y(tǒng)上禁止運(yùn)行腳本。這是 PowerShell 的執(zhí)行策略問題不是 Node 的問題。解決辦法是以管理員身份運(yùn)行 PowerShell執(zhí)行Set-ExecutionPolicy RemoteSigned然后確認(rèn)?；蛘吒挠?CMD 而不是 PowerShell。nvm 切換 Node 版本后全局包丟失。這是 nvm 的設(shè)計(jì)使然每個(gè) Node 版本有獨(dú)立的全局包目錄。切換版本后需要重新npm install -g你需要的工具。如果嫌麻煩可以用nvm reinstall-packages從舊版本遷移。國(guó)內(nèi)鏡像配置。默認(rèn)源下載慢是常態(tài)配置鏡像能省很多時(shí)間npm config set registry https://registry.npmmirror.comLinux 離線安裝。內(nèi)網(wǎng)環(huán)境沒法直接下載需要提前在有網(wǎng)機(jī)器上下好對(duì)應(yīng)架構(gòu)的二進(jìn)制包解壓后配置 PATH。注意 glibc 版本兼容性高版本 Node 可能依賴較新的 glibc。node-gyp和 Node 版本對(duì)應(yīng)。編譯原生模塊時(shí)node-gyp需要匹配的 Python 和編譯工具鏈。Node 18 和 Node 20 對(duì) Python 版本要求不同裝錯(cuò)了會(huì)報(bào)一堆看不懂的錯(cuò)。建議用nvm管理版本遇到原生模塊編譯問題先確認(rèn) Node 版本。5.2 流式場(chǎng)景下的分片上傳報(bào)錯(cuò)熱詞里有個(gè)node 分片上傳文件時(shí)報(bào)錯(cuò) request aborted。這個(gè)和 SSE 是兩回事但都涉及流式處理值得說一句。request aborted通常意味著客戶端在服務(wù)端還沒讀完流的時(shí)候就斷開了連接。常見原因客戶端超時(shí)設(shè)置太短、服務(wù)端處理太慢、中間代理有超時(shí)限制。排查思路是先在服務(wù)端監(jiān)聽req.on(aborted)和req.on(close)確認(rèn)是客戶端主動(dòng)斷還是服務(wù)端出錯(cuò)。如果是代理超時(shí)調(diào)整代理的proxy_read_timeout。如果是客戶端超時(shí)檢查 fetch 或 axios 的 timeout 配置。注意SSE 長(zhǎng)連接同樣會(huì)遇到代理超時(shí)問題。默認(rèn)情況下很多代理 60 秒沒數(shù)據(jù)就斷連。解決辦法是服務(wù)端定期發(fā)送心跳注釋行: ping\n\n保持連接活躍。5.3 一個(gè)容易被忽略的心跳機(jī)制SSE 連接如果長(zhǎng)時(shí)間沒有數(shù)據(jù)中間的網(wǎng)絡(luò)設(shè)備可能認(rèn)為連接空閑而斷開。瀏覽器雖然會(huì)自動(dòng)重連但頻繁重連會(huì)造成消息重復(fù)或丟失。標(biāo)準(zhǔn)做法是服務(wù)端每隔 15 到 30 秒發(fā)一個(gè)心跳const heartbeat setInterval(() { res.write(: heartbeat\n\n); }, 20000); req.on(close, () { clearInterval(heartbeat); });以:開頭的行是 SSE 注釋瀏覽器會(huì)忽略它但連接保持活躍。這個(gè)細(xì)節(jié)在本地開發(fā)時(shí)完全看不出問題一上生產(chǎn)就暴露。6. 選型與演進(jìn)SSE、WebSocket、LangGraph 該怎么擺6.1 SSE 和 WebSocket 不是二選一很多人糾結(jié)到底用哪個(gè)其實(shí)它們可以共存。一個(gè)典型的 AI 應(yīng)用可能是這樣對(duì)話流式輸出用 SSE單向、簡(jiǎn)單、穿透好協(xié)同編輯或?qū)崟r(shí)雙向控制用 WebSocket雙向、低延遲。按功能模塊選而不是全站統(tǒng)一。判斷標(biāo)準(zhǔn)可以簡(jiǎn)化成一句話如果客戶端不需要在這條連接上主動(dòng)發(fā)消息優(yōu)先 SSE。6.2 LangChain 和 LangGraph 的關(guān)系熱詞里反復(fù)出現(xiàn)langchain 和 langgraph 的區(qū)別這里說清楚。LangChain 是基礎(chǔ)框架提供模型封裝、Chain、工具、檢索等組件。LangGraph 是構(gòu)建在 LangChain 之上的狀態(tài)機(jī)式編排框架用來處理有循環(huán)、有分支、有狀態(tài)的復(fù)雜 Agent 流程。簡(jiǎn)單說LangChain 適合線性的、簡(jiǎn)單的調(diào)用鏈LangGraph 適合需要多輪循環(huán)、條件分支、人工介入的復(fù)雜智能體。做工業(yè)智能體時(shí)如果流程里有調(diào)用工具 → 判斷結(jié)果 → 決定是否重試這種循環(huán)LangGraph 更合適。至于它們是不是過時(shí)了我的看法是框架會(huì)演進(jìn)但底層的流式輸出、事件驅(qū)動(dòng)、狀態(tài)管理這些概念不會(huì)過時(shí)。學(xué)框架的同時(shí)理解底層機(jī)制換框架時(shí)遷移成本才低。6.3 流式輸出的性能與體驗(yàn)權(quán)衡最后說一個(gè)實(shí)操心得。流式輸出不是越快越好。如果模型吐字太快前端每個(gè) chunk 都觸發(fā)一次 DOM 更新會(huì)造成大量重排頁面反而卡頓。常見的優(yōu)化是批量渲染用一個(gè)緩沖區(qū)攢一小段時(shí)間比如 50ms的 chunk然后一次性更新 DOM。這樣既保留了逐字效果又減少了渲染壓力。let pending ; let timer null; function scheduleRender(text) { pending text; if (timer) return; timer setTimeout(() { renderToDOM(pending); pending ; timer null; }, 50); }這個(gè) 50ms 是個(gè)經(jīng)驗(yàn)值可以根據(jù)實(shí)際體驗(yàn)調(diào)整。太小了沒效果太大了會(huì)感覺卡頓。我在實(shí)際項(xiàng)目里踩過的另一個(gè)坑是流式輸出和 Markdown 渲染結(jié)合時(shí)未閉合的語法會(huì)導(dǎo)致渲染錯(cuò)亂。比如模型正在輸出一個(gè)代碼塊三個(gè)反引號(hào)只到了兩個(gè)Markdown 解析器會(huì)把后面的內(nèi)容全當(dāng)成代碼。解決辦法是渲染前對(duì)未閉合的語法做容錯(cuò)處理或者用支持增量渲染的 Markdown 庫(kù)。這套東西搭下來你會(huì)發(fā)現(xiàn) SSE 本身其實(shí)很簡(jiǎn)單難的是把它和框架、環(huán)境、前端渲染、網(wǎng)絡(luò)中間件正確地串起來。每一個(gè)環(huán)節(jié)都有它自己的脾氣而經(jīng)驗(yàn)就是被這些脾氣磨出來的。