CrewAI 完全指南(三):進階技巧——Flows 事件驅動、Memory 記憶體、與生產部署
前言 前兩篇建立了 CrewAI 的基礎和實戰應用。 這篇是進階篇,涵蓋讓 CrewAI 真正走到生產環境的關鍵技術: Flows:當 Crew 的線性流程不夠用時 Memory 記憶體:讓 Agent 記得過去的對話和經驗 錯誤處理與成本控制:生產環境的必要設計 部署:把 CrewAI 包成 API 服務 Part 1:CrewAI Flows——事件驅動的複雜工作流程 Crew 的限制 Process.sequential 是線性的:任務一個接一個執行。但真實世界的工作流程往往需要: 條件分支:根據分析結果走不同的路徑 迴圈:重複執行直到滿足條件 平行執行:多個 Crew 同時跑,最後彙整 狀態管理:跨步驟保存和傳遞複雜的狀態 CrewAI Flows 就是為了處理這些複雜場景設計的。 Flow 的三個核心 Decorator 1from crewai.flow.flow import Flow, listen, start, router 2from pydantic import BaseModel 3 4class MyFlow(Flow): 5 6 @start() 7 def step_one(self): 8 """Flow 的入口點,Flow 啟動時執行""" 9 return "step one result" 10 11 @listen(step_one) 12 def step_two(self, step_one_output): 13 """當 step_one 完成後自動觸發,可以接收上一步的輸出""" 14 return f"processed: {step_one_output}" 15 16 @router(step_two) 17 def decide_next(self, step_two_output): 18 """根據 step_two 的輸出決定下一步走哪條路""" 19 if "error" in step_two_output: 20 return "error_path" 21 return "success_path" 22 23 @listen("success_path") 24 def handle_success(self): 25 return "成功!" 26 27 @listen("error_path") 28 def handle_error(self): 29 return "處理錯誤.