FastAPI LlamaIndex Agent 流式對話踩坑實錄取消標(biāo)志殘留、中間件緩沖、事件丟失全記錄文章目錄FastAPI LlamaIndex Agent 流式對話踩坑實錄取消標(biāo)志殘留、中間件緩沖、事件丟失全記錄前言一、項目背景事件類型二、問題一BaseHTTPMiddleware 導(dǎo)致流式響應(yīng)被緩沖2.1 問題現(xiàn)象2.2 原因分析2.3 解決方案2.4 關(guān)鍵結(jié)論三、問題二asyncio.Queue 間接層導(dǎo)致事件丟失3.1 問題現(xiàn)象3.2 原因分析3.3 解決方案3.4 關(guān)鍵結(jié)論四、問題三Redis 取消標(biāo)志殘留導(dǎo)致新對話被誤判核心問題4.1 問題現(xiàn)象4.2 原因分析4.3 解決方案4.4 為什么這樣修復(fù)有效4.5 關(guān)鍵結(jié)論五、完整修復(fù)代碼5.1 agent_service.py核心修改5.2 agent_controller.py傳入 Redis 實例5.3 中間件修復(fù)純 ASGI 實現(xiàn)六、排查經(jīng)驗總結(jié)6.1 流式響應(yīng)排查清單6.2 關(guān)鍵日志模板6.3 最佳實踐七、總結(jié)前言最近在基于 FastAPI LlamaIndex 構(gòu)建 Agent 智能對話系統(tǒng)時遇到了一個典型的流式響應(yīng)問題Agent 對話只返回start和end(cancelled)事件中間的內(nèi)容全部丟失。經(jīng)過多輪排查和修復(fù)最終定位到三個獨立但相互關(guān)聯(lián)的問題。本文完整記錄問題現(xiàn)象、排查過程和解決方案希望能幫助遇到類似問題的同學(xué)少走彎路。一、項目背景Web 框架FastAPI UvicornAI 框架LlamaIndexFunctionAgent流式協(xié)議NDJSONapplication/x-ndjson每行一個 JSON 對象{type: xxx, payload: {...}}取消機制通過 Redis 標(biāo)志位實現(xiàn)用戶點擊取消時設(shè)置agent_cancel:{session_id} 1Agent 事件循環(huán)中輪詢檢測事件類型事件類型說明start對話開始攜帶sessionIdmessageAI 回答內(nèi)容增量片段sources檢索來源切片列表end對話結(jié)束error異常信息二、問題一BaseHTTPMiddleware 導(dǎo)致流式響應(yīng)被緩沖2.1 問題現(xiàn)象Agent 流式接口返回的StreamingResponse在客戶端只能收到start事件后續(xù)的message、sources、end等事件全部丟失。2.2 原因分析項目中有兩個中間件使用了 Starlette 的BaseHTTPMiddleware# 中間件 1上下文清理classContextCleanupMiddleware(BaseHTTPMiddleware):asyncdefdispatch(self,request,call_next):responseawaitcall_next(request)RequestContext.clear_all()returnresponse# 中間件 2響應(yīng)頭追加classApiResponseHeaderMiddleware(BaseHTTPMiddleware):asyncdefdispatch(self,request,call_next):responseawaitcall_next(request)# 追加自定義響應(yīng)頭...returnresponseBaseHTTPMiddleware的call_next()內(nèi)部會攔截響應(yīng)流導(dǎo)致StreamingResponse的 NDJSON chunk 被緩沖或提前終止。這是 Starlette 的已知問題BaseHTTPMiddleware不適合處理流式響應(yīng)。2.3 解決方案將兩個中間件從BaseHTTPMiddleware轉(zhuǎn)為純 ASGI 實現(xiàn)直接傳遞send函數(shù)不攔截響應(yīng)流# 純 ASGI 中間件上下文清理classContextCleanupMiddleware:def__init__(self,app)-None:self.appappasyncdef__call__(self,scope,receive,send)-None:ifscope[type]!http:awaitself.app(scope,receive,send)returntry:awaitself.app(scope,receive,send)finally:RequestContext.clear_all()# 純 ASGI 中間件響應(yīng)頭追加classApiResponseHeaderMiddleware:def__init__(self,app)-None:self.appappasyncdef__call__(self,scope,receive,send)-None:ifscope[type]!http:awaitself.app(scope,receive,send)returnrequestRequest(scope,receive)api_response_headersgetattr(request.state,api_response_headers,None)ifnotapi_response_headers:awaitself.app(scope,receive,send)returnasyncdefsend_with_headers(message)-None:ifmessage[type]http.response.start:headersdict(message.get(headers,[]))forkey,valueinapi_response_headers.items():headers[key.encode(utf-8)]value.encode(utf-8)message{**message,headers:list(headers.items())}awaitsend(message)awaitself.app(scope,receive,send_with_headers)2.4 關(guān)鍵結(jié)論凡是涉及流式響應(yīng)SSE、NDJSON、WebSocket 等的 FastAPI 項目中間件必須使用純 ASGI 實現(xiàn)不能使用BaseHTTPMiddleware。三、問題二asyncio.Queue 間接層導(dǎo)致事件丟失3.1 問題現(xiàn)象中間件修復(fù)后流式接口仍然只返回start事件。后端日志顯示 Agent 正常執(zhí)行RAG 檢索命中、LLM 調(diào)用成功但事件無法送達客戶端。3.2 原因分析之前的架構(gòu)使用了asyncio.Queue 后臺任務(wù)來解耦 Agent 執(zhí)行和 HTTP 流Agent 事件 → queue.put() → queue.get() → yield → 中間件 → 客戶端這個架構(gòu)在 Agent 事件和 HTTP 流之間增加了間接層導(dǎo)致事件在 queue 傳遞過程中丟失或阻塞。3.3 解決方案徹底去掉 Queue改為直接流式傳輸# 修復(fù)后的架構(gòu)Agent 事件 →yield→ 中間件 → 客戶端classmethodasyncdefchat_stream(cls,db,request,user_id,app_redisNone):# ... setup 代碼 ...# 創(chuàng)建 Agent 并運行agentAgentFactory.create_agent(...)handleragent.run(user_msgrequest.query,chat_historyllama_messages)yieldcls._ndjson(start,{sessionId:actual_session_id})# 直接從 Agent 事件流轉(zhuǎn)發(fā)給客戶端asyncforeventinhandler.stream_events():event_typetype(event).__name__ifevent_typeAgentStream:deltagetattr(event,delta,)ifdelta:full_answerdeltayieldcls._ndjson(message,{content:delta})elifevent_typeToolCallResult:# 提取來源...passyieldcls._ndjson(end,{})3.4 關(guān)鍵結(jié)論對于 LlamaIndex Agent 的流式場景直接從handler.stream_events()yield 事件給客戶端即可不需要 Queue 間接層。Queue 架構(gòu)適合需要復(fù)雜的生產(chǎn)者-消費者模式但在簡單的流式轉(zhuǎn)發(fā)場景中反而增加了不必要的復(fù)雜度和出錯概率。四、問題三Redis 取消標(biāo)志殘留導(dǎo)致新對話被誤判核心問題4.1 問題現(xiàn)象用戶先取消一次對話然后在同一個會話中再次提問新對話立即返回{type: end, payload: {cancelled: true}}完全沒有內(nèi)容。4.2 原因分析取消機制使用 Redis 標(biāo)志位# 取消接口awaitredis.set(fagent_cancel:{session_id},1,ex3600)# 事件循環(huán)中檢測flagawaitredis.get(fagent_cancel:{session_id})ifflag1:yieldcls._ndjson(end,{cancelled:True})return問題 1標(biāo)志殘留取消標(biāo)志的 TTL 是 3600 秒1 小時。用戶取消后標(biāo)志留在 Redis 中。同一會話的后續(xù)請求會檢測到這個殘留標(biāo)志被誤判為已取消。問題 2時序競爭即使在新對話開始時清除標(biāo)志仍然存在時序競爭時間線 13.638 → 請求 2 開始 setup保存問題、構(gòu)建記憶、創(chuàng)建 Agent... 14.228 → 用戶點擊取消仍在請求 2 的 setup 階段 14.467 → 取消標(biāo)志被寫入 Redis 14.500 → 請求 2 的 setup 結(jié)束 14.504 → 請求 2 的事件循環(huán)檢測到標(biāo)志 → 誤判如果把清除標(biāo)志的代碼放在 setup之前clear 在 13.638 執(zhí)行 → 標(biāo)志還不存在查了個空cancel 在 14.467 設(shè)置標(biāo)志事件循環(huán)在 14.504 檢測到標(biāo)志 → 誤判4.3 解決方案雙保險策略將清除標(biāo)志的代碼移到 setup 之后、事件循環(huán)之前檢測到取消后立即清除標(biāo)志classmethodasyncdefchat_stream(cls,db,request,user_id,app_redisNone):actual_session_idrequest.session_idorstr(uuid.uuid4())# 獲取 Redis 實例redisapp_redisorawaitcls._get_redis(db)# Setup 階段 # 1. 保存用戶問題# 2. 構(gòu)建會話記憶# 3. 創(chuàng)建 Agent# 4. 啟動 Agent 運行# yieldcls._ndjson(start,{sessionId:actual_session_id})# ★ 關(guān)鍵修復(fù) 1setup 完成后、事件循環(huán)開始前清除殘留的取消標(biāo)志ifredis:cancel_keyfagent_cancel:{actual_session_id}old_flagawaitredis.get(cancel_key)ifold_flag:awaitredis.delete(cancel_key)logger.info(f已清除殘留取消標(biāo)志:{cancel_key}(舊值{old_flag}))try:asyncforeventinhandler.stream_events():# 檢查用戶是否已取消ifredisandawaitcls._is_cancelled(redis,actual_session_id):# ★ 關(guān)鍵修復(fù) 2檢測到取消后立即清除標(biāo)志try:awaitredis.delete(fagent_cancel:{actual_session_id})exceptException:passlogger.info(f用戶已取消對話: session_id{actual_session_id})awaitcls._save_cancelled_answer(...)yieldcls._ndjson(end,{cancelled:True})return# 處理事件...4.4 為什么這樣修復(fù)有效修復(fù)后的時序時間線修復(fù)后 13.638 → 請求 2 開始 setup 14.228 → 用戶點擊取消 14.467 → 取消標(biāo)志被寫入 Redis 14.500 → setup 結(jié)束清除取消標(biāo)志 ← 標(biāo)志被清除 14.504 → 事件循環(huán)開始 → 沒有標(biāo)志 → 正常運行 ?清除標(biāo)志的代碼從 setup之前移到 setup之后確保了即使取消請求在 setup 期間到達并設(shè)置了標(biāo)志clear 也會在事件循環(huán)開始前把它清掉事件循環(huán)開始時看到的永遠是干凈的狀態(tài)4.5 關(guān)鍵結(jié)論取消標(biāo)志的生命周期應(yīng)該與請求綁定而不是與會話綁定。每次新請求開始時清除舊標(biāo)志每次取消被處理后也立即清除標(biāo)志確保標(biāo)志不會殘留影響后續(xù)請求。五、完整修復(fù)代碼5.1 agent_service.py核心修改classmethodasyncdefchat_stream(cls,db:AsyncSession,request,user_id:int,app_redisNone)-AsyncGenerator[str,None]:Agent 流式對話入口frommodule_rag.service.rag_chat_history_serviceimportRagChatHistoryService actual_session_idrequest.session_idorstr(uuid.uuid4())# 0. 獲取 Redis 實例優(yōu)先使用應(yīng)用級 Redisredisapp_redisifnotredis:try:redisawaitcls._get_redis(db)exceptException:redisNone# Setup 階段保存問題、構(gòu)建記憶、創(chuàng)建 Agent 等# ... 省略 setup 代碼 ...# 啟動 AgentagentAgentFactory.create_agent(...)handleragent.run(user_msgrequest.query,chat_historyllama_messages)yieldcls._ndjson(start,{sessionId:actual_session_id})# ★ 修復(fù)setup 完成后、事件循環(huán)前清除殘留取消標(biāo)志ifredis:try:cancel_keyfagent_cancel:{actual_session_id}old_flagawaitredis.get(cancel_key)ifold_flag:awaitredis.delete(cancel_key)logger.info(f已清除殘留取消標(biāo)志:{cancel_key}(舊值{old_flag}))exceptExceptionasclear_err:logger.warning(f清除取消標(biāo)志失敗:{clear_err})try:asyncforeventinhandler.stream_events():ifredisandawaitcls._is_cancelled(redis,actual_session_id):# ★ 修復(fù)檢測到取消后立即清除標(biāo)志try:awaitredis.delete(fagent_cancel:{actual_session_id})exceptException:passawaitcls._save_cancelled_answer(...)yieldcls._ndjson(end,{cancelled:True})returnevent_typetype(event).__name__ifevent_typeAgentStream:deltagetattr(event,delta,)ifdelta:yieldcls._ndjson(message,{content:delta})yieldcls._ndjson(end,{})# 后臺保存回答fire-and-forgetcls._post_chat_tasks(...)exceptExceptionasstream_err:yieldcls._ndjson(error,{message:str(stream_err)})5.2 agent_controller.py傳入 Redis 實例agent_controller.post(/chat/stream)asyncdefagent_chat_stream(request:Request,chat_req:AgentChatRequestModel,query_db:Annotated[AsyncSession,DBSessionDependency()],current_user:Annotated[CurrentUserModel,CurrentUserDependency()],)-StreamingResponse:user_idcurrent_user.user.user_id# 獲取應(yīng)用級 Redis 實例與取消接口使用同一個連接app_redisrequest.app.state.redisifhasattr(request.app.state,redis)elseNoneevent_streamAgentService.chat_stream(query_db,chat_req,user_id,app_redisapp_redis)returnStreamingResponse(contentevent_stream,media_typeapplication/x-ndjson)5.3 中間件修復(fù)純 ASGI 實現(xiàn)# context_middleware.pyclassContextCleanupMiddleware:上下文清理中間件純 ASGI 實現(xiàn)def__init__(self,app)-None:self.appappasyncdef__call__(self,scope,receive,send)-None:ifscope[type]!http:awaitself.app(scope,receive,send)returntry:awaitself.app(scope,receive,send)finally:RequestContext.clear_all()六、排查經(jīng)驗總結(jié)6.1 流式響應(yīng)排查清單檢查中間件鏈所有BaseHTTPMiddleware都可能緩沖流式響應(yīng)檢查事件傳遞路徑Agent → Queue → yield → 中間件 → 客戶端每一環(huán)都可能丟失事件檢查 Redis 狀態(tài)取消標(biāo)志、會話標(biāo)志等是否殘留檢查時序異步場景下的競爭條件特別是 setup 階段和事件循環(huán)之間的時間窗口6.2 關(guān)鍵日志模板# 請求開始logger.info(f[AGENT][STREAM] 開始: session_id{actual_session_id})# 清除標(biāo)志logger.info(f[AGENT][STREAM] 已清除殘留取消標(biāo)志:{cancel_key}(舊值{old_flag}))# 事件循環(huán)logger.info(f[AGENT][STREAM] 事件 #{event_count}:{event_type})# 取消檢測logger.info(f[AGENT] _is_cancelled: key{cancel_key}, flag{flag})6.3 最佳實踐場景推薦做法避免做法流式中間件純 ASGI 實現(xiàn)BaseHTTPMiddlewareAgent 流式傳輸直接yield事件asyncio.Queue間接層取消標(biāo)志管理請求開始時清除取消后立即清除依賴 TTL 自動過期Redis 實例共享使用request.app.state.redis每次創(chuàng)建新連接七、總結(jié)本文記錄了 FastAPI LlamaIndex Agent 流式對話的三個典型問題BaseHTTPMiddleware 緩沖流式響應(yīng)→ 轉(zhuǎn)為純 ASGI 中間件Queue 間接層導(dǎo)致事件丟失→ 直接流式傳輸Redis 取消標(biāo)志殘留 時序競爭→ 雙保險清除策略這三個問題獨立存在但相互關(guān)聯(lián)任何一個都可能導(dǎo)致流式響應(yīng)異常。排查時需要從中間件鏈、事件傳遞路徑、Redis 狀態(tài)、時序競爭等多個維度綜合分析。希望本文能幫助遇到類似問題的同學(xué)快速定位和解決。如果覺得有用歡迎點贊、收藏、轉(zhuǎn)發(fā)