1. 流式輸出的本質(zhì)為什么我們需要 SSE1.1 從“等一鍋飯”到“邊炒邊上桌”的思維轉(zhuǎn)變做過大模型應(yīng)用的人都有一個共同體會用戶等一個完整回答的耐心遠(yuǎn)比我們想象的要短。早期做對話產(chǎn)品時我試過讓前端一直轉(zhuǎn)圈等后端把整段回答生成完再一次性返回結(jié)果就是超過三秒用戶就開始懷疑是不是卡死了超過五秒直接關(guān)頁面走人。這個體驗問題不是靠優(yōu)化模型推理速度能解決的因為大模型逐 token 生成的物理特性擺在那里你不可能讓一個需要生成五百字的回答在一瞬間全部蹦出來。流式輸出解決的正是這個“等待焦慮”問題。它的核心思路很簡單模型每生成一小段內(nèi)容就立刻推給前端渲染而不是攢齊了再發(fā)。用戶看到文字一個一個蹦出來哪怕總時長沒變主觀感受上也會覺得“它在思考、它在回應(yīng)”這就是所謂的打字機(jī)效果。而實現(xiàn)這種效果最成熟、最通用的底層協(xié)議就是 SSE全稱 Server-Sent Events。SSE 本質(zhì)上是一個基于 HTTP 長連接的單項推送協(xié)議?？蛻舳税l(fā)起一個普通 HTTP 請求服務(wù)端在響應(yīng)頭里聲明Content-Type: text/event-stream然后保持這個連接不關(guān)閉持續(xù)往客戶端寫數(shù)據(jù)。每一條數(shù)據(jù)以data:開頭以兩個換行符結(jié)束格式非常樸素。瀏覽器端有原生的EventSourceAPI 可以直接消費但實際項目里我們更多用fetch配合ReadableStream來手動解析因為EventSource只支持 GET 請求沒法攜帶復(fù)雜的請求體這在需要傳對話歷史的場景下是硬傷。1.2 SSE 與 WebSocket 的選型邏輯很多人一提到實時推送就想到 WebSocket覺得雙向通信肯定比單向強(qiáng)。但在大模型對話這個場景里這個想法是錯的。WebSocket 建立的是全雙工連接協(xié)議更重需要額外的握手升級過程服務(wù)端維護(hù)連接的成本也更高。而大模型對話的數(shù)據(jù)流向是典型的“客戶端發(fā)一次請求服務(wù)端持續(xù)推多次響應(yīng)”本質(zhì)上是單向的。用 WebSocket 就像為了送一趟快遞專門修了一條雙向高速公路殺雞用牛刀。SSE 的優(yōu)勢在于它復(fù)用了 HTTP 協(xié)議棧不需要額外的協(xié)議升級穿透代理和網(wǎng)關(guān)的能力更強(qiáng)斷線重連機(jī)制也是瀏覽器原生支持的。當(dāng)然它也有短板比如默認(rèn)不支持二進(jìn)制傳輸、連接數(shù)在 HTTP/1.1 下有限制但這些在大模型文本對話場景里都不是問題。我個人的經(jīng)驗是純文本流式推送用 SSE需要雙向?qū)崟r交互比如協(xié)同編輯、游戲才上 WebSocket不要為了技術(shù)時髦而過度設(shè)計。1.3 一次完整的 SSE 數(shù)據(jù)流長什么樣在動手寫代碼之前先把 SSE 的數(shù)據(jù)格式徹底搞清楚后面解析才不會踩坑。服務(wù)端推給客戶端的數(shù)據(jù)在網(wǎng)絡(luò)上實際傳輸?shù)臉幼邮沁@樣的data: {type:token,content:你} data: {type:token,content:好} data: {type:done,finish_reason:stop}注意幾個關(guān)鍵細(xì)節(jié)。第一每條消息以data:開頭冒號后面有一個空格這個空格是規(guī)范的一部分解析時要去掉。第二每條消息以兩個換行符\n\n結(jié)尾這是消息之間的分隔符。第三如果一條消息內(nèi)容很長可以分成多個data:行客戶端會把它們用換行符拼接起來。第四服務(wù)端可以發(fā)送event:字段來指定事件類型發(fā)送id:字段來標(biāo)記消息序號發(fā)送retry:字段來指定重連間隔。實際項目中OpenAI 兼容的接口返回格式通常是每個 chunk 一個 JSON里面包含choices[0].delta.content這樣的結(jié)構(gòu)。而 LangChain 的流式輸出會把這些 chunk 統(tǒng)一封裝成AIMessageChunk對象。理解這個底層格式是后面所有解析工作的基礎(chǔ)。2. LangChain 流式輸出的接入與封裝2.1 LangChain 的流式接口到底怎么用LangChain 從 0.1 版本開始對流式輸出的支持已經(jīng)相當(dāng)完善了。最基礎(chǔ)的用法是調(diào)用模型的stream方法它會返回一個生成器每次 yield 一個AIMessageChunk。我拿 OpenAI 兼容的模型舉例代碼大概長這樣from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, streamingTrue) for chunk in llm.stream(給我講講 SSE 的原理): print(chunk.content, end, flushTrue)這段代碼跑起來就能看到文字一個一個蹦出來。但這里有個坑很多人第一次用的時候發(fā)現(xiàn)還是等全部生成完才輸出原因通常是忘了在初始化時設(shè)置streamingTrue或者用錯了方法。invoke是同步阻塞的stream才是流式的astream是異步流式的。在 FastAPI 這類異步框架里一定要用astream否則會阻塞事件循環(huán)導(dǎo)致整個服務(wù)卡住。再往上一個層級如果你用的是 Chain 或者 AgentLangChain 也提供了統(tǒng)一的流式接口。Chain 有stream和astreamAgent 在 LangGraph 體系下也有對應(yīng)的流式方法。但 Agent 的流式輸出比單純 LLM 復(fù)雜得多因為它中間可能涉及工具調(diào)用、多輪推理流出來的不只是最終回答的 token還有中間步驟的事件。這個后面單獨講。2.2 把 LangChain 的 chunk 轉(zhuǎn)成 SSE 格式LangChain 的AIMessageChunk對象不能直接扔給前端必須轉(zhuǎn)成 SSE 格式的字符串。我封裝過一個通用的轉(zhuǎn)換函數(shù)核心邏輯就是把 chunk 的內(nèi)容包裝成 JSON再套上data:前綴和雙換行后綴import json def chunk_to_sse(chunk): payload { type: token, content: chunk.content, finish_reason: chunk.response_metadata.get(finish_reason) } return fdata: {json.dumps(payload, ensure_asciiFalse)}\n\n這里有幾個細(xì)節(jié)值得說。第一ensure_asciiFalse必須加否則中文會被轉(zhuǎn)義成\uXXXX的形式雖然前端也能解析但傳輸體積會變大調(diào)試時看著也難受。第二finish_reason要透傳出去前端需要知道什么時候流結(jié)束了才能關(guān)閉連接、停止 loading 動畫。第三如果 chunk 的 content 是空字符串比如第一個 chunk 通常只有 role 信息可以選擇跳過不發(fā)送減少無效傳輸。在 FastAPI 里返回 SSE 響應(yīng)用的是StreamingResponse配合一個異步生成器from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() async def event_generator(prompt: str): async for chunk in llm.astream(prompt): if chunk.content: yield chunk_to_sse(chunk) yield data: {\type\:\done\}\n\n app.get(/chat) async def chat(prompt: str): return StreamingResponse( event_generator(prompt), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no } )X-Accel-Buffering: no這個頭非常關(guān)鍵如果你前面掛了 Nginx不加這個頭 Nginx 會默認(rèn)緩沖響應(yīng)導(dǎo)致流式效果失效用戶還是等全部生成完才看到內(nèi)容。這個坑我踩過不止一次排查了半天才發(fā)現(xiàn)是網(wǎng)關(guān)層在緩沖。2.3 封裝一個可復(fù)用的 SSE 流式接口調(diào)用邏輯后端封裝好了前端消費也不能馬虎。瀏覽器原生EventSource只支持 GET傳不了復(fù)雜的請求體所以實際項目里我推薦用fetch加ReadableStream手動解析。下面是我常用的一個封裝async function streamChat(prompt, onToken, onDone) { const response await fetch(/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ prompt }) }); const reader response.body.getReader(); const decoder new TextDecoder(utf-8); 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 data JSON.parse(line.slice(6)); if (data.type token) onToken(data.content); if (data.type done) onDone(); } } }這段代碼的核心在于buffer的處理。網(wǎng)絡(luò)傳輸是分片的一個 SSE 消息可能被拆到兩個 TCP 包里所以不能假設(shè)每次read()拿到的都是完整消息。正確做法是把已接收的內(nèi)容拼到 buffer 里按\n\n切分最后一段可能不完整留在 buffer 里等下次拼接。這個細(xì)節(jié)如果處理不好會出現(xiàn) JSON 解析報錯而且報錯是偶發(fā)的特別難排查。3. 結(jié)構(gòu)化輸出讓 AI 吐出能直接用的 JSON3.1 為什么自由文本不夠用流式輸出解決了體驗問題但還有一個更根本的問題大模型默認(rèn)吐出來的是自然語言而程序需要的是結(jié)構(gòu)化數(shù)據(jù)。比如你想讓模型從一段用戶評論里提取情感傾向、關(guān)鍵詞、評分如果它返回“這段評論看起來是正面的用戶提到了物流快和服務(wù)好大概能打四星”你沒法直接拿這個結(jié)果去寫數(shù)據(jù)庫。結(jié)構(gòu)化輸出要解決的就是這個問題約束模型的輸出格式讓它返回符合特定 schema 的 JSON。LangChain 在這方面提供了好幾層工具從最簡單的PydanticOutputParser到更現(xiàn)代的with_structured_output方法各有適用場景。3.2 用 Pydantic 定義輸出 schemaPydantic 是 Python 生態(tài)里做數(shù)據(jù)校驗的事實標(biāo)準(zhǔn)LangChain 的結(jié)構(gòu)化輸出深度集成了它。定義一個 schema 非常直觀from pydantic import BaseModel, Field from typing import List class ReviewAnalysis(BaseModel): sentiment: str Field(description情感傾向只能是 positive/negative/neutral) score: int Field(description評分1 到 5 的整數(shù)) keywords: List[str] Field(description評論中提到的關(guān)鍵詞列表) summary: str Field(description一句話總結(jié))每個字段的description非常重要它不是給人看的注釋而是會作為提示詞的一部分發(fā)給模型告訴模型這個字段該填什么。description 寫得越清楚模型填錯格式的概率越低。我見過很多人 schema 定義得很隨意description 空著不寫然后抱怨模型輸出不穩(wěn)定其實問題出在自己這邊。3.3 with_structured_output 的實戰(zhàn)用法LangChain 現(xiàn)在主推的是with_structured_output方法它比老的 Parser 方案更簡潔而且底層會根據(jù)模型能力自動選擇最佳實現(xiàn)方式。對于支持 function calling 的模型它會用工具調(diào)用的方式約束輸出對于不支持的模型它會退化成提示詞約束加解析。structured_llm llm.with_structured_output(ReviewAnalysis) result structured_llm.invoke(這個產(chǎn)品太棒了物流超快客服也很耐心五星好評) print(result.sentiment) # positive print(result.score) # 5返回的result直接就是ReviewAnalysis類型的對象字段訪問用點號IDE 有自動補(bǔ)全類型檢查也能過。這比手動json.loads再取字段舒服太多了。但這里有個關(guān)鍵限制with_structured_output默認(rèn)是非流式的。因為結(jié)構(gòu)化輸出需要等模型把整個 JSON 生成完才能解析中途的片段是不完整的 JSON沒法解析。這就產(chǎn)生了一個矛盾既要結(jié)構(gòu)化又要流式打字機(jī)效果怎么辦3.4 結(jié)構(gòu)化輸出與流式的矛盾及折中方案這個矛盾的本質(zhì)是JSON 的語法要求完整性而流式輸出的特點是漸進(jìn)性。一個 JSON 對象在生成到一半的時候{sentiment: pos這樣的片段是沒法解析的。我實踐下來有三種折中方案。第一種是“先流式后結(jié)構(gòu)化”讓模型先用自然語言流式回答回答完再單獨調(diào)一次結(jié)構(gòu)化接口提取數(shù)據(jù)。缺點是調(diào)了兩次模型成本和延遲都翻倍。第二種是“流式 JSON 增量解析”用一個能容忍不完整 JSON 的解析器邊流邊嘗試解析能解析出多少算多少。這種方案技術(shù)含量高但體驗最好。第三種是“字段級流式”把結(jié)構(gòu)化輸出拆成多個字段每個字段單獨流式生成前端按字段逐個渲染。我目前項目里用得最多的是第二種配合一個叫partial-json-parser的庫它能解析不完整的 JSON 片段返回已經(jīng)完整的部分。比如{sentiment: positive, score:這樣的片段它能解析出{sentiment: positive}。前端拿到部分?jǐn)?shù)據(jù)就能先渲染等完整了再補(bǔ)全。4. 打字機(jī)效果的前端實現(xiàn)細(xì)節(jié)4.1 逐字渲染還是逐塊渲染后端推過來的 chunk 粒度是不固定的有時候一個 chunk 是一個字有時候是一整句。如果直接按 chunk 渲染會出現(xiàn)“有時候一個字一個字蹦有時候一整句突然出現(xiàn)”的不均勻感。要做出絲滑的打字機(jī)效果前端需要做一層緩沖和勻速輸出。我的做法是維護(hù)一個待渲染隊列后端每來一個 chunk 就入隊然后用requestAnimationFrame或者setInterval以固定速度從隊列里取字符渲染。這樣無論后端推得快還是慢視覺上都是勻速的。速度一般控制在每幀 1 到 3 個字符太快了沒有打字感太慢了用戶著急。let queue ; let rendering false; function enqueue(text) { queue text; if (!rendering) renderLoop(); } function renderLoop() { rendering true; if (queue.length 0) { rendering false; return; } const char queue[0]; queue queue.slice(1); outputElement.textContent char; setTimeout(renderLoop, 30); }這個 30 毫秒的間隔是調(diào)出來的經(jīng)驗值對應(yīng)大約每秒 33 個字符接近正常人閱讀速度看起來比較自然。4.2 自動滾動與用戶打斷的處理打字機(jī)效果還有一個容易被忽略的細(xì)節(jié)自動滾動。內(nèi)容越來越多容器要自動滾到底部否則用戶得手動往下拉。但這里有個坑如果用戶主動往上滾動去看之前的內(nèi)容你還強(qiáng)制滾到底部用戶會很煩躁。正確做法是判斷當(dāng)前滾動位置只有當(dāng)用戶已經(jīng)在底部附近時才自動滾動。function autoScroll() { const el document.getElementById(chat-container); const isAtBottom el.scrollHeight - el.scrollTop - el.clientHeight 50; if (isAtBottom) { el.scrollTop el.scrollHeight; } }這個 50 像素的閾值也是經(jīng)驗值太小了稍微滾一點就觸發(fā)太大了用戶滾上去了還會被拉下來。4.3 流中斷與異常狀態(tài)的 UI 反饋流式輸出最怕的就是中途斷了。網(wǎng)絡(luò)抖動、服務(wù)端超時、模型報錯都可能導(dǎo)致流中斷。這時候前端不能一直轉(zhuǎn)圈等必須給用戶明確的反饋。我在實際項目里遇到過stream disconnected before completion: idle timeout waiting for sse這個報錯原因是服務(wù)端超過一定時間沒有推送任何數(shù)據(jù)網(wǎng)關(guān)判定連接空閑就掐斷了。解決辦法有兩個一是服務(wù)端定期發(fā)送心跳注釋以:開頭的行客戶端會忽略保持連接活躍二是前端設(shè)置超時檢測超過一定時間沒收到數(shù)據(jù)就主動斷開并提示用戶重試。async def event_generator(prompt: str): last_heartbeat time.time() async for chunk in llm.astream(prompt): if chunk.content: yield chunk_to_sse(chunk) if time.time() - last_heartbeat 15: yield : heartbeat\n\n last_heartbeat time.time() yield data: {\type\:\done\}\n\n心跳間隔設(shè) 15 秒比較穩(wěn)妥大部分網(wǎng)關(guān)的空閑超時都在 30 秒以上留一半余量。5. 常見問題排查與避坑實錄5.1 流式失效的排查思路流式失效是最常見的問題表現(xiàn)就是用戶等半天然后所有內(nèi)容一次性出現(xiàn)。排查要按鏈路逐段確認(rèn)。先確認(rèn)模型層是不是真的在流式可以在后端加日志看astream是不是逐個 yield 的。如果模型層沒問題再確認(rèn) FastAPI 的StreamingResponse有沒有被中間件緩沖。最后確認(rèn)網(wǎng)關(guān)層Nginx 需要關(guān)proxy_buffering加X-Accel-Buffering: no頭。下面這張表是我整理的排查清單按順序過一遍基本能定位問題排查環(huán)節(jié)檢查項常見問題模型層是否用 stream/astream誤用 invoke 導(dǎo)致阻塞框架層StreamingResponse 配置media_type 寫錯中間件是否有緩沖中間件GZip 中間件會緩沖網(wǎng)關(guān)層Nginx 緩沖配置proxy_buffering 默認(rèn)開前端層是否正確解析流按 chunk 而非按消息解析5.2 JSON 解析失敗的典型場景結(jié)構(gòu)化輸出解析失敗十有八九是模型輸出的 JSON 不合法。常見的有多了 markdown 代碼塊標(biāo)記json 包裹、字段類型不對該是整數(shù)給了字符串、缺少必填字段、JSON 后面跟了多余的解釋文字。LangChain 的解析器對 markdown 代碼塊標(biāo)記有一定容錯但類型錯誤和缺字段是沒法自動修復(fù)的。我的經(jīng)驗是在 schema 的 description 里把約束寫死比如“只返回 JSON不要有任何其他文字”、“score 必須是 1 到 5 的整數(shù)不要加引號”。另外可以用with_structured_output的strictTrue參數(shù)讓底層用更嚴(yán)格的約束。5.3 中文亂碼與編碼問題中文亂碼通常出在兩個地方。一是后端json.dumps沒加ensure_asciiFalse導(dǎo)致中文被轉(zhuǎn)義雖然前端能解析但看著別扭。二是前端TextDecoder沒指定utf-8或者解碼時沒加{ stream: true }參數(shù)導(dǎo)致多字節(jié)字符被截斷。{ stream: true }這個參數(shù)特別重要。UTF-8 編碼的中文一個字占三個字節(jié)如果網(wǎng)絡(luò)分片正好切在一個字的中間不加這個參數(shù)就會解碼出亂碼。加了之后TextDecoder會把不完整的字節(jié)序列緩存起來等下一個分片到了再一起解碼。5.4 并發(fā)場景下的連接管理多個用戶同時對話時每個用戶一個 SSE 連接服務(wù)端要維護(hù)大量長連接。這里要注意幾個點。一是連接要有超時機(jī)制用戶關(guān)了頁面但連接沒斷的情況很常見需要服務(wù)端定期清理。二是要限制單用戶的最大并發(fā)連接數(shù)防止惡意占用。三是如果用異步框架確保生成器里沒有阻塞操作否則會拖垮整個事件循環(huán)。我在一個項目里遇到過連接泄漏原因是用戶關(guān)閉頁面后后端的生成器還在跑因為模型還在生成。解決辦法是在生成器里檢測客戶端斷開FastAPI 里可以通過request.is_disconnected()來判斷斷開就停止生成釋放資源。6. 從單輪到多輪Agent 場景下的流式挑戰(zhàn)6.1 Agent 流式輸出的特殊性前面講的都是單輪對話的流式Agent 場景要復(fù)雜得多。一個 Agent 處理用戶請求時可能先思考、再調(diào)用工具、拿到結(jié)果再思考、最后才給出回答。這個過程中用戶希望看到的不只是最終回答還有中間的推理步驟和工具調(diào)用狀態(tài)這樣才有“AI 在干活”的感知。LangGraph 體系下Agent 的流式輸出有幾種模式。values模式每次輸出完整狀態(tài)updates模式只輸出變化的部分messages模式專門輸出消息 token。實際項目里我通常用messages模式拿 token 流同時用updates模式拿工具調(diào)用事件兩者結(jié)合給用戶完整的反饋。6.2 工具調(diào)用事件的透傳工具調(diào)用是 Agent 的特色也是流式處理的難點。當(dāng) Agent 決定調(diào)用某個工具時流里會出現(xiàn)一個帶有tool_calls的 chunk這時候前端應(yīng)該顯示“正在調(diào)用 XX 工具”的提示而不是繼續(xù)渲染文字。async for event in agent.astream_events(input, versionv2): kind event[event] if kind on_chat_model_stream: chunk event[data][chunk] if chunk.content: yield chunk_to_sse(chunk) elif kind on_tool_start: yield fdata: {{\type\:\tool_start\,\name\:\{event[name]}\}}\n\n elif kind on_tool_end: yield fdata: {{\type\:\tool_end\,\name\:\{event[name]}\}}\n\nastream_events是 LangChain 提供的統(tǒng)一事件流接口能拿到模型流、工具開始、工具結(jié)束等各種事件。用這個接口就不用自己去猜 chunk 的類型了事件類型是明確的。6.3 多輪對話歷史的流式處理多輪對話時每次請求都要把歷史消息帶上。歷史消息可能很長如果每次都全量傳輸請求體會很大。我的做法是后端維護(hù)會話狀態(tài)前端只傳一個 session_id后端根據(jù) id 取出歷史。這樣請求體小也避免了歷史被篡改的風(fēng)險。但會話狀態(tài)存哪里是個問題。存內(nèi)存最簡單但服務(wù)重啟就丟了多實例部署也不共享。存 Redis 是更穩(wěn)妥的方案設(shè)置合理的過期時間比如 30 分鐘無活動就清理。如果對話很重要不能丟那就得落庫但落庫會增加延遲需要權(quán)衡。7. 性能優(yōu)化與生產(chǎn)環(huán)境注意事項7.1 減少首字延遲首字延遲是流式體驗的關(guān)鍵指標(biāo)用戶從點擊發(fā)送到看到第一個字的時間超過一秒就會覺得慢。影響首字延遲的因素有幾個模型本身的推理啟動時間、網(wǎng)絡(luò)往返、后端處理邏輯。優(yōu)化手段上模型層可以選更快的模型或者用推理加速服務(wù)。網(wǎng)絡(luò)層可以把服務(wù)部署在離用戶近的區(qū)域。后端層要確保在調(diào)用模型之前沒有耗時操作比如查數(shù)據(jù)庫、做復(fù)雜計算這些都應(yīng)該提前做好或者異步做。我見過有人在生成器里先查一次用戶信息再調(diào)模型白白增加了幾百毫秒延遲。7.2 背壓與流量控制流式輸出是服務(wù)端推、客戶端收如果客戶端消費慢服務(wù)端推得快數(shù)據(jù)就會在緩沖區(qū)堆積。Python 的異步生成器天然有背壓機(jī)制yield會等待消費者取走才繼續(xù)所以一般不用擔(dān)心。但如果中間加了隊列做緩沖就要注意隊列長度限制防止內(nèi)存暴漲。7.3 日志與可觀測性生產(chǎn)環(huán)境一定要有完善的日志。每次請求記錄請求 id、用戶 id、prompt 長度、首字延遲、總時長、token 數(shù)、是否異常中斷。這些數(shù)據(jù)是排查問題和優(yōu)化性能的基礎(chǔ)。我習(xí)慣在 SSE 流里也帶上請求 id前端報錯時可以把 id 給到后端直接定位到具體那次請求的日志。另外要監(jiān)控異常中斷率如果這個指標(biāo)突然升高說明可能有網(wǎng)絡(luò)問題或者服務(wù)端問題。中斷率超過 5% 就值得警惕了。8. 我踩過的幾個印象深刻的坑第一個坑是 Nginx 緩沖。本地開發(fā)一切正常部署到測試環(huán)境流式就失效了排查了一下午才發(fā)現(xiàn)是 Nginx 默認(rèn)開啟了proxy_buffering。這個坑的教訓(xùn)是流式應(yīng)用部署時網(wǎng)關(guān)層的配置一定要單獨確認(rèn)不能假設(shè)默認(rèn)配置就是對的。第二個坑是TextDecoder的stream參數(shù)。前端偶爾出現(xiàn)亂碼特別是中文概率大概百分之幾。查了很久才定位到是解碼時沒加{ stream: true }導(dǎo)致多字節(jié)字符被網(wǎng)絡(luò)分片截斷。這個 bug 的隱蔽性在于它是概率性的本地測試很難復(fù)現(xiàn)。第三個坑是結(jié)構(gòu)化輸出的流式矛盾。一開始我想當(dāng)然地以為with_structured_output也能流式結(jié)果發(fā)現(xiàn)它內(nèi)部是等完整 JSON 才返回的。后來改用增量 JSON 解析才解決。這個坑讓我明白不是所有 LangChain 的方法都支持流式用之前要確認(rèn)清楚。第四個坑是連接泄漏。用戶關(guān)閉頁面后后端生成器還在跑因為模型還在生成生成器不知道客戶端已經(jīng)走了。時間一長大量僵尸連接占滿資源。解決辦法是在生成器循環(huán)里定期檢查request.is_disconnected()斷開就break。這些坑的共同點是文檔里不會寫只有真正上手做才會遇到。所以我的建議是流式應(yīng)用一定要在接近生產(chǎn)的環(huán)境里充分測試本地跑通不代表線上沒問題。