博客（AI重做）)
FitMall 運動優(yōu)選一個健身電商平臺的三階段 AI 架構(gòu)演進全記錄從規(guī)則引擎到 Dify 低代碼編排再到 LangGraph 原生智能體——本文完整拆解一個全棧健身電商項目的技術(shù)選型、架構(gòu)設(shè)計與 AI 工程化的踩坑實錄。不只是接了個 AI而是一步步演進出企業(yè)級的 Agent 架構(gòu)。目錄項目概覽技術(shù)棧一覽架構(gòu)設(shè)計AI 架構(gòu)演進從 Dify 到 LangGraph為什么要重構(gòu)Agent 與 Workflow兩種形態(tài)的選型哲學(xué)混合檢索底座Milvus ES 雙路召回聊天 AgentReAct 循環(huán) 可跳轉(zhuǎn)卡片訓(xùn)練建議與計劃固定 Workflow 強校驗 JSON慢任務(wù)異步化Celery RabbitMQ多級緩存讓重復(fù)請求秒回降級鏈AI 永遠不成為單點那些有意思的工程細節(jié)踩坑實錄總結(jié)與思考項目概覽FitMall 是一個面向運動健身人群的全棧 Web 應(yīng)用承載了電商交易、課程學(xué)習(xí)、社區(qū)互動、訓(xùn)練管理、AI 智能助手五大業(yè)務(wù)線。前后端分離Docker Compose 一鍵部署。核心數(shù)據(jù)流瀏覽器 → Nginx (80) → Vue 3 SPA → /api/* → FastAPI (8000) ─┬→ MySQL / Redis / RabbitMQ ├→ Elasticsearch商品/課程/知識庫檢索 ├→ Milvus / Zilliz Cloud向量檢索 ├→ PostgreSQLLangGraph checkpoint 持久化 ├→ MinIO頭像/帖子圖/課程視頻S3 協(xié)議 └→ LLM APISiliconFlow 等LangGraph 編排AI 能力是本項目最有故事的部分它經(jīng)歷了三個階段階段形態(tài)問題V1BMI 規(guī)則引擎死板千人一面V2Dify 云端編排依賴外部平臺意圖識別靠提示詞硬湊商品推薦鏈路繞一圈公網(wǎng)V3LangGraph 原生當(dāng)前自主可控Agent Workflow 雙形態(tài)混合檢索異步化任務(wù)技術(shù)棧一覽后端組件技術(shù)說明Web 框架FastAPI Uvicorn全異步支持 SSE 流式響應(yīng)數(shù)據(jù)庫MySQL 8.0 SQLAlchemy 2.0異步引擎 aiomysql 驅(qū)動緩存 / 會話Redis 7會話、驗證碼、分布式鎖、AI 結(jié)果緩存、任務(wù)狀態(tài)搜索引擎Elasticsearch IK 分詞器商品/課程/帖子/用戶 知識庫 BM25向量數(shù)據(jù)庫MilvusZilliz Cloud知識庫稠密向量檢索Embeddingbge-m3SiliconFlow API1024 維中文語義向量Agent 編排LangGraph LangChain聊天 Agent 固定 WorkflowCheckpointPostgreSQL AsyncPostgresSaverAgent 狀態(tài)持久化服務(wù)重啟記憶不丟消息隊列RabbitMQ 4.xES 異步同步、訂單超時、訓(xùn)練通知 Celery Broker異步任務(wù)Celery 5.6solo 池AI 訓(xùn)練計劃后臺生成認(rèn)證JWT HTTP-Only Cookie雙通道鑒權(quán)Redis Session 存儲支付支付寶沙箱RSA2-SHA256自簽名實現(xiàn)不依賴官方 SDK存儲MinIOS3 協(xié)議 boto3自建對象存儲匿名只讀桶 預(yù)簽名 URL短信阿里云 DypnsapiRedis 存驗證碼5 分鐘過期ID 生成自研 Snowflake41 位毫秒時間戳 10 位機器 12 位序列前端組件技術(shù)說明框架Vue 3.5 Composition API TypeScriptscript setup語法構(gòu)建Vite秒級熱更新狀態(tài)管理Piniauser / admin / cart 三個 Store樣式TailwindCSS自定義 “Iron Chalk” 設(shè)計系統(tǒng)HTTPAxios 攔截器Cookie 自動攜帶 并發(fā)刷新攔截圖表Chart.js體重趨勢 / 訓(xùn)練統(tǒng)計WebSocket原生 WebSocket 自封裝 Composable實時聊天 在線狀態(tài)架構(gòu)設(shè)計領(lǐng)域驅(qū)動模塊劃分后端采用按業(yè)務(wù)域垂直拆分的模塊化結(jié)構(gòu)每個模塊內(nèi)部遵循統(tǒng)一的四層范式module/ ├── models.py # SQLAlchemy 數(shù)據(jù)模型與數(shù)據(jù)庫表一一對應(yīng) ├── schemas.py # Pydantic 請求/響應(yīng) SchemaAPI 契約 ├── router.py # FastAPI 路由定義薄層純路由注冊 └── service.py # 業(yè)務(wù)邏輯核心所有復(fù)雜邏輯在這里全局共12 個業(yè)務(wù)域auth goods course training community cart order ai admin user shared middleware其中ai域在 LangGraph 重構(gòu)后內(nèi)部又拆成了清晰的三層app/ai/ ├── agents/ # 聊天智能體ReAct 循環(huán)、工具集 │ ├── chat.py # LangGraph 圖agent - tools 回環(huán) │ └── tools.py # 商品/課程搜索、知識檢索、用戶畫像 三個工具 ├── workflows/ # 固定流程訓(xùn)練建議、訓(xùn)練計劃 │ ├── advice.py # 建議工作流 │ ├── plan.py # 計劃工作流generate - repair 校驗回環(huán) │ └── schemas.py # 輸出結(jié)構(gòu)Pydantic 強校驗 ├── rag/ # 檢索底座 │ ├── retriever.py # 混合檢索 RRF 融合 │ ├── milvus.py # 向量庫適配 │ ├── es_bm25.py # BM25 適配 │ ├── embedding.py # bge-m3 封裝 │ └── ingest.py # 語料切分入庫 ├── llm/ # 模型供應(yīng)商路由 ├── checkpoint.py # LangGraph 狀態(tài)持久化Postgres checkpointer 管理 ├── chat/ advice/ plan/ search/ # 各業(yè)務(wù) API路由服務(wù) └── dify_client.py # Dify 客戶端保留作降級通道 tasks/ ├── celery_app.py # Celery 實例brokerRabbitMQbackendRedis └── plan_tasks.py # 訓(xùn)練計劃后臺生成任務(wù)請求生命周期HTTP 請求 → CORS 中間件跨域白名單校驗 → Request-ID 中間件注入 UUID 追蹤鏈路 → 日志中間件記錄 method/path/status/duration → 路由匹配 → 依賴注入DB Session / Redis / Auth User → Service 層業(yè)務(wù)邏輯 AI 編排 外部調(diào)用 → 統(tǒng)一響應(yīng)格式 → {code: 1, data: {...}, message: success} → 異常處理器兜底6 種自定義異常 → HTTP 狀態(tài)碼映射AI 架構(gòu)演進從 Dify 到 LangGraph4.1 為什么要重構(gòu)Dify 階段的 AI 助手是這樣的用戶想買蛋白粉 → Dify ChatflowLLM 意圖識別 → {category: goods, keyword: 蛋白粉} → Dify HTTP 節(jié)點 → 公網(wǎng)回調(diào)自己的 API → ES 搜索 → LLM 格式化 → 推薦文案 product:// 鏈接能用但有三個企業(yè)級場景下的硬傷鏈路繞公網(wǎng)消息從自己的服務(wù)器出發(fā)繞到 Dify 云端再讓 Dify 回調(diào)自己的公網(wǎng) API 拿數(shù)據(jù)。一個搜商品的內(nèi)部操作數(shù)據(jù)出網(wǎng)兩次延遲和暴露面都不可控。編排黑盒意圖識別的提示詞、工具的調(diào)用時機全在 Dify 控制臺里代碼倉庫里沒有版本管理出問題只能登平臺排查。形態(tài)單一聊天、建議、計劃三種完全不同性質(zhì)的任務(wù)被塞進同一種工作流形態(tài)里計劃生成這種 2 分鐘級慢任務(wù)在同步請求里直接把接口拖死。重構(gòu)后的目標(biāo)很明確Agent 管聊天、Workflow 管生成、檢索自建、慢任務(wù)異步化、每一層都有降級。4.2 Agent 與 Workflow兩種形態(tài)的選型哲學(xué)LangGraph 提供了兩種建圖方式本項目兩種都用了各管一攤聊天助手Agent訓(xùn)練建議/計劃Workflow形態(tài)agent ? tools回環(huán)LLM 自主決策start → 步驟 → END固定直線類比一個有工具箱的員工自己判斷拿哪把工具一條流水線原料進去、成品出來適用輸入不可預(yù)測需要靈活路由輸入輸出固定追求穩(wěn)定可控失敗代價高工具調(diào)錯頂多答非所問低流程不會跑偏這個判斷是整個重構(gòu)的地基不要用 Agent 去做 Workflow 的事。用戶點生成訓(xùn)練計劃時不需要 AI 思考要不要先查一下用戶資料——這些步驟是確定的寫死就行確定性場景里自主性只會引入不穩(wěn)定。4.3 混合檢索底座Milvus ES 雙路召回健身知識庫深蹲要領(lǐng)、減脂原則、增肌營養(yǎng)等語料的檢索單一手段都有短板純向量檢索Milvus懂語義“怎么練腿” 能命中 “深蹲”“臀橋”但對精確術(shù)語如專有名詞不夠敏感純關(guān)鍵詞ES BM25精確匹配強但 “怎么練腿” 和 “下肢訓(xùn)練” 永遠匹配不上于是做了經(jīng)典的雙路召回 RRF 融合用戶 query │ ┌───────────┴───────────┐ bge-m3 向量化 IK 分詞 │ │ Milvus 近鄰檢索 ES BM25 檢索 語義召回 TopK 關(guān)鍵詞召回 TopK └───────────┬───────────┘ │ RRF 融合去重 score Σ 1 / (k rank) │ 歸一化 → min_score 過濾 │ 最終 TopK 上下文幾個關(guān)鍵實現(xiàn)細節(jié)1. 兩條路徑互相兜底任一失敗不炸整體# retriever.py — 雙路并行各自失敗只降級自己tasks{dense:self.milvus.search(query_vector,top_k...,filtersfilters),bm25:self.elasticsearch.search(query,top_k...,filtersfilters),}completed:dict[str,list]{}forname,taskintasks.items():try:completed[name]awaittaskexceptExceptionaserror:logger.warning(知識檢索路徑 [%s] 失敗跳過: %s,name,error)ES 掛了照樣能出純向量結(jié)果反之亦然。2. RRF 分?jǐn)?shù)歸一化讓min_score閾值有可比性RRF 原始分?jǐn)?shù)和有幾條路徑參與相關(guān)兩路都命中時分?jǐn)?shù)天然翻倍。除以理論最優(yōu)分所有活躍路徑都排第一的分?jǐn)?shù)把結(jié)果歸一到 0~1配置的置信度閾值才有穩(wěn)定含義# 歸一化除以所有活躍路徑都排第一時的最優(yōu)分?jǐn)?shù)best_possibleactive/(self.config.rrf_k1)scorescore/best_possible3. 置信度兜底最高分低于閾值時不硬編答案而是返回置信度不足狀態(tài)——寧可承認(rèn)不知道也不拿不相關(guān)的知識糊弄用戶。這在健身這種專業(yè)領(lǐng)域很重要一句錯誤的訓(xùn)練建議可能導(dǎo)致受傷。4. 入庫側(cè)語料按標(biāo)題/段落切分bge-m3 向量化后同時寫入 Milvus帶knowledge_domain元數(shù)據(jù)過濾和 ES 知識索引BM25 用同一份內(nèi)容雙格式存儲檢索側(cè)天然對齊。4.4 聊天 AgentReAct 循環(huán) 可跳轉(zhuǎn)卡片聊天 Agent 用 LangGraph 的經(jīng)典agent ? tools回環(huán)結(jié)構(gòu)START → agentLLM 思考 ──有 tool_calls──→ tools執(zhí)行工具 ↑ │ └────────── 結(jié)果回填 ←──────────────┘ 無 tool_calls 或達到步數(shù)上限 → END三個工具覆蓋三類典型問題工具觸發(fā)場景數(shù)據(jù)來源search_goods_or_courses“想買蛋白粉”“有減脂課嗎”ES復(fù)用商城搜索服務(wù)search_fitness_knowledge“深蹲膝蓋疼怎么回事”混合檢索Milvus ESget_user_training_profile“按我的情況該練什么”MySQL畫像 近 30 天統(tǒng)計卡片推薦是體驗上的關(guān)鍵設(shè)計工具返回的每個商品/課程不只進 LLM 上下文還同時收進一個cards_holder容器。Agent 跑完后SSE 流里額外推一個cards事件前端渲染成可點擊跳轉(zhuǎn)的卡片# agents/chat.py — 流式產(chǎn)出兩類事件asyncforchunk,_metadataingraph.astream({messages:input_messages,step:0},stream_modemessages):# 工具調(diào)用輪次不產(chǎn)生正文只產(chǎn)生 tool_call 參數(shù)流跳過避免回顯 JSONifisinstance(chunk,AIMessageChunk)andchunk.tool_calls:continuetextgetattr(chunk,text,)oriftext:yield{content:text}ifcards_holder:yield{cards:cards_holder}# 最多 6 張去重限量為什么卡片不交給 LLM 輸出系統(tǒng)提示詞里寫死了鐵律“只使用工具返回的真實商品/課程絕不編造名稱、價格或鏈接”。卡片數(shù)據(jù)從工具結(jié)果直通前端LLM 只負責(zé)組織推薦語言——這是從根源上杜絕 AI 幻覺出假商品的做法。模型哪怕再想編它也沒有輸出卡片的通道。防失控max_steps6步數(shù)上限防止模型陷入調(diào)工具→不滿意→再調(diào)工具的死循環(huán)燒 Token。多輪記憶雙模式Agent 的會話狀態(tài)通過AsyncPostgresSaver寫入 Postgres checkpointthread_id直接復(fù)用聊天會話 ID——服務(wù)重啟、多 worker 部署后新一輪請求只傳新增的那條消息歷史由 checkpoint 恢復(fù)圖從中斷處無縫接力。Postgres 不可用時自動降級為「MySQL 歷史全量重放」模式把AiMessage表的全部歷史拼進初始messages功能不丟只是每次全量重放。刪除會話時同步清理對應(yīng) checkpoint thread不留孤兒數(shù)據(jù)。4.5 訓(xùn)練建議與計劃固定 Workflow 強校驗 JSON訓(xùn)練計劃生成是典型的確定場景輸入目標(biāo)/周期/水平/可用日/偏好和輸出周×天嵌套的結(jié)構(gòu)化計劃都是固定的。它的 Workflow 長這樣START → generate結(jié)構(gòu)化輸出 ──校驗通過──→ END ↑ │ │ 校驗失敗且重試 2 └──── repair帶錯誤信息重生成←┘核心技巧是with_structured_output——不是讓 LLM “輸出 JSON”那要靠祈禱而是用 Pydantic Schema 約束輸出結(jié)構(gòu)框架層保證解析# workflows/plan.py — generate 節(jié)點asyncdefgenerate(state:_PlanState)-dict:boundmodel.with_structured_output(PlanOutput)# 強校驗resultawaitbound.ainvoke([(system,_PLAN_SYSTEM_PROMPT),(user,f請基于以下要求生成訓(xùn)練計劃\n{request}),])return{output:payload,error:}# 成功# 失敗時 return {error: str(e), repairs: 1} # 進 repair 環(huán)repair 環(huán)是亮點校驗失敗不是直接拋錯而是把錯誤信息拼進提示詞讓模型自己修最多修 2 次。類比成代碼編譯報錯后讓 AI 自己改 bug——大部分格式問題一輪就修好用戶體驗從失敗重試變成稍慢一點但成功。修復(fù)仍失敗則返回None由上層降級到本地模板計劃減脂/增肌/塑形/保持 四套模板保證接口永遠有可用輸出。4.6 慢任務(wù)異步化Celery RabbitMQ問題完整訓(xùn)練計劃生成要 2~2.5 分鐘LLM 生成長 JSON 校驗。同步接口意味著用戶盯著轉(zhuǎn)圈 2 分鐘、HTTP 超時風(fēng)險、uvicorn worker 被占死。方案生成任務(wù)交給 Celery workerAPI 立即返回前端輪詢。POST /ai/plan/generate │ ├─ Redis 查任務(wù)狀態(tài)已 done → 直接返回計劃緩存命中 ├─ 狀態(tài) pending_review → 返回 pending去走確認(rèn)流程不重復(fù)生成 ├─ 狀態(tài) running → 返回 pending去重不重復(fù)派發(fā) └─ 全新請求 → 標(biāo)記 running → Celery 派發(fā) → 返回 {status:pending, task_key} Celery worker獨立進程 └─ LangGraph 生成草稿 → Redis 寫 {status:pending_review, weekly_plan} 后端先不落庫等用戶確認(rèn) 前端每 3 秒 GET /ai/plan/status?task_keyxxx └─ pending 繼續(xù)等 / pending_review 彈「草稿確認(rèn)框」→ 采納/放棄 done 渲染計劃 / failed 提示錯誤 POST /ai/plan/confirm {task_key, action} └─ actionaccept → 落庫停用舊計劃啟用新計劃→ status 變 done actionreject → status 變 rejected丟棄派發(fā)層還有一層降級RabbitMQ 不可用時自動回退到進程內(nèi)asyncio.create_task單機開發(fā)環(huán)境不裝 MQ 也能跑# plan/router.py — 派發(fā)優(yōu)先級def_dispatch_plan(task_key,user_id,plan_request)-bool:try:fromtasks.plan_tasksimportgenerate_plan_task generate_plan_task.delay(task_key,user_id,plan_request)# CeleryreturnTrueexceptException:returnstart_background_generation(...)# 進程內(nèi) asyncio 兜底Celery 任務(wù)里的一個深坑任務(wù)是同步函數(shù)內(nèi)部跑的是 asyncio 代碼LangGraph 生成所以每個任務(wù)用asyncio.run()自建事件循環(huán)。落庫動作不在任務(wù)里做——因為 Human-in-the-loop 需要先生成草稿、等用戶確認(rèn)再落庫所以任務(wù)只負責(zé)生成并寫 Redis真實庫寫發(fā)生在主進程的 confirm 接口里async 代碼直接復(fù)用主進程 engine無 loop 沖突。搭配-P solo單并發(fā)池 任務(wù)內(nèi)獨立 loop 才穩(wěn)。Worker 啟動方式RabbitMQ 4.x 有兼容配置見踩坑實錄python-mcelery-Atasks.celery_app worker-Psolo--loglevelinfo4.7 多級緩存讓重復(fù)請求秒回三類緩存各司其職緩存KeyTTL失效邏輯訓(xùn)練建議用戶畫像 近 30 天統(tǒng)計 的 SHA16 小時畫像/訓(xùn)練數(shù)據(jù)一變key 自然變化自動失效訓(xùn)練計劃生成請求參數(shù)的 SHA17 天換目標(biāo)/周期即新 key任務(wù)狀態(tài)ai:plan:task:{key}1 小時輪詢專用含 done/running/failed建議緩存的 key 設(shè)計值得一提不是簡單的user:{id}而是把影響建議內(nèi)容的所有因子目標(biāo)、水平、身高體重、BMI、30 天訓(xùn)練量、時長、卡路里hash 進 key。好處是天然自失效——用戶今天練了一節(jié)課、明早稱了體重明天的請求 key 就變了自動重新生成而數(shù)據(jù)沒變化時相同請求直接命中秒回。計劃接口的三段式# 1. 已有完成結(jié)果緩存/任務(wù) done直接返回taskawait_task_get(task_key)iftaskandtask.get(status)done:returncreated(PlanResponse(...))# 2. 已在生成中去重不重復(fù)創(chuàng)建任務(wù)iftaskandtask.get(status)running:returncreated({status:pending,task_key:task_key})# 3. 先標(biāo)記 running再派發(fā)后臺生成# 先標(biāo)記后派發(fā)的順序很關(guān)鍵避免任務(wù)瞬間完成被 running 覆蓋await_task_set_running(task_key)_dispatch_plan(...)注意第 3 步的注釋——先標(biāo)記 running 再派發(fā)因為極端情況下 Celery 任務(wù)可能秒級完成緩存命中 LLM 響應(yīng)后寫 running 會把 done 結(jié)果覆蓋回 running前端永遠輪詢不到結(jié)果。4.8 降級鏈AI 永遠不成為單點每個 AI 功能都是三級火箭逐級點火聊天 LangGraph Agent → Dify 兜底 → 本地占位回復(fù) Agent記憶 Postgres checkpoint → MySQL 歷史全量重放 建議 LangGraph Workflow → Dify → BMI 規(guī)則引擎5 級分類 計劃 LangGraph Workflow → 本地模板計劃4 種目標(biāo) 計劃派發(fā)Celery(RabbitMQ) → 進程內(nèi) asyncio 知識檢索Milvus ES 雙路 → 單路降級 → 知識庫暫不可用實測中最底層從不缺席即使 LLM API、Dify、Milvus、RabbitMQ全部同時掛掉用戶依然能拿到一份 BMI 分級的文字建議和一份模板訓(xùn)練計劃。AI 是錦上添花不是核心鏈路——這條原則貫穿了整個設(shè)計。那些有意思的工程細節(jié)1. HTTP-Only Cookie 會話認(rèn)證用 Redis 替代 JWT傳統(tǒng) JWT 的痛點Token 存localStorageXSS 腳本可竊取泄露后在過期前無法撤銷。FitMall 的方案登錄 → 生成 Snowflake Session Token → 用戶信息存 Rediskey: session:{token}, TTL: 24h → Token 寫入 HTTP-Only Cookie前端 JS 不可讀 → 后端解析 Cookie → Redis 查找 → 注入當(dāng)前用戶防 XSSHTTP-Only Cookie 對 JavaScript 完全不可見即時失效刪除 Redis Key 即可讓任意 Token 立即失效用戶/管理端隔離兩套獨立 Cookie 名和 Redis 前綴杜絕越權(quán)自動續(xù)期每次活躍請求刷新 TTL同時保留 JWT 作為 WebSocket 的鑒權(quán)通道瀏覽器 WebSocket API 無法自定義 Cookie Header雙通道并存。2. 手搓 Snowflake 分布式 ID 生成器┌─┬──────────────────────────┬──────────────┬──────────────┐ │0│ 41-bit 時間戳 (ms) │ 10-bit 機器碼 │ 12-bit 序列號 │ └─┴──────────────────────────┴──────────────┴──────────────┘自定義紀(jì)元2024-01-01可用到 2094 年、時鐘回撥容忍 5ms、threading.Lock保證線程安全、整數(shù)/Hex 雙輸出庫表主鍵 / URL 各取所需。3. 雙通道 ES 同步直寫 消息隊列兜底┌─ 通道 1: 直接寫 ES用戶立即搜到 CRUD 操作 ──────────┤ └─ 通道 2: RabbitMQ 消息 → 異步消費重試 ↑ ES 臨時不可用時兜底重放直寫保證管理員剛上架就能搜到消息隊列保證 ES 恢復(fù)后數(shù)據(jù)最終一致。消費端指數(shù)退避重試5s→10s→20s→40s→60s最多 5 次。4. IK 分詞器的雙分析器策略{ik_index_analyzer:{tokenizer:ik_max_word},// 索引細粒度高召回ik_search_analyzer:{tokenizer:ik_smart}// 搜索粗粒度高精度}“增肌蛋白粉” 索引時切成盡可能多的詞條任意組合可命中搜索時只做粗粒度切分避免蘋果手機匹配到水果。這套策略同樣服務(wù)于知識庫 BM25 檢索。5. 訂單超時取消與 Redis 分布式鎖下單后 30 分鐘未支付自動取消并回滾庫存。RabbitMQ 消息 消費端sleep剩余時間 RedisSET NX EX分布式鎖防多實例并發(fā)處理。簡單直接適合單實例場景大規(guī)模需換死信隊列。6. WebSocket 實時聊天與在線狀態(tài)內(nèi)存字典存本進程連接 Redis 集合存跨進程在線用戶雙端同步注冊/清理WebSocket 失敗自動降級 HTTP POST 發(fā)消息未讀計數(shù)實時推送。7. 對象存儲從阿里云 OSS 遷到自建 MinIO早期用的阿里云 OSSSDKoss2后來主動遷到了自建 MinIO理由有三成本OSS 按存儲 流量計費演示項目不值當(dāng)MinIO 部署在自家服務(wù)器零邊際成本。自家的井MySQL/Redis/ES/RabbitMQ/Postgres 全在自建服務(wù)器上再掛個云 OSS 屬于自家有井還去買水還多一份外部依賴要維護。遷移成本幾乎為零MinIO 完全兼容 S3 協(xié)議SDK 換成boto3S3 事實標(biāo)準(zhǔn)對外接口簽名upload_file/delete_file保持不變?nèi)齻€調(diào)用點零改動。兩個協(xié)議差異點值得記path-style 尋址MinIO 不支持 AWS 的 virtual-host 風(fēng)格URL 是endpoint/bucket/key和公開讀策略MinIO 用桶策略 JSON 授權(quán)匿名GetObject替代 OSS 的公共讀桶。業(yè)務(wù)層保留指數(shù)退避重試和預(yù)簽名 URL 能力語義完全對齊。踩坑實錄重構(gòu)過程真機聯(lián)調(diào)修掉的坑每一個都值得記錄1. RabbitMQ 4.3 拒絕 Celery 的 transient 隊列Worker 一啟動就拋541 INTERNAL_ERROR。根因RabbitMQ 4.3 起默認(rèn)禁用transient 非排他隊列而 Celery 的 control/mingle/gossip 廣播隊列正是這種類型。解法純代碼側(cè)不動服務(wù)器# celery_app.pyworker_disable_mingleTrue,worker_disable_gossipTrue,worker_enable_remote_controlFalse,control_queue_durableTrue,event_queue_durableTrue,單 worker 場景這些廣播本來就沒用關(guān)掉反而干凈。2. SQLAlchemy engine 與 event loop 的綁定Celery 任務(wù)里復(fù)用主進程的數(shù)據(jù)庫 engine 會炸——連接池綁定已關(guān)閉的 loop。每個任務(wù)必須create_async_engine自建、用完dispose()。3. Embedding 服務(wù)的dimensions參數(shù)SiliconFlow 的 bge-m3 不接受dimensions參數(shù)加了直接 400。做成配置開關(guān)EMBEDDING_SEND_DIMENSIONS默認(rèn)不發(fā)送。4. LangGraph 狀態(tài)累積用普通TypedDict定義狀態(tài)消息無法跨節(jié)點累積每個節(jié)點返回值直接覆蓋。改繼承MessagesState其內(nèi)置的add_messagesreducer 自動做追加合并。5.tool裝飾器漏加工具函數(shù)寫了但沒掛toolLangChain 不識別Agent 永遠不調(diào)用它。工具存在和可被 Agent 發(fā)現(xiàn)是兩回事。6. 工具調(diào)用流的正文回顯stream_modemessages下工具調(diào)用輪次也會產(chǎn)出 chunk內(nèi)容是 tool_call 參數(shù) JSON。不過濾的話用戶會看到聊天框里突然刷出一坨原始 JSON。按chunk.tool_calls跳過即可。7. 先標(biāo)記 running 再派發(fā)順序反了會出現(xiàn)任務(wù)毫秒級完成寫 done → 主協(xié)程再寫 running 把它覆蓋 → 前端永遠輪詢不到結(jié)果。并發(fā)寫同一個 key 時想清楚時序。8. 依賴版本沖突langchain-openai 1.6要求langchain-core1.6langgraph-checkpoint-postgres的正確版本區(qū)間是2.0,3.0。AI 生態(tài)迭代極快鎖版本區(qū)間前先看清依賴樹。9. 新版 uvicorn 在 Windows 硬編碼 ProactorEventLoop接入 Postgres checkpoint 后本地啟動一直降級——psycopg 的 async 模式只支持 SelectorEventLoop而這版 uvicorn 的asyncio_loop_factory在 Windows 上直接return asyncio.ProactorEventLoop且 uvicorn 先建事件循環(huán)、后導(dǎo)入 app 模塊所以在 main.py 里設(shè) event loop policy 永遠太晚。解法是啟動命令顯式傳自定義 looppython-muvicorn app.main:app--loopasyncio:SelectorEventLoop--loop支持任意module:attribute導(dǎo)入串SelectorEventLoop跨平臺存在Linux 本來就是默認(rèn)這條命令兩邊通用。典型教訓(xùn)event loop policy 只影響未來的循環(huán)框架自己建循環(huán)時 policy 就成了擺設(shè)。10. psycopg 連接池必須 autocommitAsyncPostgresSaver.setup()建表時會執(zhí)行CREATE INDEX CONCURRENTLY這條 SQL不能跑在事務(wù)里而 psycopg 默認(rèn)開事務(wù)。自建連接池時必須傳kwargs{autocommit: True, prepare_threshold: 0, row_factory: dict_row}——這三個參數(shù)抄官方from_conn_string的實現(xiàn)就對了自作聰明省掉任何一個都會在奇怪的地方報錯?？偨Y(jié)與思考這輪重構(gòu)做對了什么形態(tài)分離Agent 管不可預(yù)測的對話Workflow 管確定性的生成。用對工具比用好工具更重要?；糜X從架構(gòu)層封死商品卡片數(shù)據(jù)從工具直通前端LLM 沒有輸出卡片的通道——不靠提示詞祈禱靠數(shù)據(jù)通路設(shè)計。慢任務(wù)異步化2.5 分鐘的生成從接口卡死變成立即反饋 進度輪詢且?guī)Ь彺婧腿ブ?。狀態(tài)真正持久化Agent 會話狀態(tài)落 Postgres checkpoint服務(wù)重啟記憶不丟——多輪記憶從每次全量重放變成只傳增量、斷點接力。每一層都降級AI 服務(wù)掛了系統(tǒng)照常運轉(zhuǎn)只是變笨不變癱?？梢岳^續(xù)演進的方向意圖分流前置先用小模型分類閑聊/咨詢/導(dǎo)購閑聊直達 LLM 省去工具綁定開銷計劃生成提速分段生成 并發(fā)拼裝每周一個子任務(wù)2.5 分鐘壓到 30 秒級訂單超時的sleep方案換 RabbitMQ 死信隊列支撐多實例部署更強 Human-in-the-loop當(dāng)前是生成后確認(rèn)草稿結(jié)合已有 checkpoint 可演進為生成前 interrupt 等用戶確認(rèn)參數(shù)——圖跑到一半停下來用戶點確認(rèn)后從斷點繼續(xù)項目數(shù)據(jù)后端12 個業(yè)務(wù)模塊 AI 三層架構(gòu)agents/workflows/rag前端~40 個 Vue 組件3 個 Pinia Store基礎(chǔ)設(shè)施Docker Compose 編排MySQL / Redis / ES / RabbitMQ / Postgres / MinIO / Milvus云端AI 鏈路1 個 ReAct Agent3 工具、2 個固定 Workflow、雙路混合檢索、Celery 異步任務(wù)、三級緩存、Postgres checkpoint 持久化從調(diào)一個 API到設(shè)計一套 AI 架構(gòu)這個項目最大的收獲是AI 工程化的核心不是模型是數(shù)據(jù)通路、失敗路徑和響應(yīng)時間的設(shè)計。模型會越來越好但這些工程問題永遠存在。 FitMall技術(shù)博客（AI重做）
FitMall 運動優(yōu)選一個健身電商平臺的三階段 AI 架構(gòu)演進全記錄從規(guī)則引擎到 Dify 低代碼編排再到 LangGraph 原生智能體——本文完整拆解一個全棧健身電商項目的技術(shù)選型、架構(gòu)設(shè)計與 AI 工程化的踩坑實錄。不只是接了個 AI而是一步步演進出企業(yè)級的 Agent 架構(gòu)。目錄項目概覽技術(shù)棧一覽架構(gòu)設(shè)計AI 架構(gòu)演進從 Dify 到 LangGraph為什么要重構(gòu)Agent 與 Workflow兩種形態(tài)的選型哲學(xué)混合檢索底座Milvus ES 雙路召回聊天 AgentReAct 循環(huán) 可跳轉(zhuǎn)卡片訓(xùn)練建議與計劃固定 Workflow 強校驗 JSON慢任務(wù)異步化Celery RabbitMQ多級緩存讓重復(fù)請求秒回降級鏈AI 永遠不成為單點那些有意思的工程細節(jié)踩坑實錄總結(jié)與思考項目概覽FitMall 是一個面向運動健身人群的全棧 Web 應(yīng)用承載了電商交易、課程學(xué)習(xí)、社區(qū)互動、訓(xùn)練管理、AI 智能助手五大業(yè)務(wù)線。前后端分離Docker Compose 一鍵部署。核心數(shù)據(jù)流瀏覽器 → Nginx (80) → Vue 3 SPA → /api/* → FastAPI (8000) ─┬→ MySQL / Redis / RabbitMQ ├→ Elasticsearch商品/課程/知識庫檢索 ├→ Milvus / Zilliz Cloud向量檢索 ├→ PostgreSQLLangGraph checkpoint 持久化 ├→ MinIO頭像/帖子圖/課程視頻S3 協(xié)議 └→ LLM APISiliconFlow 等LangGraph 編排AI 能力是本項目最有故事的部分它經(jīng)歷了三個階段階段形態(tài)問題V1BMI 規(guī)則引擎死板千人一面V2Dify 云端編排依賴外部平臺意圖識別靠提示詞硬湊商品推薦鏈路繞一圈公網(wǎng)V3LangGraph 原生當(dāng)前自主可控Agent Workflow 雙形態(tài)混合檢索異步化任務(wù)技術(shù)棧一覽后端組件技術(shù)說明Web 框架FastAPI Uvicorn全異步支持 SSE 流式響應(yīng)數(shù)據(jù)庫MySQL 8.0 SQLAlchemy 2.0異步引擎 aiomysql 驅(qū)動緩存 / 會話Redis 7會話、驗證碼、分布式鎖、AI 結(jié)果緩存、任務(wù)狀態(tài)搜索引擎Elasticsearch IK 分詞器商品/課程/帖子/用戶 知識庫 BM25向量數(shù)據(jù)庫MilvusZilliz Cloud知識庫稠密向量檢索Embeddingbge-m3SiliconFlow API1024 維中文語義向量Agent 編排LangGraph LangChain聊天 Agent 固定 WorkflowCheckpointPostgreSQL AsyncPostgresSaverAgent 狀態(tài)持久化服務(wù)重啟記憶不丟消息隊列RabbitMQ 4.xES 異步同步、訂單超時、訓(xùn)練通知 Celery Broker異步任務(wù)Celery 5.6solo 池AI 訓(xùn)練計劃后臺生成認(rèn)證JWT HTTP-Only Cookie雙通道鑒權(quán)Redis Session 存儲支付支付寶沙箱RSA2-SHA256自簽名實現(xiàn)不依賴官方 SDK存儲MinIOS3 協(xié)議 boto3自建對象存儲匿名只讀桶 預(yù)簽名 URL短信阿里云 DypnsapiRedis 存驗證碼5 分鐘過期ID 生成自研 Snowflake41 位毫秒時間戳 10 位機器 12 位序列前端組件技術(shù)說明框架Vue 3.5 Composition API TypeScriptscript setup語法構(gòu)建Vite秒級熱更新狀態(tài)管理Piniauser / admin / cart 三個 Store樣式TailwindCSS自定義 “Iron Chalk” 設(shè)計系統(tǒng)HTTPAxios 攔截器Cookie 自動攜帶 并發(fā)刷新攔截圖表Chart.js體重趨勢 / 訓(xùn)練統(tǒng)計WebSocket原生 WebSocket 自封裝 Composable實時聊天 在線狀態(tài)架構(gòu)設(shè)計領(lǐng)域驅(qū)動模塊劃分后端采用按業(yè)務(wù)域垂直拆分的模塊化結(jié)構(gòu)每個模塊內(nèi)部遵循統(tǒng)一的四層范式module/ ├── models.py # SQLAlchemy 數(shù)據(jù)模型與數(shù)據(jù)庫表一一對應(yīng) ├── schemas.py # Pydantic 請求/響應(yīng) SchemaAPI 契約 ├── router.py # FastAPI 路由定義薄層純路由注冊 └── service.py # 業(yè)務(wù)邏輯核心所有復(fù)雜邏輯在這里全局共12 個業(yè)務(wù)域auth goods course training community cart order ai admin user shared middleware其中ai域在 LangGraph 重構(gòu)后內(nèi)部又拆成了清晰的三層app/ai/ ├── agents/ # 聊天智能體ReAct 循環(huán)、工具集 │ ├── chat.py # LangGraph 圖agent - tools 回環(huán) │ └── tools.py # 商品/課程搜索、知識檢索、用戶畫像 三個工具 ├── workflows/ # 固定流程訓(xùn)練建議、訓(xùn)練計劃 │ ├── advice.py # 建議工作流 │ ├── plan.py # 計劃工作流generate - repair 校驗回環(huán) │ └── schemas.py # 輸出結(jié)構(gòu)Pydantic 強校驗 ├── rag/ # 檢索底座 │ ├── retriever.py # 混合檢索 RRF 融合 │ ├── milvus.py # 向量庫適配 │ ├── es_bm25.py # BM25 適配 │ ├── embedding.py # bge-m3 封裝 │ └── ingest.py # 語料切分入庫 ├── llm/ # 模型供應(yīng)商路由 ├── checkpoint.py # LangGraph 狀態(tài)持久化Postgres checkpointer 管理 ├── chat/ advice/ plan/ search/ # 各業(yè)務(wù) API路由服務(wù) └── dify_client.py # Dify 客戶端保留作降級通道 tasks/ ├── celery_app.py # Celery 實例brokerRabbitMQbackendRedis └── plan_tasks.py # 訓(xùn)練計劃后臺生成任務(wù)請求生命周期HTTP 請求 → CORS 中間件跨域白名單校驗 → Request-ID 中間件注入 UUID 追蹤鏈路 → 日志中間件記錄 method/path/status/duration → 路由匹配 → 依賴注入DB Session / Redis / Auth User → Service 層業(yè)務(wù)邏輯 AI 編排 外部調(diào)用 → 統(tǒng)一響應(yīng)格式 → {code: 1, data: {...}, message: success} → 異常處理器兜底6 種自定義異常 → HTTP 狀態(tài)碼映射AI 架構(gòu)演進從 Dify 到 LangGraph4.1 為什么要重構(gòu)Dify 階段的 AI 助手是這樣的用戶想買蛋白粉 → Dify ChatflowLLM 意圖識別 → {category: goods, keyword: 蛋白粉} → Dify HTTP 節(jié)點 → 公網(wǎng)回調(diào)自己的 API → ES 搜索 → LLM 格式化 → 推薦文案 product:// 鏈接能用但有三個企業(yè)級場景下的硬傷鏈路繞公網(wǎng)消息從自己的服務(wù)器出發(fā)繞到 Dify 云端再讓 Dify 回調(diào)自己的公網(wǎng) API 拿數(shù)據(jù)。一個搜商品的內(nèi)部操作數(shù)據(jù)出網(wǎng)兩次延遲和暴露面都不可控。編排黑盒意圖識別的提示詞、工具的調(diào)用時機全在 Dify 控制臺里代碼倉庫里沒有版本管理出問題只能登平臺排查。形態(tài)單一聊天、建議、計劃三種完全不同性質(zhì)的任務(wù)被塞進同一種工作流形態(tài)里計劃生成這種 2 分鐘級慢任務(wù)在同步請求里直接把接口拖死。重構(gòu)后的目標(biāo)很明確Agent 管聊天、Workflow 管生成、檢索自建、慢任務(wù)異步化、每一層都有降級。4.2 Agent 與 Workflow兩種形態(tài)的選型哲學(xué)LangGraph 提供了兩種建圖方式本項目兩種都用了各管一攤聊天助手Agent訓(xùn)練建議/計劃Workflow形態(tài)agent ? tools回環(huán)LLM 自主決策start → 步驟 → END固定直線類比一個有工具箱的員工自己判斷拿哪把工具一條流水線原料進去、成品出來適用輸入不可預(yù)測需要靈活路由輸入輸出固定追求穩(wěn)定可控失敗代價高工具調(diào)錯頂多答非所問低流程不會跑偏這個判斷是整個重構(gòu)的地基不要用 Agent 去做 Workflow 的事。用戶點生成訓(xùn)練計劃時不需要 AI 思考要不要先查一下用戶資料——這些步驟是確定的寫死就行確定性場景里自主性只會引入不穩(wěn)定。4.3 混合檢索底座Milvus ES 雙路召回健身知識庫深蹲要領(lǐng)、減脂原則、增肌營養(yǎng)等語料的檢索單一手段都有短板純向量檢索Milvus懂語義“怎么練腿” 能命中 “深蹲”“臀橋”但對精確術(shù)語如專有名詞不夠敏感純關(guān)鍵詞ES BM25精確匹配強但 “怎么練腿” 和 “下肢訓(xùn)練” 永遠匹配不上于是做了經(jīng)典的雙路召回 RRF 融合用戶 query │ ┌───────────┴───────────┐ bge-m3 向量化 IK 分詞 │ │ Milvus 近鄰檢索 ES BM25 檢索 語義召回 TopK 關(guān)鍵詞召回 TopK └───────────┬───────────┘ │ RRF 融合去重 score Σ 1 / (k rank) │ 歸一化 → min_score 過濾 │ 最終 TopK 上下文幾個關(guān)鍵實現(xiàn)細節(jié)1. 兩條路徑互相兜底任一失敗不炸整體# retriever.py — 雙路并行各自失敗只降級自己tasks{dense:self.milvus.search(query_vector,top_k...,filtersfilters),bm25:self.elasticsearch.search(query,top_k...,filtersfilters),}completed:dict[str,list]{}forname,taskintasks.items():try:completed[name]awaittaskexceptExceptionaserror:logger.warning(知識檢索路徑 [%s] 失敗跳過: %s,name,error)ES 掛了照樣能出純向量結(jié)果反之亦然。2. RRF 分?jǐn)?shù)歸一化讓min_score閾值有可比性RRF 原始分?jǐn)?shù)和有幾條路徑參與相關(guān)兩路都命中時分?jǐn)?shù)天然翻倍。除以理論最優(yōu)分所有活躍路徑都排第一的分?jǐn)?shù)把結(jié)果歸一到 0~1配置的置信度閾值才有穩(wěn)定含義# 歸一化除以所有活躍路徑都排第一時的最優(yōu)分?jǐn)?shù)best_possibleactive/(self.config.rrf_k1)scorescore/best_possible3. 置信度兜底最高分低于閾值時不硬編答案而是返回置信度不足狀態(tài)——寧可承認(rèn)不知道也不拿不相關(guān)的知識糊弄用戶。這在健身這種專業(yè)領(lǐng)域很重要一句錯誤的訓(xùn)練建議可能導(dǎo)致受傷。4. 入庫側(cè)語料按標(biāo)題/段落切分bge-m3 向量化后同時寫入 Milvus帶knowledge_domain元數(shù)據(jù)過濾和 ES 知識索引BM25 用同一份內(nèi)容雙格式存儲檢索側(cè)天然對齊。4.4 聊天 AgentReAct 循環(huán) 可跳轉(zhuǎn)卡片聊天 Agent 用 LangGraph 的經(jīng)典agent ? tools回環(huán)結(jié)構(gòu)START → agentLLM 思考 ──有 tool_calls──→ tools執(zhí)行工具 ↑ │ └────────── 結(jié)果回填 ←──────────────┘ 無 tool_calls 或達到步數(shù)上限 → END三個工具覆蓋三類典型問題工具觸發(fā)場景數(shù)據(jù)來源search_goods_or_courses“想買蛋白粉”“有減脂課嗎”ES復(fù)用商城搜索服務(wù)search_fitness_knowledge“深蹲膝蓋疼怎么回事”混合檢索Milvus ESget_user_training_profile“按我的情況該練什么”MySQL畫像 近 30 天統(tǒng)計卡片推薦是體驗上的關(guān)鍵設(shè)計工具返回的每個商品/課程不只進 LLM 上下文還同時收進一個cards_holder容器。Agent 跑完后SSE 流里額外推一個cards事件前端渲染成可點擊跳轉(zhuǎn)的卡片# agents/chat.py — 流式產(chǎn)出兩類事件asyncforchunk,_metadataingraph.astream({messages:input_messages,step:0},stream_modemessages):# 工具調(diào)用輪次不產(chǎn)生正文只產(chǎn)生 tool_call 參數(shù)流跳過避免回顯 JSONifisinstance(chunk,AIMessageChunk)andchunk.tool_calls:continuetextgetattr(chunk,text,)oriftext:yield{content:text}ifcards_holder:yield{cards:cards_holder}# 最多 6 張去重限量為什么卡片不交給 LLM 輸出系統(tǒng)提示詞里寫死了鐵律“只使用工具返回的真實商品/課程絕不編造名稱、價格或鏈接”。卡片數(shù)據(jù)從工具結(jié)果直通前端LLM 只負責(zé)組織推薦語言——這是從根源上杜絕 AI 幻覺出假商品的做法。模型哪怕再想編它也沒有輸出卡片的通道。防失控max_steps6步數(shù)上限防止模型陷入調(diào)工具→不滿意→再調(diào)工具的死循環(huán)燒 Token。多輪記憶雙模式Agent 的會話狀態(tài)通過AsyncPostgresSaver寫入 Postgres checkpointthread_id直接復(fù)用聊天會話 ID——服務(wù)重啟、多 worker 部署后新一輪請求只傳新增的那條消息歷史由 checkpoint 恢復(fù)圖從中斷處無縫接力。Postgres 不可用時自動降級為「MySQL 歷史全量重放」模式把AiMessage表的全部歷史拼進初始messages功能不丟只是每次全量重放。刪除會話時同步清理對應(yīng) checkpoint thread不留孤兒數(shù)據(jù)。4.5 訓(xùn)練建議與計劃固定 Workflow 強校驗 JSON訓(xùn)練計劃生成是典型的確定場景輸入目標(biāo)/周期/水平/可用日/偏好和輸出周×天嵌套的結(jié)構(gòu)化計劃都是固定的。它的 Workflow 長這樣START → generate結(jié)構(gòu)化輸出 ──校驗通過──→ END ↑ │ │ 校驗失敗且重試 2 └──── repair帶錯誤信息重生成←┘核心技巧是with_structured_output——不是讓 LLM “輸出 JSON”那要靠祈禱而是用 Pydantic Schema 約束輸出結(jié)構(gòu)框架層保證解析# workflows/plan.py — generate 節(jié)點asyncdefgenerate(state:_PlanState)-dict:boundmodel.with_structured_output(PlanOutput)# 強校驗resultawaitbound.ainvoke([(system,_PLAN_SYSTEM_PROMPT),(user,f請基于以下要求生成訓(xùn)練計劃\n{request}),])return{output:payload,error:}# 成功# 失敗時 return {error: str(e), repairs: 1} # 進 repair 環(huán)repair 環(huán)是亮點校驗失敗不是直接拋錯而是把錯誤信息拼進提示詞讓模型自己修最多修 2 次。類比成代碼編譯報錯后讓 AI 自己改 bug——大部分格式問題一輪就修好用戶體驗從失敗重試變成稍慢一點但成功。修復(fù)仍失敗則返回None由上層降級到本地模板計劃減脂/增肌/塑形/保持 四套模板保證接口永遠有可用輸出。4.6 慢任務(wù)異步化Celery RabbitMQ問題完整訓(xùn)練計劃生成要 2~2.5 分鐘LLM 生成長 JSON 校驗。同步接口意味著用戶盯著轉(zhuǎn)圈 2 分鐘、HTTP 超時風(fēng)險、uvicorn worker 被占死。方案生成任務(wù)交給 Celery workerAPI 立即返回前端輪詢。POST /ai/plan/generate │ ├─ Redis 查任務(wù)狀態(tài)已 done → 直接返回計劃緩存命中 ├─ 狀態(tài) pending_review → 返回 pending去走確認(rèn)流程不重復(fù)生成 ├─ 狀態(tài) running → 返回 pending去重不重復(fù)派發(fā) └─ 全新請求 → 標(biāo)記 running → Celery 派發(fā) → 返回 {status:pending, task_key} Celery worker獨立進程 └─ LangGraph 生成草稿 → Redis 寫 {status:pending_review, weekly_plan} 后端先不落庫等用戶確認(rèn) 前端每 3 秒 GET /ai/plan/status?task_keyxxx └─ pending 繼續(xù)等 / pending_review 彈「草稿確認(rèn)框」→ 采納/放棄 done 渲染計劃 / failed 提示錯誤 POST /ai/plan/confirm {task_key, action} └─ actionaccept → 落庫停用舊計劃啟用新計劃→ status 變 done actionreject → status 變 rejected丟棄派發(fā)層還有一層降級RabbitMQ 不可用時自動回退到進程內(nèi)asyncio.create_task單機開發(fā)環(huán)境不裝 MQ 也能跑# plan/router.py — 派發(fā)優(yōu)先級def_dispatch_plan(task_key,user_id,plan_request)-bool:try:fromtasks.plan_tasksimportgenerate_plan_task generate_plan_task.delay(task_key,user_id,plan_request)# CeleryreturnTrueexceptException:returnstart_background_generation(...)# 進程內(nèi) asyncio 兜底Celery 任務(wù)里的一個深坑任務(wù)是同步函數(shù)內(nèi)部跑的是 asyncio 代碼LangGraph 生成所以每個任務(wù)用asyncio.run()自建事件循環(huán)。落庫動作不在任務(wù)里做——因為 Human-in-the-loop 需要先生成草稿、等用戶確認(rèn)再落庫所以任務(wù)只負責(zé)生成并寫 Redis真實庫寫發(fā)生在主進程的 confirm 接口里async 代碼直接復(fù)用主進程 engine無 loop 沖突。搭配-P solo單并發(fā)池 任務(wù)內(nèi)獨立 loop 才穩(wěn)。Worker 啟動方式RabbitMQ 4.x 有兼容配置見踩坑實錄python-mcelery-Atasks.celery_app worker-Psolo--loglevelinfo4.7 多級緩存讓重復(fù)請求秒回三類緩存各司其職緩存KeyTTL失效邏輯訓(xùn)練建議用戶畫像 近 30 天統(tǒng)計 的 SHA16 小時畫像/訓(xùn)練數(shù)據(jù)一變key 自然變化自動失效訓(xùn)練計劃生成請求參數(shù)的 SHA17 天換目標(biāo)/周期即新 key任務(wù)狀態(tài)ai:plan:task:{key}1 小時輪詢專用含 done/running/failed建議緩存的 key 設(shè)計值得一提不是簡單的user:{id}而是把影響建議內(nèi)容的所有因子目標(biāo)、水平、身高體重、BMI、30 天訓(xùn)練量、時長、卡路里hash 進 key。好處是天然自失效——用戶今天練了一節(jié)課、明早稱了體重明天的請求 key 就變了自動重新生成而數(shù)據(jù)沒變化時相同請求直接命中秒回。計劃接口的三段式# 1. 已有完成結(jié)果緩存/任務(wù) done直接返回taskawait_task_get(task_key)iftaskandtask.get(status)done:returncreated(PlanResponse(...))# 2. 已在生成中去重不重復(fù)創(chuàng)建任務(wù)iftaskandtask.get(status)running:returncreated({status:pending,task_key:task_key})# 3. 先標(biāo)記 running再派發(fā)后臺生成# 先標(biāo)記后派發(fā)的順序很關(guān)鍵避免任務(wù)瞬間完成被 running 覆蓋await_task_set_running(task_key)_dispatch_plan(...)注意第 3 步的注釋——先標(biāo)記 running 再派發(fā)因為極端情況下 Celery 任務(wù)可能秒級完成緩存命中 LLM 響應(yīng)后寫 running 會把 done 結(jié)果覆蓋回 running前端永遠輪詢不到結(jié)果。4.8 降級鏈AI 永遠不成為單點每個 AI 功能都是三級火箭逐級點火聊天 LangGraph Agent → Dify 兜底 → 本地占位回復(fù) Agent記憶 Postgres checkpoint → MySQL 歷史全量重放 建議 LangGraph Workflow → Dify → BMI 規(guī)則引擎5 級分類 計劃 LangGraph Workflow → 本地模板計劃4 種目標(biāo) 計劃派發(fā)Celery(RabbitMQ) → 進程內(nèi) asyncio 知識檢索Milvus ES 雙路 → 單路降級 → 知識庫暫不可用實測中最底層從不缺席即使 LLM API、Dify、Milvus、RabbitMQ全部同時掛掉用戶依然能拿到一份 BMI 分級的文字建議和一份模板訓(xùn)練計劃。AI 是錦上添花不是核心鏈路——這條原則貫穿了整個設(shè)計。那些有意思的工程細節(jié)1. HTTP-Only Cookie 會話認(rèn)證用 Redis 替代 JWT傳統(tǒng) JWT 的痛點Token 存localStorageXSS 腳本可竊取泄露后在過期前無法撤銷。FitMall 的方案登錄 → 生成 Snowflake Session Token → 用戶信息存 Rediskey: session:{token}, TTL: 24h → Token 寫入 HTTP-Only Cookie前端 JS 不可讀 → 后端解析 Cookie → Redis 查找 → 注入當(dāng)前用戶防 XSSHTTP-Only Cookie 對 JavaScript 完全不可見即時失效刪除 Redis Key 即可讓任意 Token 立即失效用戶/管理端隔離兩套獨立 Cookie 名和 Redis 前綴杜絕越權(quán)自動續(xù)期每次活躍請求刷新 TTL同時保留 JWT 作為 WebSocket 的鑒權(quán)通道瀏覽器 WebSocket API 無法自定義 Cookie Header雙通道并存。2. 手搓 Snowflake 分布式 ID 生成器┌─┬──────────────────────────┬──────────────┬──────────────┐ │0│ 41-bit 時間戳 (ms) │ 10-bit 機器碼 │ 12-bit 序列號 │ └─┴──────────────────────────┴──────────────┴──────────────┘自定義紀(jì)元2024-01-01可用到 2094 年、時鐘回撥容忍 5ms、threading.Lock保證線程安全、整數(shù)/Hex 雙輸出庫表主鍵 / URL 各取所需。3. 雙通道 ES 同步直寫 消息隊列兜底┌─ 通道 1: 直接寫 ES用戶立即搜到 CRUD 操作 ──────────┤ └─ 通道 2: RabbitMQ 消息 → 異步消費重試 ↑ ES 臨時不可用時兜底重放直寫保證管理員剛上架就能搜到消息隊列保證 ES 恢復(fù)后數(shù)據(jù)最終一致。消費端指數(shù)退避重試5s→10s→20s→40s→60s最多 5 次。4. IK 分詞器的雙分析器策略{ik_index_analyzer:{tokenizer:ik_max_word},// 索引細粒度高召回ik_search_analyzer:{tokenizer:ik_smart}// 搜索粗粒度高精度}“增肌蛋白粉” 索引時切成盡可能多的詞條任意組合可命中搜索時只做粗粒度切分避免蘋果手機匹配到水果。這套策略同樣服務(wù)于知識庫 BM25 檢索。5. 訂單超時取消與 Redis 分布式鎖下單后 30 分鐘未支付自動取消并回滾庫存。RabbitMQ 消息 消費端sleep剩余時間 RedisSET NX EX分布式鎖防多實例并發(fā)處理。簡單直接適合單實例場景大規(guī)模需換死信隊列。6. WebSocket 實時聊天與在線狀態(tài)內(nèi)存字典存本進程連接 Redis 集合存跨進程在線用戶雙端同步注冊/清理WebSocket 失敗自動降級 HTTP POST 發(fā)消息未讀計數(shù)實時推送。7. 對象存儲從阿里云 OSS 遷到自建 MinIO早期用的阿里云 OSSSDKoss2后來主動遷到了自建 MinIO理由有三成本OSS 按存儲 流量計費演示項目不值當(dāng)MinIO 部署在自家服務(wù)器零邊際成本。自家的井MySQL/Redis/ES/RabbitMQ/Postgres 全在自建服務(wù)器上再掛個云 OSS 屬于自家有井還去買水還多一份外部依賴要維護。遷移成本幾乎為零MinIO 完全兼容 S3 協(xié)議SDK 換成boto3S3 事實標(biāo)準(zhǔn)對外接口簽名upload_file/delete_file保持不變?nèi)齻€調(diào)用點零改動。兩個協(xié)議差異點值得記path-style 尋址MinIO 不支持 AWS 的 virtual-host 風(fēng)格URL 是endpoint/bucket/key和公開讀策略MinIO 用桶策略 JSON 授權(quán)匿名GetObject替代 OSS 的公共讀桶。業(yè)務(wù)層保留指數(shù)退避重試和預(yù)簽名 URL 能力語義完全對齊。踩坑實錄重構(gòu)過程真機聯(lián)調(diào)修掉的坑每一個都值得記錄1. RabbitMQ 4.3 拒絕 Celery 的 transient 隊列Worker 一啟動就拋541 INTERNAL_ERROR。根因RabbitMQ 4.3 起默認(rèn)禁用transient 非排他隊列而 Celery 的 control/mingle/gossip 廣播隊列正是這種類型。解法純代碼側(cè)不動服務(wù)器# celery_app.pyworker_disable_mingleTrue,worker_disable_gossipTrue,worker_enable_remote_controlFalse,control_queue_durableTrue,event_queue_durableTrue,單 worker 場景這些廣播本來就沒用關(guān)掉反而干凈。2. SQLAlchemy engine 與 event loop 的綁定Celery 任務(wù)里復(fù)用主進程的數(shù)據(jù)庫 engine 會炸——連接池綁定已關(guān)閉的 loop。每個任務(wù)必須create_async_engine自建、用完dispose()。3. Embedding 服務(wù)的dimensions參數(shù)SiliconFlow 的 bge-m3 不接受dimensions參數(shù)加了直接 400。做成配置開關(guān)EMBEDDING_SEND_DIMENSIONS默認(rèn)不發(fā)送。4. LangGraph 狀態(tài)累積用普通TypedDict定義狀態(tài)消息無法跨節(jié)點累積每個節(jié)點返回值直接覆蓋。改繼承MessagesState其內(nèi)置的add_messagesreducer 自動做追加合并。5.tool裝飾器漏加工具函數(shù)寫了但沒掛toolLangChain 不識別Agent 永遠不調(diào)用它。工具存在和可被 Agent 發(fā)現(xiàn)是兩回事。6. 工具調(diào)用流的正文回顯stream_modemessages下工具調(diào)用輪次也會產(chǎn)出 chunk內(nèi)容是 tool_call 參數(shù) JSON。不過濾的話用戶會看到聊天框里突然刷出一坨原始 JSON。按chunk.tool_calls跳過即可。7. 先標(biāo)記 running 再派發(fā)順序反了會出現(xiàn)任務(wù)毫秒級完成寫 done → 主協(xié)程再寫 running 把它覆蓋 → 前端永遠輪詢不到結(jié)果。并發(fā)寫同一個 key 時想清楚時序。8. 依賴版本沖突langchain-openai 1.6要求langchain-core1.6langgraph-checkpoint-postgres的正確版本區(qū)間是2.0,3.0。AI 生態(tài)迭代極快鎖版本區(qū)間前先看清依賴樹。9. 新版 uvicorn 在 Windows 硬編碼 ProactorEventLoop接入 Postgres checkpoint 后本地啟動一直降級——psycopg 的 async 模式只支持 SelectorEventLoop而這版 uvicorn 的asyncio_loop_factory在 Windows 上直接return asyncio.ProactorEventLoop且 uvicorn 先建事件循環(huán)、后導(dǎo)入 app 模塊所以在 main.py 里設(shè) event loop policy 永遠太晚。解法是啟動命令顯式傳自定義 looppython-muvicorn app.main:app--loopasyncio:SelectorEventLoop--loop支持任意module:attribute導(dǎo)入串SelectorEventLoop跨平臺存在Linux 本來就是默認(rèn)這條命令兩邊通用。典型教訓(xùn)event loop policy 只影響未來的循環(huán)框架自己建循環(huán)時 policy 就成了擺設(shè)。10. psycopg 連接池必須 autocommitAsyncPostgresSaver.setup()建表時會執(zhí)行CREATE INDEX CONCURRENTLY這條 SQL不能跑在事務(wù)里而 psycopg 默認(rèn)開事務(wù)。自建連接池時必須傳kwargs{autocommit: True, prepare_threshold: 0, row_factory: dict_row}——這三個參數(shù)抄官方from_conn_string的實現(xiàn)就對了自作聰明省掉任何一個都會在奇怪的地方報錯?？偨Y(jié)與思考這輪重構(gòu)做對了什么形態(tài)分離Agent 管不可預(yù)測的對話Workflow 管確定性的生成。用對工具比用好工具更重要?；糜X從架構(gòu)層封死商品卡片數(shù)據(jù)從工具直通前端LLM 沒有輸出卡片的通道——不靠提示詞祈禱靠數(shù)據(jù)通路設(shè)計。慢任務(wù)異步化2.5 分鐘的生成從接口卡死變成立即反饋 進度輪詢且?guī)Ь彺婧腿ブ?。狀態(tài)真正持久化Agent 會話狀態(tài)落 Postgres checkpoint服務(wù)重啟記憶不丟——多輪記憶從每次全量重放變成只傳增量、斷點接力。每一層都降級AI 服務(wù)掛了系統(tǒng)照常運轉(zhuǎn)只是變笨不變癱?？梢岳^續(xù)演進的方向意圖分流前置先用小模型分類閑聊/咨詢/導(dǎo)購閑聊直達 LLM 省去工具綁定開銷計劃生成提速分段生成 并發(fā)拼裝每周一個子任務(wù)2.5 分鐘壓到 30 秒級訂單超時的sleep方案換 RabbitMQ 死信隊列支撐多實例部署更強 Human-in-the-loop當(dāng)前是生成后確認(rèn)草稿結(jié)合已有 checkpoint 可演進為生成前 interrupt 等用戶確認(rèn)參數(shù)——圖跑到一半停下來用戶點確認(rèn)后從斷點繼續(xù)項目數(shù)據(jù)后端12 個業(yè)務(wù)模塊 AI 三層架構(gòu)agents/workflows/rag前端~40 個 Vue 組件3 個 Pinia Store基礎(chǔ)設(shè)施Docker Compose 編排MySQL / Redis / ES / RabbitMQ / Postgres / MinIO / Milvus云端AI 鏈路1 個 ReAct Agent3 工具、2 個固定 Workflow、雙路混合檢索、Celery 異步任務(wù)、三級緩存、Postgres checkpoint 持久化從調(diào)一個 API到設(shè)計一套 AI 架構(gòu)這個項目最大的收獲是AI 工程化的核心不是模型是數(shù)據(jù)通路、失敗路徑和響應(yīng)時間的設(shè)計。模型會越來越好但這些工程問題永遠存在。