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