構(gòu)化返回實(shí)戰(zhàn)：SSE與OutputParser協(xié)同指南)
1. 為什么流式輸出和結(jié)構(gòu)化返回總是打架做 GenAI 應(yīng)用做了這幾年我最大的感受是流式輸出和結(jié)構(gòu)化返回天然就是一對(duì)矛盾體。大模型默認(rèn)吐出來的是自然語言你要它一段一段地流式返回給前端打字機(jī)效果又要它在最后給出一個(gè)能被程序直接校驗(yàn)、入庫、調(diào)用工具的 JSON 結(jié)構(gòu)這中間如果不做設(shè)計(jì)十有八九會(huì)翻車。先說一個(gè)最典型的場景你用 FastAPI 起了一個(gè) Agent 服務(wù)前端用打字機(jī)效果展示大模型的回答同時(shí)后端還要把是否調(diào)用了某個(gè)工具工具參數(shù)是什么最終結(jié)果字段有哪些解析出來用于日志審計(jì)和業(yè)務(wù)判斷。如果只簡單地把模型輸出整段塞給前端前端拿到的是一堆夾雜著廢話的純文本如果只等全部生成完再一次性返回流式體驗(yàn)就沒了。SSEServer-Sent Events恰好是這兩者之間的橋——它能保證文本一塊一塊地推給前端又能在每個(gè)事件塊里攜帶結(jié)構(gòu)化元數(shù)據(jù)。這篇文章我就完整復(fù)盤一下怎么用 LangChain 的三大 OutputParser 配合 ToolCall在 SSE 流式場景下既保體驗(yàn)、又保結(jié)構(gòu)化。這篇文章適合誰看已經(jīng)在用 LangChain 寫 Agent、但被流式解析折磨過的后端工程師或者正準(zhǔn)備把 LLM 能力封裝成標(biāo)準(zhǔn) API 服務(wù)、又不想丟掉打字機(jī)效果的前后端同學(xué)。我會(huì)把每一步的取舍和原理都講清楚不是讓你照抄代碼而是讓你下次遇到類似需求時(shí)能自己拍板選型。2. 先搞清楚 SSE 在這個(gè)場景里到底扮演什么角色2.1 SSE 和 WebSocket為什么聊天場景常選 SSE很多人一想到實(shí)時(shí)推送就默認(rèn) WebSocket但在大模型應(yīng)用里SSE 往往是更務(wù)實(shí)的選擇。SSE 是單向的服務(wù)器往客戶端推數(shù)據(jù)客戶端不需要也不應(yīng)該頻繁回傳。這和 LLM 生成的場景天然匹配——用戶提問之后剩下的就是模型一直說前端一直聽。協(xié)議層面SSE 就是普通的 HTTP 響應(yīng)Content-Type 設(shè)為text/event-stream然后在響應(yīng)體里按固定格式寫事件event: message data: {type: token, content: 你} event: message data: {type: token, content: 好} event: done data: {type: done, session_id: abc123}每個(gè)事件之間用空行隔開data字段是真正的載荷event是事件類型id可以用來做斷點(diǎn)續(xù)傳retry是客戶端自動(dòng)重連的時(shí)間間隔。就這么簡單。WebSocket 呢它是全雙工要握手、要維護(hù)長連接心跳、要處理斷線重連邏輯在只需要服務(wù)器單向推送的場景里屬于殺雞用牛刀。更現(xiàn)實(shí)的問題很多企業(yè)內(nèi)網(wǎng)的網(wǎng)關(guān)、Nginx 配置對(duì) WebSocket 的升級(jí)請求支持不友好但對(duì) SSE 這種普通 HTTP 長響應(yīng)基本上零成本透明轉(zhuǎn)發(fā)。我在實(shí)際項(xiàng)目里用 SSE 遇到過的最大坑反而很簡單網(wǎng)關(guān)的超時(shí)時(shí)間設(shè)太短模型思考超過 60 秒連接直接被掐斷前端就收到一個(gè)stream disconnected before completion: idle timeout waiting for sse。這個(gè)問題后面在排查章節(jié)我會(huì)詳細(xì)說。2.2 從 LangChain 的流式機(jī)制到 SSE 事件拼裝LangChain 從Runnable體系開始把流式能力統(tǒng)一成了stream/astream兩個(gè)接口。chain.stream(query)會(huì)一塊一塊地yield輸出。需要注意的是這個(gè)一塊不一定是模型吐的一個(gè) token而是 LangChain 每個(gè)Runnable步驟產(chǎn)出的一個(gè)完整單元。比如Retriever步驟吐出文檔列表LLM步驟吐出 token 片段。在 Agent 場景下中間可能還夾著Tool的執(zhí)行結(jié)果。所以設(shè)計(jì) SSE 接口時(shí)我的做法是先約定一套內(nèi)部事件協(xié)議而不是直接把 token 裸推出去。每個(gè)事件我用 JSON 串里面至少帶兩個(gè)字段type和payload。event: message data: {type: start, payload: {session_id: uuid}} event: message data: {type: token, payload: {content: 你好}} event: message data: {type: tool_call, payload: {name: search_news, args: {keyword: 人工智能}}} event: message data: {type: tool_result, payload: {result: ..., duration_ms: 1200}} event: message data: {type: structured, payload: {title: ..., summary: ...}} event: message data: {type: done, payload: {finish_reason: stop}}前端只需要根據(jù)type決定怎么渲染token追加到正文tool_call可以展示正在調(diào)用工具的動(dòng)畫structured存到表單里做后續(xù)業(yè)務(wù)。這樣流式體驗(yàn)和結(jié)構(gòu)化數(shù)據(jù)就各歸其位了。后面我講的三大 OutputParser本質(zhì)都是在最后那個(gè)結(jié)構(gòu)化事件這個(gè)環(huán)節(jié)里保證你拿到的payload是干凈、可校驗(yàn)的 JSON。3. 三大 OutputParser把模型的話轉(zhuǎn)成程序能用的結(jié)構(gòu)OutputParser 在 LangChain 里的定位是從 LLM 的原始輸出里抽取出程序需要的結(jié)構(gòu)。注意它通常不是魔法它靠的是提示詞約束 格式校驗(yàn)兩步走。讓模型按指定格式輸出再用解析器校驗(yàn)、糾錯(cuò)。理解了這一點(diǎn)你就能明白為什么換個(gè)模型解析失敗中文環(huán)境下 JSON 不標(biāo)準(zhǔn)這類問題會(huì)反復(fù)出現(xiàn)了。3.1 PydanticOutputParser給 JSON 上一份類型保票PydanticOutputParser是項(xiàng)目里最常用的一個(gè)。它結(jié)合了 Pydantic 的BaseModel把輸出格式約束成明確的字段類型——字符串、整型、列表、嵌套對(duì)象類型不對(duì)直接校驗(yàn)失敗。用法上核心三步定義模型類、創(chuàng)建解析器、把格式指令塞進(jìn)提示詞。from typing import List from pydantic import BaseModel, Field from langchain.output_parsers import PydanticOutputParser class ArticleSummary(BaseModel): title: str Field(description生成的文章標(biāo)題不超過20字) keywords: List[str] Field(description3-5個(gè)關(guān)鍵詞) summary: str Field(description100字以內(nèi)的核心摘要) confidence: float Field(description模型對(duì)摘要質(zhì)量的自信度0到1之間) parser PydanticOutputParser(pydantic_objectArticleSummary) prompt PromptTemplate( template請分析下面這段文本并嚴(yán)格按照格式要求輸出。\n文本{text}\n{format_instructions}\n, input_variables[text], partial_variables{format_instructions: parser.get_format_instructions()}, )parser.get_format_instructions()生成的那段指令本質(zhì)是把你定義的字段、類型、約束翻譯成自然語言模板告訴模型必須輸出一個(gè) JSONkey 有哪些value 是什么類型。它還會(huì)補(bǔ)充一句不要輸出其他內(nèi)容之類的強(qiáng)調(diào)。實(shí)際效果上模型越強(qiáng)GPT-4 級(jí)別、Claude 3.5遵循度越高小模型經(jīng)常把 JSON 包在 Markdown 代碼塊里或者多解釋一句。解析器的parse方法還內(nèi)置了糾錯(cuò)能力如果模型輸出的文本可以被eval成 JSON但類型不對(duì)它會(huì)嘗試用 LLM 自動(dòng)修復(fù)。不過這個(gè)修復(fù)是有損的——它需要額外調(diào)一次模型速度和成本都要考慮。所以在流式場景里我通常不在最后階段用自動(dòng)修復(fù)而是做兩段式先流式展示再后端靜默校驗(yàn)校驗(yàn)失敗才觸發(fā)修復(fù)。3.2 StructuredOutputParser輕量到不需要定義類StructuredOutputParser適合那種不想建 Pydantic 模型只要幾個(gè)簡單字段的場景。它通過ResponseSchema列表來聲明字段名、類型和描述使用起來比 Pydantic 版更輕。from langchain.output_parsers import StructuredOutputParser, ResponseSchema response_schemas [ ResponseSchema(nameanswer, description對(duì)問題的直接回答, typestring), ResponseSchema(namesource, description答案的參考來源如果沒有則為null, typestring), ] parser StructuredOutputParser.from_response_schemas(response_schemas)從源碼實(shí)現(xiàn)看StructuredOutputParser內(nèi)部并沒有把輸出嚴(yán)格轉(zhuǎn)成 Pydantic 對(duì)象而是返回一個(gè)字典。它對(duì)字段順序、缺失字段的處理更寬松底層用的其實(shí)是類似正則 字典提取的簡易邏輯。所以它適合內(nèi)部接口、日志記錄、字段不多的場景一旦你的下游真的要用強(qiáng)類型做入?yún)⑿ｒ?yàn)還是 Pydantic 版更省心。這里有個(gè)經(jīng)驗(yàn)如果返回結(jié)構(gòu)里嵌套層級(jí)很深或者字段會(huì)因?yàn)槟Ｐ洼敵鲲L(fēng)格波動(dòng)我建議直接用 Pydantic 版不要在 Structured 版上強(qiáng)行造輪子。輕量方案省下的代碼量會(huì)在排障時(shí)加倍還回去。3.3 JsonOutputParser最容易被低估的那個(gè)第三個(gè)其實(shí)是JsonOutputParser—— LangChain 里專門用來只要 JSON不要類定義的解析器。它和PydanticOutputParser最大的區(qū)別是不要求目標(biāo)類型是 Pydantic 模型你用普通 dict 聲明一個(gè)期望的 JSON 結(jié)構(gòu)模板它就能照這個(gè)模板校驗(yàn)。from langchain.output_parsers import JsonOutputParser parser JsonOutputParser() prompt PromptTemplate( template輸出JSON格式結(jié)果字段包括: title(string), items(array of string)。\n{format_instructions}\n, input_variables[], partial_variables{format_instructions: parser.get_format_instructions()}, )它做的事情是拿到模型文本提取并解析出 JSON 對(duì)象。它不會(huì)去做嚴(yán)格類型強(qiáng)轉(zhuǎn)保持了 dict 的靈活性。實(shí)際項(xiàng)目中我經(jīng)常把它作為流式過程中增量解析 JSON 的工具——不是等模型完整輸出后一次性解析而是配合字節(jié)流每次拿到新的 token 片段就去試著解析能解析出部分字段就先緩存。這個(gè)思路在長回答場景里特別有用你可以在模型還在生成正文時(shí)就把標(biāo)題、關(guān)鍵列表等輪廓字段提前推給前端?？偨Y(jié)一下三者的選擇邏輯要強(qiáng)類型校驗(yàn)、下游要嚴(yán)格入庫選 Pydantic只要幾個(gè)字段、隨拿隨用選 Structured既要 JSON 又不想綁定模型定義、或需要在流中做增量解析選 Json。沒有絕對(duì)好壞只看約束強(qiáng)度。4. ToolCall 方案讓模型把工具意圖直接交出來4.1 為什么不用讓模型自己拼工具調(diào)用文本早期 LangChain 的 Agent 實(shí)現(xiàn)里模型是用純文本的方式假裝調(diào)用工具——輸出一行Action: search_news\nAction Input: 人工智能然后 AgentExecutor 去解析這段文本。這種方案在模型能力弱的時(shí)候還算勉強(qiáng)能用但問題很明顯模型一旦在文本里多加一句解釋、少寫一個(gè)換行整個(gè)解析就崩了。而且文本格式因模型而異換模型就要調(diào)解析規(guī)則?，F(xiàn)在主流方案是ToolCall函數(shù)調(diào)用。模型在生成時(shí)除了輸出自然語言還可以輸出一個(gè)結(jié)構(gòu)化的工具調(diào)用意圖——方法名、參數(shù) JSON。OpenAI 的 function calling、Claude 的 tool use、國產(chǎn)模型不少也兼容這個(gè)協(xié)議。LangChain 里的做法是把工具定義綁定到模型上這一類模型能力稱為bind_tools。from langchain_openai import ChatOpenAI from langchain_core.tools import tool tool def search_news(keyword: str, limit: int 5) - list: 搜索新聞資訊keyword為關(guān)鍵詞limit為返回條數(shù)。 # 這里寫真實(shí)檢索邏輯 return [{title: 示例新聞, url: https://example.com}] llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools([search_news]) response llm_with_tools.invoke(幫我搜一下今天人工智能領(lǐng)域的新聞)關(guān)鍵在于response是一個(gè)AIMessage如果模型決定調(diào)用工具它的tool_calls屬性里會(huì)帶上結(jié)構(gòu)化調(diào)用信息response.tool_calls # [{name: search_news, args: {keyword: 人工智能, limit: 5}, id: call_xxx}]這個(gè)id字段很重要尤其是做并發(fā)工具調(diào)用時(shí)它用來關(guān)聯(lián)工具結(jié)果和對(duì)應(yīng)的調(diào)用請求。4.2 ToolCall 事件在 SSE 里怎么推既然AIMessage.tool_calls是結(jié)構(gòu)化的那么從流式事件角度它也能流式地分片到達(dá)——模型先生成工具名再一點(diǎn)一點(diǎn)生成參數(shù) JSON。LangChain 的astream_events可以讓你捕獲on_chat_model_stream事件從而拿到 token 級(jí)的流。但我要提醒你工具參數(shù)這種 JSON前端完全沒必要做打字機(jī)效果。你只要在tool_call開始事件里推一條正在調(diào)用工具等工具結(jié)果出來再推一條結(jié)構(gòu)化結(jié)果就行參數(shù) JSON 在中間過程可以直接攢在后端。這是我的實(shí)踐結(jié)論——不要一上來就把所有 token 都推給前端做逐字渲染那樣只會(huì)讓前端渲染邏輯又復(fù)雜又容易出錯(cuò)。工具執(zhí)行完你拿到的結(jié)果同樣建議包裝成結(jié)構(gòu)化事件推給前端。同時(shí)把工具結(jié)果作為新的上下文消息再喂回給模型讓它基于結(jié)果生成最終回答。這個(gè)模型→工具→模型的循環(huán)如果自己用for循環(huán)寫很容易在異常分支和超時(shí)控制上出問題——這也是為什么存在 LangGraph 這類帶狀態(tài)編排的框架。不過對(duì)于單輪工具調(diào)用場景手動(dòng)循環(huán)完全可控不需要上重型框架。4.3 OutputParser 和 ToolCall 怎么配合這就是這個(gè)方案的精髓了ToolCall 解決模型要調(diào)用什么工具、參數(shù)是什么的結(jié)構(gòu)化提取OutputParser 解決模型最終要返回給業(yè)務(wù)的最終結(jié)論的結(jié)構(gòu)化提取。兩者是在一次請求的不同階段各司其職。class FinalAnswer(BaseModel): reply: str Field(description面向用戶的最終回答) used_tools: List[str] Field(description本次實(shí)際使用到的工具名稱列表) data_source: List[str] Field(description參考信息的來源列表) final_parser PydanticOutputParser(pydantic_objectFinalAnswer)流程大致是用戶提問 → 模型決定調(diào)用工具ToolCall 結(jié)構(gòu)化→ 執(zhí)行工具 → 把工具結(jié)果拼進(jìn)上下文 → 模型生成最終回答OutputParser 結(jié)構(gòu)化→ 通過 SSE 推送給前端。中間環(huán)節(jié)的結(jié)構(gòu)化靠tool_calls最后的業(yè)務(wù)結(jié)構(gòu)靠 OutputParser。兩條線涇渭分明誰也不會(huì)干擾誰。5. FastAPI LangChain 完整落地一條 SSE 接口打通全流程5.1 服務(wù)端異步流式接口的分層設(shè)計(jì)我強(qiáng)烈建議把模型調(diào)用邏輯和HTTP 流式協(xié)議分開。模型調(diào)用邏輯是一個(gè)普通的異步生成器它只負(fù)責(zé)產(chǎn)出結(jié)構(gòu)化事件字典HTTP 層只負(fù)責(zé)把事件字典按 SSE 協(xié)議編碼。這樣拆開你可以對(duì)模型邏輯單測不用每次起服務(wù)。from fastapi import FastAPI from fastapi.responses import StreamingResponse import json, asyncio from langchain_openai import ChatOpenAI from langchain_core.output_parsers import PydanticOutputParser from pydantic import BaseModel, Field from typing import List app FastAPI() llm ChatOpenAI(modelgpt-4o-mini, temperature0.3, streamingTrue) class FinalAnswer(BaseModel): reply: str Field(description面向用戶的最終回答) keywords: List[str] Field(description3-5個(gè)關(guān)鍵詞) parser PydanticOutputParser(pydantic_objectFinalAnswer) async def event_generator(prompt: str): # 1. 自動(dòng)構(gòu)建帶格式約束的提示詞 formatted_prompt ( 請回答用戶問題并輸出嚴(yán)格JSON。\n問題{q}\n{fmt}\n ).format(qprompt, fmtparser.get_format_instructions()) # 2. 第一個(gè)事件告知開始 yield { event: message, data: json.dumps({type: start, payload: {time: asyncio.time()}}, ensure_asciiFalse) } # 3. 流式輸出 token 事件 collected async for chunk in llm.astream(formatted_prompt): collected chunk.content yield { event: message, data: json.dumps({type: token, payload: {content: chunk.content}}, ensure_asciiFalse) } # 控制推送節(jié)奏避免瞬間把積壓的token全倒出去 await asyncio.sleep(0) # 4. 結(jié)構(gòu)化解析并推給前端 try: parsed parser.parse(collected) yield { event: message, data: json.dumps({type: structured, payload: parsed.model_dump()}, ensure_asciiFalse) } except Exception as exc: yield { event: message, data: json.dumps({type: parse_error, payload: {error: str(exc)}}, ensure_asciiFalse) } # 5. 結(jié)束事件 yield { event: message, data: json.dumps({type: done, payload: {finish_reason: stop}}, ensure_asciiFalse) } app.post(/chat/stream) async def chat_stream(payload: dict): prompt payload.get(prompt, ) return StreamingResponse( event_generator(prompt), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, # 重要禁止Nginx緩沖 }, )X-Accel-Buffering: no這個(gè) header是很多人在 Nginx 反代下 SSE 不流式的元兇。Nginx 默認(rèn)會(huì)緩沖響應(yīng)攢滿 4KB 或者等連接結(jié)束才發(fā)給前端你明明在服務(wù)端yield了前端卻半天沒動(dòng)靜。加了這個(gè) header 就是明確告訴 Nginx 別緩沖。5.2 前端事件分發(fā)與渲染解耦前端用fetch配合ReadableStream解析 SSE 就夠了不一定非要引eventsource-parser這類庫但引了確實(shí)省事。核心邏輯是讀到一行data:解析 JSON按type走不同的渲染函數(shù)。async function streamChat(prompt) { const resp await fetch(/chat/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ prompt }), }); const reader resp.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { value, done } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); // SSE事件以空行分隔 let sepIndex; while ((sepIndex buffer.indexOf(\n\n)) ! -1) { const rawEvent buffer.slice(0, sepIndex); buffer buffer.slice(sepIndex 2); const dataLine rawEvent.split(\n).find(line line.startsWith(data:)); if (!dataLine) continue; const msg JSON.parse(dataLine.slice(5).trim()); handleEvent(msg); } } } function handleEvent(msg) { switch (msg.type) { case token: appendText(msg.payload.content); break; case structured: fillMetaPanel(msg.payload); break; case tool_call: showToolIndicator(msg.payload.name); break; case done: stopLoading(); break; } }注意千萬別直接用瀏覽器的原生EventSource—— 它只支持 GET 請求而我們往往需要 POST 傳遞 prompt。原生EventSource也沒有自定義 header 的能力鑒權(quán)都麻煩。用fetch流式讀取是最通用的方案。5.3 一邊流式一邊結(jié)構(gòu)化增量解析的實(shí)踐前面提到的JsonOutputParser增量解析在實(shí)際項(xiàng)目中可以這樣用每收到一段新 token就把collected追加后嘗試parser.parse如果解析成功哪怕還不完整只要能出部分字段就把部分結(jié)果推給結(jié)構(gòu)化預(yù)覽事件如果解析失敗因?yàn)?JSON 還沒閉合忽略即可不算錯(cuò)誤。這個(gè)模式我用來解決一個(gè)具體的痛點(diǎn)用戶問幫我總結(jié)這份文檔并給出三個(gè)要點(diǎn)模型正文還沒寫完我希望前端右側(cè)欄已經(jīng)先把要點(diǎn)標(biāo)題渲染出來。雖然嚴(yán)格說最終結(jié)果要以最后完整解析為準(zhǔn)但增量預(yù)覽的體驗(yàn)提升非常明顯。代價(jià)是每次追加 token 都會(huì)觸發(fā)一次 JSON 解析token 非常長時(shí)會(huì)有少量 CPU 開銷。我實(shí)測下來對(duì)普通問答長度的文本這個(gè)開銷可以忽略。6. 常見問題與排查技巧實(shí)錄6.1 SSE 流中途斷開空閑超時(shí)是頭號(hào)殺手提示stream disconnected before completion: idle timeout waiting for sse這類報(bào)錯(cuò)絕大多數(shù)不是代碼問題是鏈路中的代理/網(wǎng)關(guān)配置問題。我遇到過最典型的三層排查順序第一層本地測試。先用curl -N直接打服務(wù)接口觀察事件是不是正常持續(xù)輸出。curl -N能實(shí)時(shí)打印服務(wù)器推來的每個(gè)事件如果這一步正常問題就不在后端。第二層查反向代理。Nginx 的proxy_read_timeout默認(rèn) 60 秒模型思考時(shí)間一旦超過代理直接斷連。調(diào)大或用proxy_read_timeout 300s可以緩解。另外確認(rèn)proxy_buffering off;或X-Accel-Buffering: no已生效。第三層查云廠商網(wǎng)關(guān)。很多云負(fù)載均衡器對(duì)長連接也有空閑超時(shí)限制比如 60 秒、120 秒。盡量用 WebSocket 或 SSE 都能走的長連接配置同時(shí)后端在流式傳輸過程中即使沒有數(shù)據(jù)也要定期發(fā)一個(gè): ping注釋行作為心跳。SSE 規(guī)范里以冒號(hào)開頭的行是注釋客戶端會(huì)忽略它但能刷新代理的空閑計(jì)時(shí)器。async def keepalive(): while True: yield : ping\n\n await asyncio.sleep(15)把這個(gè)生成器和主事件生成器用asyncio.gather合并就能在模型長時(shí)間思考時(shí)維持連接活性。6.2 OutputParser 拿到半截 JSON或 Markdown 代碼塊模型輸出里常見的臟格式有兩種一是把 JSON 藏在json代碼塊里二是前后夾帶解釋文字。PydanticOutputParser本身會(huì)嘗試從文本里提取 JSON 塊但并不可靠。我的兜底方案是寫一個(gè)基礎(chǔ)清洗函數(shù)在喂給解析器之前先做預(yù)處理。import re, json def extract_json_string(text: str) - str: text text.strip() # 去掉首尾的 markdown 代碼塊標(biāo)記 code_block_pattern re.compile(r(?:json)?\s*(.*?)\s*, re.DOTALL) match code_block_pattern.search(text) if match: return match.group(1) # 嘗試從第一個(gè) { 到最后一個(gè) } 截取 start text.find({) end text.rfind(}) if start ! -1 and end ! -1 and end start: return text[start:end1] return text然后統(tǒng)一走parse。注意如果清洗后還是解析失敗再去觸發(fā) LLM 原文本修復(fù)。一定不要默認(rèn)讓每個(gè)失敗都走修復(fù)不然成本和延遲都會(huì)失控。6.3 并發(fā)請求下事件錯(cuò)亂上下文變量與隊(duì)列如果你的 FastAPI 服務(wù)同時(shí)處理多個(gè) SSE 會(huì)話每個(gè)會(huì)話的生成器是獨(dú)立的理論上不會(huì)串。但我踩過一個(gè)實(shí)際的坑在生成器內(nèi)部用了模塊級(jí)的全局變量緩存工具結(jié)果兩個(gè)用戶同時(shí)觸發(fā)同一個(gè)工具調(diào)用時(shí)A 用戶的結(jié)果可能被 B 用戶覆蓋。解決方案很簡單每個(gè)會(huì)話的事件生成器必須是自包含的所有狀態(tài)都放在生成器內(nèi)部不要依賴模塊級(jí)可變對(duì)象。需要跨函數(shù)傳狀態(tài)就用contextvars或者干脆把 session_id 作為 key 放進(jìn)一個(gè)字典管理隊(duì)列。我在項(xiàng)目里用的模式是每個(gè)會(huì)話一個(gè)asyncio.Queue生成器往隊(duì)列放事件SSE 層從隊(duì)列取事件編碼輸出。這個(gè)抽象能讓你在后續(xù)擴(kuò)展多 Agent 編排時(shí)游刃有余。7. 一些我踩過坑之后的固定習(xí)慣先說工具聲明。LangChain 的tool裝飾器會(huì)讀取函數(shù)的 docstring 和類型注解來生成工具的 schema。docstring 里的描述、參數(shù)的類型提示、默認(rèn)值都會(huì)成為傳給模型的 tool schema 的一部分。所以我在寫工具函數(shù)時(shí)會(huì)強(qiáng)制自己把每個(gè)參數(shù)的單位、取值范圍、邊界情況寫進(jìn) docstring這不是為了寫注釋好看而是直接決定模型能不能正確填參。一個(gè)只寫 keyword: 搜索關(guān)鍵詞 的工具和一個(gè)寫著 keyword: 搜索關(guān)鍵詞最長20字符不要帶引號(hào) 的工具在模型調(diào)參準(zhǔn)確率上差很多。然后是解析器的temperature。做結(jié)構(gòu)化輸出時(shí)模型溫度不建議設(shè)太高0 到 0.3 之間最穩(wěn)。溫度高了模型更容易發(fā)揮創(chuàng)造力去改格式、加注釋這對(duì)我們的 JSON 解析是災(zāi)難。如果既要?jiǎng)?chuàng)意又要結(jié)構(gòu)化我一般拆兩條鏈一條低溫度出結(jié)構(gòu)化摘要一條高溫度潤色成自然語言回復(fù)最后再把兩段結(jié)果拼進(jìn) SSE 事件里。最后再分享一個(gè)小技巧給事件協(xié)議加一個(gè)trace_id字段。每個(gè) SSE 會(huì)話生成一個(gè)trace_id在start事件里發(fā)給前端同時(shí)在服務(wù)端日志里打出來。前端報(bào) bug 時(shí)直接甩這個(gè) ID 給你你能在日志里把整條鏈路還原出來省掉大量你剛才問的什么來著的溝通成本。我在生產(chǎn)環(huán)境靠這個(gè)字段排查過很多偶發(fā)問題尤其是模型偶發(fā)出錯(cuò)那種玄學(xué)問題有 trace_id 才能對(duì)上號(hào)。這套方案跑穩(wěn)之后你會(huì)發(fā)現(xiàn)流式體驗(yàn)和結(jié)構(gòu)化返回其實(shí)不是二選一關(guān)鍵是讓它們各走各的通道文本走 token 事件結(jié)構(gòu)化走獨(dú)立事件中間用統(tǒng)一的 JSON 協(xié)議串聯(lián)。理解了這層設(shè)計(jì)后續(xù)換模型、加工具、上 LangGraph 編排都不會(huì)再被到底是文本還是 JSON這個(gè)問題卡住。 LangChain流式輸出與結(jié)構(gòu)化返回實(shí)戰(zhàn)：SSE與OutputParser協(xié)同指南
1. 為什么流式輸出和結(jié)構(gòu)化返回總是打架做 GenAI 應(yīng)用做了這幾年我最大的感受是流式輸出和結(jié)構(gòu)化返回天然就是一對(duì)矛盾體。大模型默認(rèn)吐出來的是自然語言你要它一段一段地流式返回給前端打字機(jī)效果又要它在最后給出一個(gè)能被程序直接校驗(yàn)、入庫、調(diào)用工具的 JSON 結(jié)構(gòu)這中間如果不做設(shè)計(jì)十有八九會(huì)翻車。先說一個(gè)最典型的場景你用 FastAPI 起了一個(gè) Agent 服務(wù)前端用打字機(jī)效果展示大模型的回答同時(shí)后端還要把是否調(diào)用了某個(gè)工具工具參數(shù)是什么最終結(jié)果字段有哪些解析出來用于日志審計(jì)和業(yè)務(wù)判斷。如果只簡單地把模型輸出整段塞給前端前端拿到的是一堆夾雜著廢話的純文本如果只等全部生成完再一次性返回流式體驗(yàn)就沒了。SSEServer-Sent Events恰好是這兩者之間的橋——它能保證文本一塊一塊地推給前端又能在每個(gè)事件塊里攜帶結(jié)構(gòu)化元數(shù)據(jù)。這篇文章我就完整復(fù)盤一下怎么用 LangChain 的三大 OutputParser 配合 ToolCall在 SSE 流式場景下既保體驗(yàn)、又保結(jié)構(gòu)化。這篇文章適合誰看已經(jīng)在用 LangChain 寫 Agent、但被流式解析折磨過的后端工程師或者正準(zhǔn)備把 LLM 能力封裝成標(biāo)準(zhǔn) API 服務(wù)、又不想丟掉打字機(jī)效果的前后端同學(xué)。我會(huì)把每一步的取舍和原理都講清楚不是讓你照抄代碼而是讓你下次遇到類似需求時(shí)能自己拍板選型。2. 先搞清楚 SSE 在這個(gè)場景里到底扮演什么角色2.1 SSE 和 WebSocket為什么聊天場景常選 SSE很多人一想到實(shí)時(shí)推送就默認(rèn) WebSocket但在大模型應(yīng)用里SSE 往往是更務(wù)實(shí)的選擇。SSE 是單向的服務(wù)器往客戶端推數(shù)據(jù)客戶端不需要也不應(yīng)該頻繁回傳。這和 LLM 生成的場景天然匹配——用戶提問之后剩下的就是模型一直說前端一直聽。協(xié)議層面SSE 就是普通的 HTTP 響應(yīng)Content-Type 設(shè)為text/event-stream然后在響應(yīng)體里按固定格式寫事件event: message data: {type: token, content: 你} event: message data: {type: token, content: 好} event: done data: {type: done, session_id: abc123}每個(gè)事件之間用空行隔開data字段是真正的載荷event是事件類型id可以用來做斷點(diǎn)續(xù)傳retry是客戶端自動(dòng)重連的時(shí)間間隔。就這么簡單。WebSocket 呢它是全雙工要握手、要維護(hù)長連接心跳、要處理斷線重連邏輯在只需要服務(wù)器單向推送的場景里屬于殺雞用牛刀。更現(xiàn)實(shí)的問題很多企業(yè)內(nèi)網(wǎng)的網(wǎng)關(guān)、Nginx 配置對(duì) WebSocket 的升級(jí)請求支持不友好但對(duì) SSE 這種普通 HTTP 長響應(yīng)基本上零成本透明轉(zhuǎn)發(fā)。我在實(shí)際項(xiàng)目里用 SSE 遇到過的最大坑反而很簡單網(wǎng)關(guān)的超時(shí)時(shí)間設(shè)太短模型思考超過 60 秒連接直接被掐斷前端就收到一個(gè)stream disconnected before completion: idle timeout waiting for sse。這個(gè)問題后面在排查章節(jié)我會(huì)詳細(xì)說。2.2 從 LangChain 的流式機(jī)制到 SSE 事件拼裝LangChain 從Runnable體系開始把流式能力統(tǒng)一成了stream/astream兩個(gè)接口。chain.stream(query)會(huì)一塊一塊地yield輸出。需要注意的是這個(gè)一塊不一定是模型吐的一個(gè) token而是 LangChain 每個(gè)Runnable步驟產(chǎn)出的一個(gè)完整單元。比如Retriever步驟吐出文檔列表LLM步驟吐出 token 片段。在 Agent 場景下中間可能還夾著Tool的執(zhí)行結(jié)果。所以設(shè)計(jì) SSE 接口時(shí)我的做法是先約定一套內(nèi)部事件協(xié)議而不是直接把 token 裸推出去。每個(gè)事件我用 JSON 串里面至少帶兩個(gè)字段type和payload。event: message data: {type: start, payload: {session_id: uuid}} event: message data: {type: token, payload: {content: 你好}} event: message data: {type: tool_call, payload: {name: search_news, args: {keyword: 人工智能}}} event: message data: {type: tool_result, payload: {result: ..., duration_ms: 1200}} event: message data: {type: structured, payload: {title: ..., summary: ...}} event: message data: {type: done, payload: {finish_reason: stop}}前端只需要根據(jù)type決定怎么渲染token追加到正文tool_call可以展示正在調(diào)用工具的動(dòng)畫structured存到表單里做后續(xù)業(yè)務(wù)。這樣流式體驗(yàn)和結(jié)構(gòu)化數(shù)據(jù)就各歸其位了。后面我講的三大 OutputParser本質(zhì)都是在最后那個(gè)結(jié)構(gòu)化事件這個(gè)環(huán)節(jié)里保證你拿到的payload是干凈、可校驗(yàn)的 JSON。3. 三大 OutputParser把模型的話轉(zhuǎn)成程序能用的結(jié)構(gòu)OutputParser 在 LangChain 里的定位是從 LLM 的原始輸出里抽取出程序需要的結(jié)構(gòu)。注意它通常不是魔法它靠的是提示詞約束 格式校驗(yàn)兩步走。讓模型按指定格式輸出再用解析器校驗(yàn)、糾錯(cuò)。理解了這一點(diǎn)你就能明白為什么換個(gè)模型解析失敗中文環(huán)境下 JSON 不標(biāo)準(zhǔn)這類問題會(huì)反復(fù)出現(xiàn)了。3.1 PydanticOutputParser給 JSON 上一份類型保票PydanticOutputParser是項(xiàng)目里最常用的一個(gè)。它結(jié)合了 Pydantic 的BaseModel把輸出格式約束成明確的字段類型——字符串、整型、列表、嵌套對(duì)象類型不對(duì)直接校驗(yàn)失敗。用法上核心三步定義模型類、創(chuàng)建解析器、把格式指令塞進(jìn)提示詞。from typing import List from pydantic import BaseModel, Field from langchain.output_parsers import PydanticOutputParser class ArticleSummary(BaseModel): title: str Field(description生成的文章標(biāo)題不超過20字) keywords: List[str] Field(description3-5個(gè)關(guān)鍵詞) summary: str Field(description100字以內(nèi)的核心摘要) confidence: float Field(description模型對(duì)摘要質(zhì)量的自信度0到1之間) parser PydanticOutputParser(pydantic_objectArticleSummary) prompt PromptTemplate( template請分析下面這段文本并嚴(yán)格按照格式要求輸出。\n文本{text}\n{format_instructions}\n, input_variables[text], partial_variables{format_instructions: parser.get_format_instructions()}, )parser.get_format_instructions()生成的那段指令本質(zhì)是把你定義的字段、類型、約束翻譯成自然語言模板告訴模型必須輸出一個(gè) JSONkey 有哪些value 是什么類型。它還會(huì)補(bǔ)充一句不要輸出其他內(nèi)容之類的強(qiáng)調(diào)。實(shí)際效果上模型越強(qiáng)GPT-4 級(jí)別、Claude 3.5遵循度越高小模型經(jīng)常把 JSON 包在 Markdown 代碼塊里或者多解釋一句。解析器的parse方法還內(nèi)置了糾錯(cuò)能力如果模型輸出的文本可以被eval成 JSON但類型不對(duì)它會(huì)嘗試用 LLM 自動(dòng)修復(fù)。不過這個(gè)修復(fù)是有損的——它需要額外調(diào)一次模型速度和成本都要考慮。所以在流式場景里我通常不在最后階段用自動(dòng)修復(fù)而是做兩段式先流式展示再后端靜默校驗(yàn)校驗(yàn)失敗才觸發(fā)修復(fù)。3.2 StructuredOutputParser輕量到不需要定義類StructuredOutputParser適合那種不想建 Pydantic 模型只要幾個(gè)簡單字段的場景。它通過ResponseSchema列表來聲明字段名、類型和描述使用起來比 Pydantic 版更輕。from langchain.output_parsers import StructuredOutputParser, ResponseSchema response_schemas [ ResponseSchema(nameanswer, description對(duì)問題的直接回答, typestring), ResponseSchema(namesource, description答案的參考來源如果沒有則為null, typestring), ] parser StructuredOutputParser.from_response_schemas(response_schemas)從源碼實(shí)現(xiàn)看StructuredOutputParser內(nèi)部并沒有把輸出嚴(yán)格轉(zhuǎn)成 Pydantic 對(duì)象而是返回一個(gè)字典。它對(duì)字段順序、缺失字段的處理更寬松底層用的其實(shí)是類似正則 字典提取的簡易邏輯。所以它適合內(nèi)部接口、日志記錄、字段不多的場景一旦你的下游真的要用強(qiáng)類型做入?yún)⑿ｒ?yàn)還是 Pydantic 版更省心。這里有個(gè)經(jīng)驗(yàn)如果返回結(jié)構(gòu)里嵌套層級(jí)很深或者字段會(huì)因?yàn)槟Ｐ洼敵鲲L(fēng)格波動(dòng)我建議直接用 Pydantic 版不要在 Structured 版上強(qiáng)行造輪子。輕量方案省下的代碼量會(huì)在排障時(shí)加倍還回去。3.3 JsonOutputParser最容易被低估的那個(gè)第三個(gè)其實(shí)是JsonOutputParser—— LangChain 里專門用來只要 JSON不要類定義的解析器。它和PydanticOutputParser最大的區(qū)別是不要求目標(biāo)類型是 Pydantic 模型你用普通 dict 聲明一個(gè)期望的 JSON 結(jié)構(gòu)模板它就能照這個(gè)模板校驗(yàn)。from langchain.output_parsers import JsonOutputParser parser JsonOutputParser() prompt PromptTemplate( template輸出JSON格式結(jié)果字段包括: title(string), items(array of string)。\n{format_instructions}\n, input_variables[], partial_variables{format_instructions: parser.get_format_instructions()}, )它做的事情是拿到模型文本提取并解析出 JSON 對(duì)象。它不會(huì)去做嚴(yán)格類型強(qiáng)轉(zhuǎn)保持了 dict 的靈活性。實(shí)際項(xiàng)目中我經(jīng)常把它作為流式過程中增量解析 JSON 的工具——不是等模型完整輸出后一次性解析而是配合字節(jié)流每次拿到新的 token 片段就去試著解析能解析出部分字段就先緩存。這個(gè)思路在長回答場景里特別有用你可以在模型還在生成正文時(shí)就把標(biāo)題、關(guān)鍵列表等輪廓字段提前推給前端?？偨Y(jié)一下三者的選擇邏輯要強(qiáng)類型校驗(yàn)、下游要嚴(yán)格入庫選 Pydantic只要幾個(gè)字段、隨拿隨用選 Structured既要 JSON 又不想綁定模型定義、或需要在流中做增量解析選 Json。沒有絕對(duì)好壞只看約束強(qiáng)度。4. ToolCall 方案讓模型把工具意圖直接交出來4.1 為什么不用讓模型自己拼工具調(diào)用文本早期 LangChain 的 Agent 實(shí)現(xiàn)里模型是用純文本的方式假裝調(diào)用工具——輸出一行Action: search_news\nAction Input: 人工智能然后 AgentExecutor 去解析這段文本。這種方案在模型能力弱的時(shí)候還算勉強(qiáng)能用但問題很明顯模型一旦在文本里多加一句解釋、少寫一個(gè)換行整個(gè)解析就崩了。而且文本格式因模型而異換模型就要調(diào)解析規(guī)則?，F(xiàn)在主流方案是ToolCall函數(shù)調(diào)用。模型在生成時(shí)除了輸出自然語言還可以輸出一個(gè)結(jié)構(gòu)化的工具調(diào)用意圖——方法名、參數(shù) JSON。OpenAI 的 function calling、Claude 的 tool use、國產(chǎn)模型不少也兼容這個(gè)協(xié)議。LangChain 里的做法是把工具定義綁定到模型上這一類模型能力稱為bind_tools。from langchain_openai import ChatOpenAI from langchain_core.tools import tool tool def search_news(keyword: str, limit: int 5) - list: 搜索新聞資訊keyword為關(guān)鍵詞limit為返回條數(shù)。 # 這里寫真實(shí)檢索邏輯 return [{title: 示例新聞, url: https://example.com}] llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools([search_news]) response llm_with_tools.invoke(幫我搜一下今天人工智能領(lǐng)域的新聞)關(guān)鍵在于response是一個(gè)AIMessage如果模型決定調(diào)用工具它的tool_calls屬性里會(huì)帶上結(jié)構(gòu)化調(diào)用信息response.tool_calls # [{name: search_news, args: {keyword: 人工智能, limit: 5}, id: call_xxx}]這個(gè)id字段很重要尤其是做并發(fā)工具調(diào)用時(shí)它用來關(guān)聯(lián)工具結(jié)果和對(duì)應(yīng)的調(diào)用請求。4.2 ToolCall 事件在 SSE 里怎么推既然AIMessage.tool_calls是結(jié)構(gòu)化的那么從流式事件角度它也能流式地分片到達(dá)——模型先生成工具名再一點(diǎn)一點(diǎn)生成參數(shù) JSON。LangChain 的astream_events可以讓你捕獲on_chat_model_stream事件從而拿到 token 級(jí)的流。但我要提醒你工具參數(shù)這種 JSON前端完全沒必要做打字機(jī)效果。你只要在tool_call開始事件里推一條正在調(diào)用工具等工具結(jié)果出來再推一條結(jié)構(gòu)化結(jié)果就行參數(shù) JSON 在中間過程可以直接攢在后端。這是我的實(shí)踐結(jié)論——不要一上來就把所有 token 都推給前端做逐字渲染那樣只會(huì)讓前端渲染邏輯又復(fù)雜又容易出錯(cuò)。工具執(zhí)行完你拿到的結(jié)果同樣建議包裝成結(jié)構(gòu)化事件推給前端。同時(shí)把工具結(jié)果作為新的上下文消息再喂回給模型讓它基于結(jié)果生成最終回答。這個(gè)模型→工具→模型的循環(huán)如果自己用for循環(huán)寫很容易在異常分支和超時(shí)控制上出問題——這也是為什么存在 LangGraph 這類帶狀態(tài)編排的框架。不過對(duì)于單輪工具調(diào)用場景手動(dòng)循環(huán)完全可控不需要上重型框架。4.3 OutputParser 和 ToolCall 怎么配合這就是這個(gè)方案的精髓了ToolCall 解決模型要調(diào)用什么工具、參數(shù)是什么的結(jié)構(gòu)化提取OutputParser 解決模型最終要返回給業(yè)務(wù)的最終結(jié)論的結(jié)構(gòu)化提取。兩者是在一次請求的不同階段各司其職。class FinalAnswer(BaseModel): reply: str Field(description面向用戶的最終回答) used_tools: List[str] Field(description本次實(shí)際使用到的工具名稱列表) data_source: List[str] Field(description參考信息的來源列表) final_parser PydanticOutputParser(pydantic_objectFinalAnswer)流程大致是用戶提問 → 模型決定調(diào)用工具ToolCall 結(jié)構(gòu)化→ 執(zhí)行工具 → 把工具結(jié)果拼進(jìn)上下文 → 模型生成最終回答OutputParser 結(jié)構(gòu)化→ 通過 SSE 推送給前端。中間環(huán)節(jié)的結(jié)構(gòu)化靠tool_calls最后的業(yè)務(wù)結(jié)構(gòu)靠 OutputParser。兩條線涇渭分明誰也不會(huì)干擾誰。5. FastAPI LangChain 完整落地一條 SSE 接口打通全流程5.1 服務(wù)端異步流式接口的分層設(shè)計(jì)我強(qiáng)烈建議把模型調(diào)用邏輯和HTTP 流式協(xié)議分開。模型調(diào)用邏輯是一個(gè)普通的異步生成器它只負(fù)責(zé)產(chǎn)出結(jié)構(gòu)化事件字典HTTP 層只負(fù)責(zé)把事件字典按 SSE 協(xié)議編碼。這樣拆開你可以對(duì)模型邏輯單測不用每次起服務(wù)。from fastapi import FastAPI from fastapi.responses import StreamingResponse import json, asyncio from langchain_openai import ChatOpenAI from langchain_core.output_parsers import PydanticOutputParser from pydantic import BaseModel, Field from typing import List app FastAPI() llm ChatOpenAI(modelgpt-4o-mini, temperature0.3, streamingTrue) class FinalAnswer(BaseModel): reply: str Field(description面向用戶的最終回答) keywords: List[str] Field(description3-5個(gè)關(guān)鍵詞) parser PydanticOutputParser(pydantic_objectFinalAnswer) async def event_generator(prompt: str): # 1. 自動(dòng)構(gòu)建帶格式約束的提示詞 formatted_prompt ( 請回答用戶問題并輸出嚴(yán)格JSON。\n問題{q}\n{fmt}\n ).format(qprompt, fmtparser.get_format_instructions()) # 2. 第一個(gè)事件告知開始 yield { event: message, data: json.dumps({type: start, payload: {time: asyncio.time()}}, ensure_asciiFalse) } # 3. 流式輸出 token 事件 collected async for chunk in llm.astream(formatted_prompt): collected chunk.content yield { event: message, data: json.dumps({type: token, payload: {content: chunk.content}}, ensure_asciiFalse) } # 控制推送節(jié)奏避免瞬間把積壓的token全倒出去 await asyncio.sleep(0) # 4. 結(jié)構(gòu)化解析并推給前端 try: parsed parser.parse(collected) yield { event: message, data: json.dumps({type: structured, payload: parsed.model_dump()}, ensure_asciiFalse) } except Exception as exc: yield { event: message, data: json.dumps({type: parse_error, payload: {error: str(exc)}}, ensure_asciiFalse) } # 5. 結(jié)束事件 yield { event: message, data: json.dumps({type: done, payload: {finish_reason: stop}}, ensure_asciiFalse) } app.post(/chat/stream) async def chat_stream(payload: dict): prompt payload.get(prompt, ) return StreamingResponse( event_generator(prompt), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, # 重要禁止Nginx緩沖 }, )X-Accel-Buffering: no這個(gè) header是很多人在 Nginx 反代下 SSE 不流式的元兇。Nginx 默認(rèn)會(huì)緩沖響應(yīng)攢滿 4KB 或者等連接結(jié)束才發(fā)給前端你明明在服務(wù)端yield了前端卻半天沒動(dòng)靜。加了這個(gè) header 就是明確告訴 Nginx 別緩沖。5.2 前端事件分發(fā)與渲染解耦前端用fetch配合ReadableStream解析 SSE 就夠了不一定非要引eventsource-parser這類庫但引了確實(shí)省事。核心邏輯是讀到一行data:解析 JSON按type走不同的渲染函數(shù)。async function streamChat(prompt) { const resp await fetch(/chat/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ prompt }), }); const reader resp.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { value, done } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); // SSE事件以空行分隔 let sepIndex; while ((sepIndex buffer.indexOf(\n\n)) ! -1) { const rawEvent buffer.slice(0, sepIndex); buffer buffer.slice(sepIndex 2); const dataLine rawEvent.split(\n).find(line line.startsWith(data:)); if (!dataLine) continue; const msg JSON.parse(dataLine.slice(5).trim()); handleEvent(msg); } } } function handleEvent(msg) { switch (msg.type) { case token: appendText(msg.payload.content); break; case structured: fillMetaPanel(msg.payload); break; case tool_call: showToolIndicator(msg.payload.name); break; case done: stopLoading(); break; } }注意千萬別直接用瀏覽器的原生EventSource—— 它只支持 GET 請求而我們往往需要 POST 傳遞 prompt。原生EventSource也沒有自定義 header 的能力鑒權(quán)都麻煩。用fetch流式讀取是最通用的方案。5.3 一邊流式一邊結(jié)構(gòu)化增量解析的實(shí)踐前面提到的JsonOutputParser增量解析在實(shí)際項(xiàng)目中可以這樣用每收到一段新 token就把collected追加后嘗試parser.parse如果解析成功哪怕還不完整只要能出部分字段就把部分結(jié)果推給結(jié)構(gòu)化預(yù)覽事件如果解析失敗因?yàn)?JSON 還沒閉合忽略即可不算錯(cuò)誤。這個(gè)模式我用來解決一個(gè)具體的痛點(diǎn)用戶問幫我總結(jié)這份文檔并給出三個(gè)要點(diǎn)模型正文還沒寫完我希望前端右側(cè)欄已經(jīng)先把要點(diǎn)標(biāo)題渲染出來。雖然嚴(yán)格說最終結(jié)果要以最后完整解析為準(zhǔn)但增量預(yù)覽的體驗(yàn)提升非常明顯。代價(jià)是每次追加 token 都會(huì)觸發(fā)一次 JSON 解析token 非常長時(shí)會(huì)有少量 CPU 開銷。我實(shí)測下來對(duì)普通問答長度的文本這個(gè)開銷可以忽略。6. 常見問題與排查技巧實(shí)錄6.1 SSE 流中途斷開空閑超時(shí)是頭號(hào)殺手提示stream disconnected before completion: idle timeout waiting for sse這類報(bào)錯(cuò)絕大多數(shù)不是代碼問題是鏈路中的代理/網(wǎng)關(guān)配置問題。我遇到過最典型的三層排查順序第一層本地測試。先用curl -N直接打服務(wù)接口觀察事件是不是正常持續(xù)輸出。curl -N能實(shí)時(shí)打印服務(wù)器推來的每個(gè)事件如果這一步正常問題就不在后端。第二層查反向代理。Nginx 的proxy_read_timeout默認(rèn) 60 秒模型思考時(shí)間一旦超過代理直接斷連。調(diào)大或用proxy_read_timeout 300s可以緩解。另外確認(rèn)proxy_buffering off;或X-Accel-Buffering: no已生效。第三層查云廠商網(wǎng)關(guān)。很多云負(fù)載均衡器對(duì)長連接也有空閑超時(shí)限制比如 60 秒、120 秒。盡量用 WebSocket 或 SSE 都能走的長連接配置同時(shí)后端在流式傳輸過程中即使沒有數(shù)據(jù)也要定期發(fā)一個(gè): ping注釋行作為心跳。SSE 規(guī)范里以冒號(hào)開頭的行是注釋客戶端會(huì)忽略它但能刷新代理的空閑計(jì)時(shí)器。async def keepalive(): while True: yield : ping\n\n await asyncio.sleep(15)把這個(gè)生成器和主事件生成器用asyncio.gather合并就能在模型長時(shí)間思考時(shí)維持連接活性。6.2 OutputParser 拿到半截 JSON或 Markdown 代碼塊模型輸出里常見的臟格式有兩種一是把 JSON 藏在json代碼塊里二是前后夾帶解釋文字。PydanticOutputParser本身會(huì)嘗試從文本里提取 JSON 塊但并不可靠。我的兜底方案是寫一個(gè)基礎(chǔ)清洗函數(shù)在喂給解析器之前先做預(yù)處理。import re, json def extract_json_string(text: str) - str: text text.strip() # 去掉首尾的 markdown 代碼塊標(biāo)記 code_block_pattern re.compile(r(?:json)?\s*(.*?)\s*, re.DOTALL) match code_block_pattern.search(text) if match: return match.group(1) # 嘗試從第一個(gè) { 到最后一個(gè) } 截取 start text.find({) end text.rfind(}) if start ! -1 and end ! -1 and end start: return text[start:end1] return text然后統(tǒng)一走parse。注意如果清洗后還是解析失敗再去觸發(fā) LLM 原文本修復(fù)。一定不要默認(rèn)讓每個(gè)失敗都走修復(fù)不然成本和延遲都會(huì)失控。6.3 并發(fā)請求下事件錯(cuò)亂上下文變量與隊(duì)列如果你的 FastAPI 服務(wù)同時(shí)處理多個(gè) SSE 會(huì)話每個(gè)會(huì)話的生成器是獨(dú)立的理論上不會(huì)串。但我踩過一個(gè)實(shí)際的坑在生成器內(nèi)部用了模塊級(jí)的全局變量緩存工具結(jié)果兩個(gè)用戶同時(shí)觸發(fā)同一個(gè)工具調(diào)用時(shí)A 用戶的結(jié)果可能被 B 用戶覆蓋。解決方案很簡單每個(gè)會(huì)話的事件生成器必須是自包含的所有狀態(tài)都放在生成器內(nèi)部不要依賴模塊級(jí)可變對(duì)象。需要跨函數(shù)傳狀態(tài)就用contextvars或者干脆把 session_id 作為 key 放進(jìn)一個(gè)字典管理隊(duì)列。我在項(xiàng)目里用的模式是每個(gè)會(huì)話一個(gè)asyncio.Queue生成器往隊(duì)列放事件SSE 層從隊(duì)列取事件編碼輸出。這個(gè)抽象能讓你在后續(xù)擴(kuò)展多 Agent 編排時(shí)游刃有余。7. 一些我踩過坑之后的固定習(xí)慣先說工具聲明。LangChain 的tool裝飾器會(huì)讀取函數(shù)的 docstring 和類型注解來生成工具的 schema。docstring 里的描述、參數(shù)的類型提示、默認(rèn)值都會(huì)成為傳給模型的 tool schema 的一部分。所以我在寫工具函數(shù)時(shí)會(huì)強(qiáng)制自己把每個(gè)參數(shù)的單位、取值范圍、邊界情況寫進(jìn) docstring這不是為了寫注釋好看而是直接決定模型能不能正確填參。一個(gè)只寫 keyword: 搜索關(guān)鍵詞 的工具和一個(gè)寫著 keyword: 搜索關(guān)鍵詞最長20字符不要帶引號(hào) 的工具在模型調(diào)參準(zhǔn)確率上差很多。然后是解析器的temperature。做結(jié)構(gòu)化輸出時(shí)模型溫度不建議設(shè)太高0 到 0.3 之間最穩(wěn)。溫度高了模型更容易發(fā)揮創(chuàng)造力去改格式、加注釋這對(duì)我們的 JSON 解析是災(zāi)難。如果既要?jiǎng)?chuàng)意又要結(jié)構(gòu)化我一般拆兩條鏈一條低溫度出結(jié)構(gòu)化摘要一條高溫度潤色成自然語言回復(fù)最后再把兩段結(jié)果拼進(jìn) SSE 事件里。最后再分享一個(gè)小技巧給事件協(xié)議加一個(gè)trace_id字段。每個(gè) SSE 會(huì)話生成一個(gè)trace_id在start事件里發(fā)給前端同時(shí)在服務(wù)端日志里打出來。前端報(bào) bug 時(shí)直接甩這個(gè) ID 給你你能在日志里把整條鏈路還原出來省掉大量你剛才問的什么來著的溝通成本。我在生產(chǎn)環(huán)境靠這個(gè)字段排查過很多偶發(fā)問題尤其是模型偶發(fā)出錯(cuò)那種玄學(xué)問題有 trace_id 才能對(duì)上號(hào)。這套方案跑穩(wěn)之后你會(huì)發(fā)現(xiàn)流式體驗(yàn)和結(jié)構(gòu)化返回其實(shí)不是二選一關(guān)鍵是讓它們各走各的通道文本走 token 事件結(jié)構(gòu)化走獨(dú)立事件中間用統(tǒng)一的 JSON 協(xié)議串聯(lián)。理解了這層設(shè)計(jì)后續(xù)換模型、加工具、上 LangGraph 編排都不會(huì)再被到底是文本還是 JSON這個(gè)問題卡住。