博客（AI重做）)
FitMall 運(yùn)動優(yōu)選一個健身電商平臺的三階段 AI 架構(gòu)演進(jìn)全記錄從規(guī)則引擎到 Dify 低代碼編排再到 LangGraph 原生智能體——本文完整拆解一個全棧健身電商項(xiàng)目的技術(shù)選型、架構(gòu)設(shè)計與 AI 工程化的踩坑實(shí)錄。不只是接了個 AI而是一步步演進(jìn)出企業(yè)級的 Agent 架構(gòu)。目錄項(xiàng)目概覽技術(shù)棧一覽架構(gòu)設(shè)計AI 架構(gòu)演進(jìn)從 Dify 到 LangGraph為什么要重構(gòu)Agent 與 Workflow兩種形態(tài)的選型哲學(xué)混合檢索底座Milvus ES 雙路召回聊天 AgentReAct 循環(huán) 可跳轉(zhuǎn)卡片訓(xùn)練建議與計劃固定 Workflow 強(qiáng)校驗(yàn) JSON慢任務(wù)異步化Celery RabbitMQ多級緩存讓重復(fù)請求秒回降級鏈AI 永遠(yuǎn)不成為單點(diǎn)那些有意思的工程細(xì)節(jié)踩坑實(shí)錄總結(jié)與思考項(xiàng)目概覽FitMall 是一個面向運(yùn)動健身人群的全棧 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 能力是本項(xiàng)目最有故事的部分它經(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會話、驗(yàn)證碼、分布式鎖、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自簽名實(shí)現(xiàn)不依賴官方 SDK存儲MinIOS3 協(xié)議 boto3自建對象存儲匿名只讀桶 預(yù)簽名 URL短信阿里云 DypnsapiRedis 存驗(yàn)證碼5 分鐘過期ID 生成自研 Snowflake41 位毫秒時間戳 10 位機(jī)器 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實(shí)時聊天 在線狀態(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 校驗(yàn)回環(huán) │ └── schemas.py # 輸出結(jié)構(gòu)Pydantic 強(qiáng)校驗(yàn) ├── 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 實(shí)例brokerRabbitMQbackendRedis └── plan_tasks.py # 訓(xùn)練計劃后臺生成任務(wù)請求生命周期HTTP 請求 → CORS 中間件跨域白名單校驗(yàn) → 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)演進(jìn)從 Dify 到 LangGraph4.1 為什么要重構(gòu)Dify 階段的 AI 助手是這樣的用戶想買蛋白粉 → Dify ChatflowLLM 意圖識別 → {category: goods, keyword: 蛋白粉} → Dify HTTP 節(jié)點(diǎn) → 公網(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)用時機(jī)全在 Dify 控制臺里代碼倉庫里沒有版本管理出問題只能登平臺排查。形態(tài)單一聊天、建議、計劃三種完全不同性質(zhì)的任務(wù)被塞進(jìn)同一種工作流形態(tài)里計劃生成這種 2 分鐘級慢任務(wù)在同步請求里直接把接口拖死。重構(gòu)后的目標(biāo)很明確Agent 管聊天、Workflow 管生成、檢索自建、慢任務(wù)異步化、每一層都有降級。4.2 Agent 與 Workflow兩種形態(tài)的選型哲學(xué)LangGraph 提供了兩種建圖方式本項(xiàng)目兩種都用了各管一攤聊天助手Agent訓(xùn)練建議/計劃Workflow形態(tài)agent ? tools回環(huán)LLM 自主決策start → 步驟 → END固定直線類比一個有工具箱的員工自己判斷拿哪把工具一條流水線原料進(jìn)去、成品出來適用輸入不可預(yù)測需要靈活路由輸入輸出固定追求穩(wěn)定可控失敗代價高工具調(diào)錯頂多答非所問低流程不會跑偏這個判斷是整個重構(gòu)的地基不要用 Agent 去做 Workflow 的事。用戶點(diǎn)生成訓(xùn)練計劃時不需要 AI 思考要不要先查一下用戶資料——這些步驟是確定的寫死就行確定性場景里自主性只會引入不穩(wěn)定。4.3 混合檢索底座Milvus ES 雙路召回健身知識庫深蹲要領(lǐng)、減脂原則、增肌營養(yǎng)等語料的檢索單一手段都有短板純向量檢索Milvus懂語義“怎么練腿” 能命中 “深蹲”“臀橋”但對精確術(shù)語如專有名詞不夠敏感純關(guān)鍵詞ES BM25精確匹配強(qiáng)但 “怎么練腿” 和 “下肢訓(xùn)練” 永遠(yuǎ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)鍵實(shí)現(xiàn)細(xì)節(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 或達(dá)到步數(shù)上限 → END三個工具覆蓋三類典型問題工具觸發(fā)場景數(shù)據(jù)來源search_goods_or_courses“想買蛋白粉”“有減脂課嗎”ES復(fù)用商城搜索服務(wù)search_fitness_knowledge“深蹲膝蓋疼怎么回事”混合檢索Milvus ESget_user_training_profile“按我的情況該練什么”MySQL畫像 近 30 天統(tǒng)計卡片推薦是體驗(yàn)上的關(guān)鍵設(shè)計工具返回的每個商品/課程不只進(jìn) LLM 上下文還同時收進(jìn)一個cards_holder容器。Agent 跑完后SSE 流里額外推一個cards事件前端渲染成可點(diǎn)擊跳轉(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í)商品/課程絕不編造名稱、價格或鏈接”?？ㄆ瑪?shù)據(jù)從工具結(jié)果直通前端LLM 只負(fù)責(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表的全部歷史拼進(jìn)初始messages功能不丟只是每次全量重放。刪除會話時同步清理對應(yīng) checkpoint thread不留孤兒數(shù)據(jù)。4.5 訓(xùn)練建議與計劃固定 Workflow 強(qiáng)校驗(yàn) JSON訓(xùn)練計劃生成是典型的確定場景輸入目標(biāo)/周期/水平/可用日/偏好和輸出周×天嵌套的結(jié)構(gòu)化計劃都是固定的。它的 Workflow 長這樣START → generate結(jié)構(gòu)化輸出 ──校驗(yàn)通過──→ END ↑ │ │ 校驗(yàn)失敗且重試 2 └──── repair帶錯誤信息重生成←┘核心技巧是with_structured_output——不是讓 LLM “輸出 JSON”那要靠祈禱而是用 Pydantic Schema 約束輸出結(jié)構(gòu)框架層保證解析# workflows/plan.py — generate 節(jié)點(diǎn)asyncdefgenerate(state:_PlanState)-dict:boundmodel.with_structured_output(PlanOutput)# 強(qiáng)校驗(yàn)resultawaitbound.ainvoke([(system,_PLAN_SYSTEM_PROMPT),(user,f請基于以下要求生成訓(xùn)練計劃\n{request}),])return{output:payload,error:}# 成功# 失敗時 return {error: str(e), repairs: 1} # 進(jìn) repair 環(huán)repair 環(huán)是亮點(diǎn)校驗(yàn)失敗不是直接拋錯而是把錯誤信息拼進(jìn)提示詞讓模型自己修最多修 2 次。類比成代碼編譯報錯后讓 AI 自己改 bug——大部分格式問題一輪就修好用戶體驗(yàn)從失敗重試變成稍慢一點(diǎn)但成功。修復(fù)仍失敗則返回None由上層降級到本地模板計劃減脂/增肌/塑形/保持 四套模板保證接口永遠(yuǎn)有可用輸出。4.6 慢任務(wù)異步化Celery RabbitMQ問題完整訓(xùn)練計劃生成要 2~2.5 分鐘LLM 生成長 JSON 校驗(yàn)。同步接口意味著用戶盯著轉(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獨(dú)立進(jìn)程 └─ 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 不可用時自動回退到進(jìn)程內(nèi)asyncio.create_task單機(jī)開發(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(...)# 進(jìn)程內(nèi) asyncio 兜底Celery 任務(wù)里的一個深坑任務(wù)是同步函數(shù)內(nèi)部跑的是 asyncio 代碼LangGraph 生成所以每個任務(wù)用asyncio.run()自建事件循環(huán)。落庫動作不在任務(wù)里做——因?yàn)?Human-in-the-loop 需要先生成草稿、等用戶確認(rèn)再落庫所以任務(wù)只負(fù)責(zé)生成并寫 Redis真實(shí)庫寫發(fā)生在主進(jìn)程的 confirm 接口里async 代碼直接復(fù)用主進(jìn)程 engine無 loop 沖突。搭配-P solo單并發(fā)池 任務(wù)內(nèi)獨(dú)立 loop 才穩(wěn)。Worker 啟動方式RabbitMQ 4.x 有兼容配置見踩坑實(shí)錄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 進(jìn) 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ā)因?yàn)闃O端情況下 Celery 任務(wù)可能秒級完成緩存命中 LLM 響應(yīng)后寫 running 會把 done 結(jié)果覆蓋回 running前端永遠(yuǎn)輪詢不到結(jié)果。4.8 降級鏈AI 永遠(yuǎn)不成為單點(diǎn)每個 AI 功能都是三級火箭逐級點(diǎn)火聊天 LangGraph Agent → Dify 兜底 → 本地占位回復(fù) Agent記憶 Postgres checkpoint → MySQL 歷史全量重放 建議 LangGraph Workflow → Dify → BMI 規(guī)則引擎5 級分類 計劃 LangGraph Workflow → 本地模板計劃4 種目標(biāo) 計劃派發(fā)Celery(RabbitMQ) → 進(jìn)程內(nèi) asyncio 知識檢索Milvus ES 雙路 → 單路降級 → 知識庫暫不可用實(shí)測中最底層從不缺席即使 LLM API、Dify、Milvus、RabbitMQ全部同時掛掉用戶依然能拿到一份 BMI 分級的文字建議和一份模板訓(xùn)練計劃。AI 是錦上添花不是核心鏈路——這條原則貫穿了整個設(shè)計。那些有意思的工程細(xì)節(jié)1. HTTP-Only Cookie 會話認(rèn)證用 Redis 替代 JWT傳統(tǒng) JWT 的痛點(diǎn)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 立即失效用戶/管理端隔離兩套獨(dú)立 Cookie 名和 Redis 前綴杜絕越權(quán)自動續(xù)期每次活躍請求刷新 TTL同時保留 JWT 作為 WebSocket 的鑒權(quán)通道瀏覽器 WebSocket API 無法自定義 Cookie Header雙通道并存。2. 手搓 Snowflake 分布式 ID 生成器┌─┬──────────────────────────┬──────────────┬──────────────┐ │0│ 41-bit 時間戳 (ms) │ 10-bit 機(jī)器碼 │ 12-bit 序列號 │ └─┴──────────────────────────┴──────────────┴──────────────┘自定義紀(jì)元2024-01-01可用到 2094 年、時鐘回?fù)苋萑?5ms、threading.Lock保證線程安全、整數(shù)/Hex 雙輸出庫表主鍵 / URL 各取所需。3. 雙通道 ES 同步直寫 消息隊列兜底┌─ 通道 1: 直接寫 ES用戶立即搜到 CRUD 操作 ──────────┤ └─ 通道 2: RabbitMQ 消息 → 異步消費(fèi)重試 ↑ ES 臨時不可用時兜底重放直寫保證管理員剛上架就能搜到消息隊列保證 ES 恢復(fù)后數(shù)據(jù)最終一致。消費(fèi)端指數(shù)退避重試5s→10s→20s→40s→60s最多 5 次。4. IK 分詞器的雙分析器策略{ik_index_analyzer:{tokenizer:ik_max_word},// 索引細(xì)粒度高召回ik_search_analyzer:{tokenizer:ik_smart}// 搜索粗粒度高精度}“增肌蛋白粉” 索引時切成盡可能多的詞條任意組合可命中搜索時只做粗粒度切分避免蘋果手機(jī)匹配到水果。這套策略同樣服務(wù)于知識庫 BM25 檢索。5. 訂單超時取消與 Redis 分布式鎖下單后 30 分鐘未支付自動取消并回滾庫存。RabbitMQ 消息 消費(fèi)端sleep剩余時間 RedisSET NX EX分布式鎖防多實(shí)例并發(fā)處理。簡單直接適合單實(shí)例場景大規(guī)模需換死信隊列。6. WebSocket 實(shí)時聊天與在線狀態(tài)內(nèi)存字典存本進(jìn)程連接 Redis 集合存跨進(jìn)程在線用戶雙端同步注冊/清理WebSocket 失敗自動降級 HTTP POST 發(fā)消息未讀計數(shù)實(shí)時推送。7. 對象存儲從阿里云 OSS 遷到自建 MinIO早期用的阿里云 OSSSDKoss2后來主動遷到了自建 MinIO理由有三成本OSS 按存儲 流量計費(fèi)演示項(xiàng)目不值當(dāng)MinIO 部署在自家服務(wù)器零邊際成本。自家的井MySQL/Redis/ES/RabbitMQ/Postgres 全在自建服務(wù)器上再掛個云 OSS 屬于自家有井還去買水還多一份外部依賴要維護(hù)。遷移成本幾乎為零MinIO 完全兼容 S3 協(xié)議SDK 換成boto3S3 事實(shí)標(biāo)準(zhǔn)對外接口簽名upload_file/delete_file保持不變?nèi)齻€調(diào)用點(diǎn)零改動。兩個協(xié)議差異點(diǎn)值得記path-style 尋址MinIO 不支持 AWS 的 virtual-host 風(fēng)格URL 是endpoint/bucket/key和公開讀策略MinIO 用桶策略 JSON 授權(quán)匿名GetObject替代 OSS 的公共讀桶。業(yè)務(wù)層保留指數(shù)退避重試和預(yù)簽名 URL 能力語義完全對齊。踩坑實(shí)錄重構(gòu)過程真機(jī)聯(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ù)用主進(jìn)程的數(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é)點(diǎn)累積每個節(jié)點(diǎn)返回值直接覆蓋。改繼承MessagesState其內(nèi)置的add_messagesreducer 自動做追加合并。5.tool裝飾器漏加工具函數(shù)寫了但沒掛toolLangChain 不識別Agent 永遠(yuǎn)不調(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 把它覆蓋 → 前端永遠(yuǎn)輪詢不到結(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 永遠(yuǎn)太晚。解法是啟動命令顯式傳自定義 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的實(shí)現(xiàn)就對了自作聰明省掉任何一個都會在奇怪的地方報錯?？偨Y(jié)與思考這輪重構(gòu)做對了什么形態(tài)分離Agent 管不可預(yù)測的對話Workflow 管確定性的生成。用對工具比用好工具更重要。幻覺從架構(gòu)層封死商品卡片數(shù)據(jù)從工具直通前端LLM 沒有輸出卡片的通道——不靠提示詞祈禱靠數(shù)據(jù)通路設(shè)計。慢任務(wù)異步化2.5 分鐘的生成從接口卡死變成立即反饋 進(jìn)度輪詢且?guī)Ь彺婧腿ブ?。狀態(tài)真正持久化Agent 會話狀態(tài)落 Postgres checkpoint服務(wù)重啟記憶不丟——多輪記憶從每次全量重放變成只傳增量、斷點(diǎn)接力。每一層都降級AI 服務(wù)掛了系統(tǒng)照常運(yùn)轉(zhuǎn)只是變笨不變癱?？梢岳^續(xù)演進(jìn)的方向意圖分流前置先用小模型分類閑聊/咨詢/導(dǎo)購閑聊直達(dá) LLM 省去工具綁定開銷計劃生成提速分段生成 并發(fā)拼裝每周一個子任務(wù)2.5 分鐘壓到 30 秒級訂單超時的sleep方案換 RabbitMQ 死信隊列支撐多實(shí)例部署更強(qiáng) Human-in-the-loop當(dāng)前是生成后確認(rèn)草稿結(jié)合已有 checkpoint 可演進(jìn)為生成前 interrupt 等用戶確認(rèn)參數(shù)——圖跑到一半停下來用戶點(diǎn)確認(rèn)后從斷點(diǎn)繼續(xù)項(xiàng)目數(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)這個項(xiàng)目最大的收獲是AI 工程化的核心不是模型是數(shù)據(jù)通路、失敗路徑和響應(yīng)時間的設(shè)計。模型會越來越好但這些工程問題永遠(yuǎn)存在。 FitMall技術(shù)博客（AI重做）
FitMall 運(yùn)動優(yōu)選一個健身電商平臺的三階段 AI 架構(gòu)演進(jìn)全記錄從規(guī)則引擎到 Dify 低代碼編排再到 LangGraph 原生智能體——本文完整拆解一個全棧健身電商項(xiàng)目的技術(shù)選型、架構(gòu)設(shè)計與 AI 工程化的踩坑實(shí)錄。不只是接了個 AI而是一步步演進(jìn)出企業(yè)級的 Agent 架構(gòu)。目錄項(xiàng)目概覽技術(shù)棧一覽架構(gòu)設(shè)計AI 架構(gòu)演進(jìn)從 Dify 到 LangGraph為什么要重構(gòu)Agent 與 Workflow兩種形態(tài)的選型哲學(xué)混合檢索底座Milvus ES 雙路召回聊天 AgentReAct 循環(huán) 可跳轉(zhuǎn)卡片訓(xùn)練建議與計劃固定 Workflow 強(qiáng)校驗(yàn) JSON慢任務(wù)異步化Celery RabbitMQ多級緩存讓重復(fù)請求秒回降級鏈AI 永遠(yuǎn)不成為單點(diǎn)那些有意思的工程細(xì)節(jié)踩坑實(shí)錄總結(jié)與思考項(xiàng)目概覽FitMall 是一個面向運(yùn)動健身人群的全棧 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 能力是本項(xiàng)目最有故事的部分它經(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會話、驗(yàn)證碼、分布式鎖、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自簽名實(shí)現(xiàn)不依賴官方 SDK存儲MinIOS3 協(xié)議 boto3自建對象存儲匿名只讀桶 預(yù)簽名 URL短信阿里云 DypnsapiRedis 存驗(yàn)證碼5 分鐘過期ID 生成自研 Snowflake41 位毫秒時間戳 10 位機(jī)器 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實(shí)時聊天 在線狀態(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 校驗(yàn)回環(huán) │ └── schemas.py # 輸出結(jié)構(gòu)Pydantic 強(qiáng)校驗(yàn) ├── 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 實(shí)例brokerRabbitMQbackendRedis └── plan_tasks.py # 訓(xùn)練計劃后臺生成任務(wù)請求生命周期HTTP 請求 → CORS 中間件跨域白名單校驗(yàn) → 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)演進(jìn)從 Dify 到 LangGraph4.1 為什么要重構(gòu)Dify 階段的 AI 助手是這樣的用戶想買蛋白粉 → Dify ChatflowLLM 意圖識別 → {category: goods, keyword: 蛋白粉} → Dify HTTP 節(jié)點(diǎn) → 公網(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)用時機(jī)全在 Dify 控制臺里代碼倉庫里沒有版本管理出問題只能登平臺排查。形態(tài)單一聊天、建議、計劃三種完全不同性質(zhì)的任務(wù)被塞進(jìn)同一種工作流形態(tài)里計劃生成這種 2 分鐘級慢任務(wù)在同步請求里直接把接口拖死。重構(gòu)后的目標(biāo)很明確Agent 管聊天、Workflow 管生成、檢索自建、慢任務(wù)異步化、每一層都有降級。4.2 Agent 與 Workflow兩種形態(tài)的選型哲學(xué)LangGraph 提供了兩種建圖方式本項(xiàng)目兩種都用了各管一攤聊天助手Agent訓(xùn)練建議/計劃Workflow形態(tài)agent ? tools回環(huán)LLM 自主決策start → 步驟 → END固定直線類比一個有工具箱的員工自己判斷拿哪把工具一條流水線原料進(jìn)去、成品出來適用輸入不可預(yù)測需要靈活路由輸入輸出固定追求穩(wěn)定可控失敗代價高工具調(diào)錯頂多答非所問低流程不會跑偏這個判斷是整個重構(gòu)的地基不要用 Agent 去做 Workflow 的事。用戶點(diǎn)生成訓(xùn)練計劃時不需要 AI 思考要不要先查一下用戶資料——這些步驟是確定的寫死就行確定性場景里自主性只會引入不穩(wěn)定。4.3 混合檢索底座Milvus ES 雙路召回健身知識庫深蹲要領(lǐng)、減脂原則、增肌營養(yǎng)等語料的檢索單一手段都有短板純向量檢索Milvus懂語義“怎么練腿” 能命中 “深蹲”“臀橋”但對精確術(shù)語如專有名詞不夠敏感純關(guān)鍵詞ES BM25精確匹配強(qiáng)但 “怎么練腿” 和 “下肢訓(xùn)練” 永遠(yuǎ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)鍵實(shí)現(xiàn)細(xì)節(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 或達(dá)到步數(shù)上限 → END三個工具覆蓋三類典型問題工具觸發(fā)場景數(shù)據(jù)來源search_goods_or_courses“想買蛋白粉”“有減脂課嗎”ES復(fù)用商城搜索服務(wù)search_fitness_knowledge“深蹲膝蓋疼怎么回事”混合檢索Milvus ESget_user_training_profile“按我的情況該練什么”MySQL畫像 近 30 天統(tǒng)計卡片推薦是體驗(yàn)上的關(guān)鍵設(shè)計工具返回的每個商品/課程不只進(jìn) LLM 上下文還同時收進(jìn)一個cards_holder容器。Agent 跑完后SSE 流里額外推一個cards事件前端渲染成可點(diǎn)擊跳轉(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í)商品/課程絕不編造名稱、價格或鏈接”?？ㄆ瑪?shù)據(jù)從工具結(jié)果直通前端LLM 只負(fù)責(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表的全部歷史拼進(jìn)初始messages功能不丟只是每次全量重放。刪除會話時同步清理對應(yīng) checkpoint thread不留孤兒數(shù)據(jù)。4.5 訓(xùn)練建議與計劃固定 Workflow 強(qiáng)校驗(yàn) JSON訓(xùn)練計劃生成是典型的確定場景輸入目標(biāo)/周期/水平/可用日/偏好和輸出周×天嵌套的結(jié)構(gòu)化計劃都是固定的。它的 Workflow 長這樣START → generate結(jié)構(gòu)化輸出 ──校驗(yàn)通過──→ END ↑ │ │ 校驗(yàn)失敗且重試 2 └──── repair帶錯誤信息重生成←┘核心技巧是with_structured_output——不是讓 LLM “輸出 JSON”那要靠祈禱而是用 Pydantic Schema 約束輸出結(jié)構(gòu)框架層保證解析# workflows/plan.py — generate 節(jié)點(diǎn)asyncdefgenerate(state:_PlanState)-dict:boundmodel.with_structured_output(PlanOutput)# 強(qiáng)校驗(yàn)resultawaitbound.ainvoke([(system,_PLAN_SYSTEM_PROMPT),(user,f請基于以下要求生成訓(xùn)練計劃\n{request}),])return{output:payload,error:}# 成功# 失敗時 return {error: str(e), repairs: 1} # 進(jìn) repair 環(huán)repair 環(huán)是亮點(diǎn)校驗(yàn)失敗不是直接拋錯而是把錯誤信息拼進(jìn)提示詞讓模型自己修最多修 2 次。類比成代碼編譯報錯后讓 AI 自己改 bug——大部分格式問題一輪就修好用戶體驗(yàn)從失敗重試變成稍慢一點(diǎn)但成功。修復(fù)仍失敗則返回None由上層降級到本地模板計劃減脂/增肌/塑形/保持 四套模板保證接口永遠(yuǎn)有可用輸出。4.6 慢任務(wù)異步化Celery RabbitMQ問題完整訓(xùn)練計劃生成要 2~2.5 分鐘LLM 生成長 JSON 校驗(yàn)。同步接口意味著用戶盯著轉(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獨(dú)立進(jìn)程 └─ 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 不可用時自動回退到進(jìn)程內(nèi)asyncio.create_task單機(jī)開發(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(...)# 進(jìn)程內(nèi) asyncio 兜底Celery 任務(wù)里的一個深坑任務(wù)是同步函數(shù)內(nèi)部跑的是 asyncio 代碼LangGraph 生成所以每個任務(wù)用asyncio.run()自建事件循環(huán)。落庫動作不在任務(wù)里做——因?yàn)?Human-in-the-loop 需要先生成草稿、等用戶確認(rèn)再落庫所以任務(wù)只負(fù)責(zé)生成并寫 Redis真實(shí)庫寫發(fā)生在主進(jìn)程的 confirm 接口里async 代碼直接復(fù)用主進(jìn)程 engine無 loop 沖突。搭配-P solo單并發(fā)池 任務(wù)內(nèi)獨(dú)立 loop 才穩(wěn)。Worker 啟動方式RabbitMQ 4.x 有兼容配置見踩坑實(shí)錄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 進(jìn) 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ā)因?yàn)闃O端情況下 Celery 任務(wù)可能秒級完成緩存命中 LLM 響應(yīng)后寫 running 會把 done 結(jié)果覆蓋回 running前端永遠(yuǎn)輪詢不到結(jié)果。4.8 降級鏈AI 永遠(yuǎn)不成為單點(diǎn)每個 AI 功能都是三級火箭逐級點(diǎn)火聊天 LangGraph Agent → Dify 兜底 → 本地占位回復(fù) Agent記憶 Postgres checkpoint → MySQL 歷史全量重放 建議 LangGraph Workflow → Dify → BMI 規(guī)則引擎5 級分類 計劃 LangGraph Workflow → 本地模板計劃4 種目標(biāo) 計劃派發(fā)Celery(RabbitMQ) → 進(jìn)程內(nèi) asyncio 知識檢索Milvus ES 雙路 → 單路降級 → 知識庫暫不可用實(shí)測中最底層從不缺席即使 LLM API、Dify、Milvus、RabbitMQ全部同時掛掉用戶依然能拿到一份 BMI 分級的文字建議和一份模板訓(xùn)練計劃。AI 是錦上添花不是核心鏈路——這條原則貫穿了整個設(shè)計。那些有意思的工程細(xì)節(jié)1. HTTP-Only Cookie 會話認(rèn)證用 Redis 替代 JWT傳統(tǒng) JWT 的痛點(diǎn)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 立即失效用戶/管理端隔離兩套獨(dú)立 Cookie 名和 Redis 前綴杜絕越權(quán)自動續(xù)期每次活躍請求刷新 TTL同時保留 JWT 作為 WebSocket 的鑒權(quán)通道瀏覽器 WebSocket API 無法自定義 Cookie Header雙通道并存。2. 手搓 Snowflake 分布式 ID 生成器┌─┬──────────────────────────┬──────────────┬──────────────┐ │0│ 41-bit 時間戳 (ms) │ 10-bit 機(jī)器碼 │ 12-bit 序列號 │ └─┴──────────────────────────┴──────────────┴──────────────┘自定義紀(jì)元2024-01-01可用到 2094 年、時鐘回?fù)苋萑?5ms、threading.Lock保證線程安全、整數(shù)/Hex 雙輸出庫表主鍵 / URL 各取所需。3. 雙通道 ES 同步直寫 消息隊列兜底┌─ 通道 1: 直接寫 ES用戶立即搜到 CRUD 操作 ──────────┤ └─ 通道 2: RabbitMQ 消息 → 異步消費(fèi)重試 ↑ ES 臨時不可用時兜底重放直寫保證管理員剛上架就能搜到消息隊列保證 ES 恢復(fù)后數(shù)據(jù)最終一致。消費(fèi)端指數(shù)退避重試5s→10s→20s→40s→60s最多 5 次。4. IK 分詞器的雙分析器策略{ik_index_analyzer:{tokenizer:ik_max_word},// 索引細(xì)粒度高召回ik_search_analyzer:{tokenizer:ik_smart}// 搜索粗粒度高精度}“增肌蛋白粉” 索引時切成盡可能多的詞條任意組合可命中搜索時只做粗粒度切分避免蘋果手機(jī)匹配到水果。這套策略同樣服務(wù)于知識庫 BM25 檢索。5. 訂單超時取消與 Redis 分布式鎖下單后 30 分鐘未支付自動取消并回滾庫存。RabbitMQ 消息 消費(fèi)端sleep剩余時間 RedisSET NX EX分布式鎖防多實(shí)例并發(fā)處理。簡單直接適合單實(shí)例場景大規(guī)模需換死信隊列。6. WebSocket 實(shí)時聊天與在線狀態(tài)內(nèi)存字典存本進(jìn)程連接 Redis 集合存跨進(jìn)程在線用戶雙端同步注冊/清理WebSocket 失敗自動降級 HTTP POST 發(fā)消息未讀計數(shù)實(shí)時推送。7. 對象存儲從阿里云 OSS 遷到自建 MinIO早期用的阿里云 OSSSDKoss2后來主動遷到了自建 MinIO理由有三成本OSS 按存儲 流量計費(fèi)演示項(xiàng)目不值當(dāng)MinIO 部署在自家服務(wù)器零邊際成本。自家的井MySQL/Redis/ES/RabbitMQ/Postgres 全在自建服務(wù)器上再掛個云 OSS 屬于自家有井還去買水還多一份外部依賴要維護(hù)。遷移成本幾乎為零MinIO 完全兼容 S3 協(xié)議SDK 換成boto3S3 事實(shí)標(biāo)準(zhǔn)對外接口簽名upload_file/delete_file保持不變?nèi)齻€調(diào)用點(diǎn)零改動。兩個協(xié)議差異點(diǎn)值得記path-style 尋址MinIO 不支持 AWS 的 virtual-host 風(fēng)格URL 是endpoint/bucket/key和公開讀策略MinIO 用桶策略 JSON 授權(quán)匿名GetObject替代 OSS 的公共讀桶。業(yè)務(wù)層保留指數(shù)退避重試和預(yù)簽名 URL 能力語義完全對齊。踩坑實(shí)錄重構(gòu)過程真機(jī)聯(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ù)用主進(jìn)程的數(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é)點(diǎn)累積每個節(jié)點(diǎn)返回值直接覆蓋。改繼承MessagesState其內(nèi)置的add_messagesreducer 自動做追加合并。5.tool裝飾器漏加工具函數(shù)寫了但沒掛toolLangChain 不識別Agent 永遠(yuǎn)不調(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 把它覆蓋 → 前端永遠(yuǎn)輪詢不到結(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 永遠(yuǎn)太晚。解法是啟動命令顯式傳自定義 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的實(shí)現(xiàn)就對了自作聰明省掉任何一個都會在奇怪的地方報錯?？偨Y(jié)與思考這輪重構(gòu)做對了什么形態(tài)分離Agent 管不可預(yù)測的對話Workflow 管確定性的生成。用對工具比用好工具更重要。幻覺從架構(gòu)層封死商品卡片數(shù)據(jù)從工具直通前端LLM 沒有輸出卡片的通道——不靠提示詞祈禱靠數(shù)據(jù)通路設(shè)計。慢任務(wù)異步化2.5 分鐘的生成從接口卡死變成立即反饋 進(jìn)度輪詢且?guī)Ь彺婧腿ブ?。狀態(tài)真正持久化Agent 會話狀態(tài)落 Postgres checkpoint服務(wù)重啟記憶不丟——多輪記憶從每次全量重放變成只傳增量、斷點(diǎn)接力。每一層都降級AI 服務(wù)掛了系統(tǒng)照常運(yùn)轉(zhuǎn)只是變笨不變癱?？梢岳^續(xù)演進(jìn)的方向意圖分流前置先用小模型分類閑聊/咨詢/導(dǎo)購閑聊直達(dá) LLM 省去工具綁定開銷計劃生成提速分段生成 并發(fā)拼裝每周一個子任務(wù)2.5 分鐘壓到 30 秒級訂單超時的sleep方案換 RabbitMQ 死信隊列支撐多實(shí)例部署更強(qiáng) Human-in-the-loop當(dāng)前是生成后確認(rèn)草稿結(jié)合已有 checkpoint 可演進(jìn)為生成前 interrupt 等用戶確認(rèn)參數(shù)——圖跑到一半停下來用戶點(diǎn)確認(rèn)后從斷點(diǎn)繼續(xù)項(xiàng)目數(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)這個項(xiàng)目最大的收獲是AI 工程化的核心不是模型是數(shù)據(jù)通路、失敗路徑和響應(yīng)時間的設(shè)計。模型會越來越好但這些工程問題永遠(yuǎn)存在。