crewAI Flows 企业级编排事件驱动、状态持久与精确控制本文基于 crewAI v1.11.0深度解析 Flows 框架的事件驱动机制、状态持久化能力和与 Crews 的协同模式。一、为什么需要 Flows当 Crews 不够用的时候Crews 模式非常适合给一个问题让 AI 团队自己解决的场景——探索性强、结果开放、过程动态。但在企业级生产环境中这种自主性有时反而是问题不可预测性Agent 自主决策会产生结果的不确定性无法处理复杂业务逻辑条件分支、循环、错误处理很难通过 Agent 对话实现状态管理困难长时间运行的工作流需要在步骤之间持久化状态缺少精细控制点无法在特定步骤前后插入人工审核、外部系统调用等Flows 就是为了解决这些问题而生。它的定位是用代码控制工作流的骨架用 AICrews填充智能的血肉。二、Flows 的核心概念2.1 基础结构一个 Flow 是一个继承Flow基类的 Python 类每个方法代表一个工作流步骤fromcrewai.flow.flowimportFlow,listen,start,routerclassContentProductionFlow(Flow):start()defresearch_phase(self):工作流入口研究阶段# 执行研究 Crew 或其他逻辑returnresearch_completelisten(research_phase)defwriting_phase(self,research_result):监听 research_phase 完成事件触发写作阶段returndraft_completelisten(writing_phase)defreview_phase(self,draft):监听写作完成触发审核returnfinal_article2.2 三大核心装饰器start()工作流入口start()definitialize(self):Flow 的起始点可以有多个 start 方法并行启动self.state.topiccrewAI 企业级应用returnself.state.topiclisten()事件订阅# 监听单个方法完成listen(initialize)defstep_a(self,input):returna_done# 使用 OR 逻辑任意一个完成即触发fromcrewai.flow.flowimportor_listen(or_(step_a,step_b))defstep_c(self):returnc_done# 使用 AND 逻辑所有完成才触发fromcrewai.flow.flowimportand_listen(and_(step_a,step_b))defstep_d(self):returnd_donerouter()条件路由router(analyze_step)defdecide_direction(self):根据分析结果决定执行哪条分支ifself.state.quality_score0.8:returnhigh_qualityelifself.state.quality_score0.5:returnmedium_qualityelse:returnlow_qualitylisten(high_quality)defpublish_immediately(self):print(质量优秀直接发布)listen(medium_quality)defreview_and_publish(self):print(需要人工审核)listen(low_quality)defreject_and_retry(self):print(质量不达标重新生成)三、状态管理跨步骤的数据共享Flow 中的所有方法共享同一个state对象这是跨步骤传递数据的核心机制。3.1 非结构化状态快速原型fromcrewai.flow.flowimportFlow,start,listenclassSimpleFlow(Flow):start()defstep_one(self):# state 是一个字典可以动态添加任意键self.state[topic]crewAIself.state[research_results][]self.state[word_count]0listen(step_one)defstep_two(self):# 读取上一步存储的状态topicself.state[topic]# 更新状态self.state[word_count]15003.2 结构化状态生产推荐frompydanticimportBaseModelfromtypingimportList,Optionalfromcrewai.flow.flowimportFlow,start,listenclassArticleState(BaseModel):工作流状态定义Pydantic 提供类型安全topic:strkeywords:List[str][]research_summary:strdraft_content:strquality_score:float0.0is_published:boolFalserevision_count:int0classArticleProductionFlow(Flow[ArticleState]):# 泛型参数指定状态类型start()definitialize(self):# IDE 有类型提示self.state.topic 是 str 类型self.state.topiccrewAI 企业级部署最佳实践print(f开始处理主题{self.state.topic})listen(initialize)defresearch(self):# 调用研究 Crew后面详细介绍resultsresearch_crew.kickoff(inputs{topic:self.state.topic})self.state.research_summaryresults.raw self.state.keywordsextract_keywords(results.raw)listen(research)defwrite(self):draftwriting_crew.kickoff(inputs{topic:self.state.topic,keywords:self.state.keywords,research:self.state.research_summary})self.state.draft_contentdraft.rawlisten(write)defevaluate_quality(self):scorequality_checker.check(self.state.draft_content)self.state.quality_scorescorerouter(evaluate_quality)defroute_by_quality(self):ifself.state.quality_score0.85:returnpublishelifself.state.revision_count3:returnreviseelse:returnescalatelisten(publish)defpublish(self):publish_to_cms(self.state.draft_content)self.state.is_publishedTrueprint(f文章已发布质量评分{self.state.quality_score})listen(revise)defrevise(self):self.state.revision_count1print(f第{self.state.revision_count}次修订...)# 重新进入写作步骤循环self.write()listen(escalate)defescalate_to_human(self):send_alert(f文章质量持续不达标需要人工介入。当前评分{self.state.quality_score})# 启动 FlowflowArticleProductionFlow()flow.kickoff()四、状态持久化与断点恢复企业级工作流的一个关键需求是长时间运行的工作流在中途失败后能够从断点继续而不是从头重做。4.1 启用状态持久化fromcrewai.flow.flowimportFlow,start,listen,persistclassLongRunningFlow(Flow[MyState]):start()persist# 该步骤完成后自动持久化状态defexpensive_research(self):# 这个步骤可能需要几分钟持久化后即使崩溃也不用重做resultlong_research_process()self.state.research_dataresultlisten(expensive_research)persistdefexpensive_analysis(self):resultlong_analysis_process(self.state.research_data)self.state.analysis_resultresult4.2 从指定状态恢复# 首次运行flowLongRunningFlow()resultflow.kickoff()# 如果 expensive_research 已完成但 expensive_analysis 崩溃了# 可以从保存的状态恢复跳过已完成的步骤flow_idflow.state.id# 每个 Flow 实例有唯一 ID# 恢复运行flowLongRunningFlow()flow.kickoff(state_idflow_id)# 自动从上次中断点继续五、在 Flow 中嵌套 Crew混合模式的精髓这是 crewAI 最强大的设计之一Flow 提供确定性的业务流程骨架Crew 在需要 AI 智能的节点填充自主执行能力。fromcrewaiimportCrew,Agent,Task,Processfromcrewai.flow.flowimportFlow,start,listen,routerfrompydanticimportBaseModel# ─── 定义各阶段 Crew ───────────────────────────────defcreate_research_crew(topic:str)-Crew:researcherAgent(role研究员,goalf深度研究{topic}的最新发展,backstory...,tools[SerperDevTool()])taskTask(descriptionf全面研究{topic}输出结构化报告,expected_output包含关键发现的研究报告Markdown格式,agentresearcher)returnCrew(agents[researcher],tasks[task],processProcess.sequential)defcreate_writing_crew(topic:str,research:str)-Crew:writerAgent(role技术写作者,goal...,backstory...)editorAgent(role技术编辑,goal...,backstory...)write_taskTask(description基于研究成果撰写技术文章,agentwriter)edit_taskTask(description审核并优化文章质量,agenteditor,context[write_task])returnCrew(agents[writer,editor],tasks[write_task,edit_task],processProcess.sequential)# ─── Flow 骨架 ─────────────────────────────────────classContentState(BaseModel):topic:strresearch:strarticle:strapproved:boolFalseclassContentFlow(Flow[ContentState]):start()defset_topic(self):self.state.topiccrewAI v1.11.0 新特性listen(set_topic)defdo_research(self):在 Flow 步骤中调用 Crewcrewcreate_research_crew(self.state.topic)resultcrew.kickoff()self.state.researchresult.rawlisten(do_research)defdo_writing(self):另一个 Crew 处理写作crewcreate_writing_crew(self.state.topic,self.state.research)resultcrew.kickoff()self.state.articleresult.rawlisten(do_writing)defhuman_review(self):人机回环等待人工审核print(\n 文章已生成等待人工审核 )print(f主题{self.state.topic})print(f摘要{self.state.article[:200]}...)# 在实际系统中这里可能是发送到审核系统并等待回调approvalinput(是否批准发布(y/n): )self.state.approvedapproval.lower()yrouter(human_review)defroute_after_review(self):returnapprovedifself.state.approvedelserejectedlisten(approved)defpublish(self):publish_to_platform(self.state.article)print(✅ 发布成功)listen(rejected)defhandle_rejection(self):print(❌ 审核未通过需要重新生成)# 可以重新触发写作流程self.do_writing()# 运行flowContentFlow()flow.kickoff()六、Flow 的可视化理解执行图crewAI 提供了 Flow 可视化功能将执行图渲染为 HTMLflowContentFlow()flow.plot(content_flow_diagram)# 生成 content_flow_diagram.html在浏览器中可视化查看执行图执行图会展示每个步骤节点依赖关系边条件路由分支并行执行路径七、实际选择何时用 Flow何时用 Crew需要确定性业务流程 ├── 是 ──► Flow │ │ │ ▼ │ 某些步骤需要AI自主处理 │ ├── 是 ──► Flow Crew混合模式 │ └── 否 ──► 纯 FlowPython 逻辑 │ └── 否 ──► Crew 即可 │ ▼ 需要质量监控和动态分配 ├── 是 ──► CrewHierarchical └── 否 ──► CrewSequential八、小结Flows 是 crewAI 从AI 工具进化为企业级平台的关键一步。它不是 Crews 的替代品而是互补品Crews处理如何思考——给 AI 足够的自主权去解决复杂问题Flows处理如何控制——给工程师足够的确定性来管理业务流程在生产环境中最有价值的通常是两者的混合Flow 定义业务流程的确定性骨架在需要 AI 智能的节点嵌入 Crew在需要人工决策的节点设置 HITL 回路。系列导航上一篇crewAI 顺序与层级流程Sequential vs Hierarchical 决策矩阵下一篇crewAI 流程可视化与调试执行链路追踪与性能剖析基于 crewAI v1.11.0 官方文档撰写于 2026 年 3 月
crewAI Flows 企业级编排：事件驱动、状态持久与精确控制
发布时间：2026/7/9 15:49:04
crewAI Flows 企业级编排事件驱动、状态持久与精确控制本文基于 crewAI v1.11.0深度解析 Flows 框架的事件驱动机制、状态持久化能力和与 Crews 的协同模式。一、为什么需要 Flows当 Crews 不够用的时候Crews 模式非常适合给一个问题让 AI 团队自己解决的场景——探索性强、结果开放、过程动态。但在企业级生产环境中这种自主性有时反而是问题不可预测性Agent 自主决策会产生结果的不确定性无法处理复杂业务逻辑条件分支、循环、错误处理很难通过 Agent 对话实现状态管理困难长时间运行的工作流需要在步骤之间持久化状态缺少精细控制点无法在特定步骤前后插入人工审核、外部系统调用等Flows 就是为了解决这些问题而生。它的定位是用代码控制工作流的骨架用 AICrews填充智能的血肉。二、Flows 的核心概念2.1 基础结构一个 Flow 是一个继承Flow基类的 Python 类每个方法代表一个工作流步骤fromcrewai.flow.flowimportFlow,listen,start,routerclassContentProductionFlow(Flow):start()defresearch_phase(self):工作流入口研究阶段# 执行研究 Crew 或其他逻辑returnresearch_completelisten(research_phase)defwriting_phase(self,research_result):监听 research_phase 完成事件触发写作阶段returndraft_completelisten(writing_phase)defreview_phase(self,draft):监听写作完成触发审核returnfinal_article2.2 三大核心装饰器start()工作流入口start()definitialize(self):Flow 的起始点可以有多个 start 方法并行启动self.state.topiccrewAI 企业级应用returnself.state.topiclisten()事件订阅# 监听单个方法完成listen(initialize)defstep_a(self,input):returna_done# 使用 OR 逻辑任意一个完成即触发fromcrewai.flow.flowimportor_listen(or_(step_a,step_b))defstep_c(self):returnc_done# 使用 AND 逻辑所有完成才触发fromcrewai.flow.flowimportand_listen(and_(step_a,step_b))defstep_d(self):returnd_donerouter()条件路由router(analyze_step)defdecide_direction(self):根据分析结果决定执行哪条分支ifself.state.quality_score0.8:returnhigh_qualityelifself.state.quality_score0.5:returnmedium_qualityelse:returnlow_qualitylisten(high_quality)defpublish_immediately(self):print(质量优秀直接发布)listen(medium_quality)defreview_and_publish(self):print(需要人工审核)listen(low_quality)defreject_and_retry(self):print(质量不达标重新生成)三、状态管理跨步骤的数据共享Flow 中的所有方法共享同一个state对象这是跨步骤传递数据的核心机制。3.1 非结构化状态快速原型fromcrewai.flow.flowimportFlow,start,listenclassSimpleFlow(Flow):start()defstep_one(self):# state 是一个字典可以动态添加任意键self.state[topic]crewAIself.state[research_results][]self.state[word_count]0listen(step_one)defstep_two(self):# 读取上一步存储的状态topicself.state[topic]# 更新状态self.state[word_count]15003.2 结构化状态生产推荐frompydanticimportBaseModelfromtypingimportList,Optionalfromcrewai.flow.flowimportFlow,start,listenclassArticleState(BaseModel):工作流状态定义Pydantic 提供类型安全topic:strkeywords:List[str][]research_summary:strdraft_content:strquality_score:float0.0is_published:boolFalserevision_count:int0classArticleProductionFlow(Flow[ArticleState]):# 泛型参数指定状态类型start()definitialize(self):# IDE 有类型提示self.state.topic 是 str 类型self.state.topiccrewAI 企业级部署最佳实践print(f开始处理主题{self.state.topic})listen(initialize)defresearch(self):# 调用研究 Crew后面详细介绍resultsresearch_crew.kickoff(inputs{topic:self.state.topic})self.state.research_summaryresults.raw self.state.keywordsextract_keywords(results.raw)listen(research)defwrite(self):draftwriting_crew.kickoff(inputs{topic:self.state.topic,keywords:self.state.keywords,research:self.state.research_summary})self.state.draft_contentdraft.rawlisten(write)defevaluate_quality(self):scorequality_checker.check(self.state.draft_content)self.state.quality_scorescorerouter(evaluate_quality)defroute_by_quality(self):ifself.state.quality_score0.85:returnpublishelifself.state.revision_count3:returnreviseelse:returnescalatelisten(publish)defpublish(self):publish_to_cms(self.state.draft_content)self.state.is_publishedTrueprint(f文章已发布质量评分{self.state.quality_score})listen(revise)defrevise(self):self.state.revision_count1print(f第{self.state.revision_count}次修订...)# 重新进入写作步骤循环self.write()listen(escalate)defescalate_to_human(self):send_alert(f文章质量持续不达标需要人工介入。当前评分{self.state.quality_score})# 启动 FlowflowArticleProductionFlow()flow.kickoff()四、状态持久化与断点恢复企业级工作流的一个关键需求是长时间运行的工作流在中途失败后能够从断点继续而不是从头重做。4.1 启用状态持久化fromcrewai.flow.flowimportFlow,start,listen,persistclassLongRunningFlow(Flow[MyState]):start()persist# 该步骤完成后自动持久化状态defexpensive_research(self):# 这个步骤可能需要几分钟持久化后即使崩溃也不用重做resultlong_research_process()self.state.research_dataresultlisten(expensive_research)persistdefexpensive_analysis(self):resultlong_analysis_process(self.state.research_data)self.state.analysis_resultresult4.2 从指定状态恢复# 首次运行flowLongRunningFlow()resultflow.kickoff()# 如果 expensive_research 已完成但 expensive_analysis 崩溃了# 可以从保存的状态恢复跳过已完成的步骤flow_idflow.state.id# 每个 Flow 实例有唯一 ID# 恢复运行flowLongRunningFlow()flow.kickoff(state_idflow_id)# 自动从上次中断点继续五、在 Flow 中嵌套 Crew混合模式的精髓这是 crewAI 最强大的设计之一Flow 提供确定性的业务流程骨架Crew 在需要 AI 智能的节点填充自主执行能力。fromcrewaiimportCrew,Agent,Task,Processfromcrewai.flow.flowimportFlow,start,listen,routerfrompydanticimportBaseModel# ─── 定义各阶段 Crew ───────────────────────────────defcreate_research_crew(topic:str)-Crew:researcherAgent(role研究员,goalf深度研究{topic}的最新发展,backstory...,tools[SerperDevTool()])taskTask(descriptionf全面研究{topic}输出结构化报告,expected_output包含关键发现的研究报告Markdown格式,agentresearcher)returnCrew(agents[researcher],tasks[task],processProcess.sequential)defcreate_writing_crew(topic:str,research:str)-Crew:writerAgent(role技术写作者,goal...,backstory...)editorAgent(role技术编辑,goal...,backstory...)write_taskTask(description基于研究成果撰写技术文章,agentwriter)edit_taskTask(description审核并优化文章质量,agenteditor,context[write_task])returnCrew(agents[writer,editor],tasks[write_task,edit_task],processProcess.sequential)# ─── Flow 骨架 ─────────────────────────────────────classContentState(BaseModel):topic:strresearch:strarticle:strapproved:boolFalseclassContentFlow(Flow[ContentState]):start()defset_topic(self):self.state.topiccrewAI v1.11.0 新特性listen(set_topic)defdo_research(self):在 Flow 步骤中调用 Crewcrewcreate_research_crew(self.state.topic)resultcrew.kickoff()self.state.researchresult.rawlisten(do_research)defdo_writing(self):另一个 Crew 处理写作crewcreate_writing_crew(self.state.topic,self.state.research)resultcrew.kickoff()self.state.articleresult.rawlisten(do_writing)defhuman_review(self):人机回环等待人工审核print(\n 文章已生成等待人工审核 )print(f主题{self.state.topic})print(f摘要{self.state.article[:200]}...)# 在实际系统中这里可能是发送到审核系统并等待回调approvalinput(是否批准发布(y/n): )self.state.approvedapproval.lower()yrouter(human_review)defroute_after_review(self):returnapprovedifself.state.approvedelserejectedlisten(approved)defpublish(self):publish_to_platform(self.state.article)print(✅ 发布成功)listen(rejected)defhandle_rejection(self):print(❌ 审核未通过需要重新生成)# 可以重新触发写作流程self.do_writing()# 运行flowContentFlow()flow.kickoff()六、Flow 的可视化理解执行图crewAI 提供了 Flow 可视化功能将执行图渲染为 HTMLflowContentFlow()flow.plot(content_flow_diagram)# 生成 content_flow_diagram.html在浏览器中可视化查看执行图执行图会展示每个步骤节点依赖关系边条件路由分支并行执行路径七、实际选择何时用 Flow何时用 Crew需要确定性业务流程 ├── 是 ──► Flow │ │ │ ▼ │ 某些步骤需要AI自主处理 │ ├── 是 ──► Flow Crew混合模式 │ └── 否 ──► 纯 FlowPython 逻辑 │ └── 否 ──► Crew 即可 │ ▼ 需要质量监控和动态分配 ├── 是 ──► CrewHierarchical └── 否 ──► CrewSequential八、小结Flows 是 crewAI 从AI 工具进化为企业级平台的关键一步。它不是 Crews 的替代品而是互补品Crews处理如何思考——给 AI 足够的自主权去解决复杂问题Flows处理如何控制——给工程师足够的确定性来管理业务流程在生产环境中最有价值的通常是两者的混合Flow 定义业务流程的确定性骨架在需要 AI 智能的节点嵌入 Crew在需要人工决策的节点设置 HITL 回路。系列导航上一篇crewAI 顺序与层级流程Sequential vs Hierarchical 决策矩阵下一篇crewAI 流程可视化与调试执行链路追踪与性能剖析基于 crewAI v1.11.0 官方文档撰写于 2026 年 3 月
