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