★ FEATURED ARTICLE CrewAI Flows从角色编排到图编排的进化start/listen/router三个装饰器让CrewAI从框架内黑盒编排进化为显式事件驱动图——附三人Crew项目的完整迁移实录本文导航传统Crew的天花板Flows是什么事件驱动的显式图核心API三个装饰器打天下状态管理从黑盒到pydantic类型化Crew与Flow混编两代范式的合体迁移实战三人Crew改造为Flow迁移前后对比五处质变小结下节预告互动区传统Crew的天花板用了两节传统Crew之后你应该已经摸到它的边界了。我把实战里撞过的墙列出来看看你有几面眼熟工序之间插不进确定性代码。想在校验和写作之间加一段如果分析简报字数超限就截断的Python逻辑没有地方写——Crew的执行循环在框架内部你的代码只能在kickoff()之前和之后存在没有重试、没有条件、没有循环。第63节结尾我提过sequential模式下第3张工单挂了整个Crew陪葬。想让审核不过就打回重写最多三轮传统Crew表达不了多Crew协作是隐式的。CrewA的产出怎么进CrewB你自己在CrewA外面拿result.raw自己写代码喂给CrewB的inputs——编排逻辑散落在你的胶水代码里没有一个统一的地方声明整个系统结构中间状态不可见。跑完一个Crewtasks_output里能拿到各工单产出但为什么走了这条路、哪一步改了什么这类过程信息是黑盒。这些墙的共同根源传统Crew把编排做成了框架内部的黑盒你只被允许声明人和事不允许声明流程本身。回想第8节讲的拓扑模式、第22节讲的条件边——图工程的全部表达力在传统Crew里都没有入口。CrewAI官方显然也撞到了这堵墙。他们的答案就是Flows在Crew之上加一层显式的、事件驱动的工作流编排。一句话定位Flow管流程图Crew管协作角色两层各干各的。Flows是什么事件驱动的显式图Flows的核心思想把系统的流程结构从框架黑盒里拿出来变成你代码里显式声明的图。图的节点是普通Python方法或Crew调用图的边由装饰器声明谁启动、谁监听谁、按什么条件路由。先看官方风格的最小例子感受画风fromcrewai.flow.flowimportFlow,listen,startclassCityFlow(Flow):modelgpt-4o-ministart()# 入口节点Flow启动即执行defgenerate_city(self):print(fFlow State ID:{self.state[id]})# 每个Flow自动带UUIDcityllm_call(返回一个随机城市名)self.state[city]city# 状态写入全Flow可见returncity# 返回值 事件输出触发下游listen(generate_city)# 边监听generate_city的输出defgenerate_fun_fact(self,city):factllm_call(f讲一个关于{city}的冷知识)self.state[fun_fact]factreturnfact flowCityFlow()resultflow.kickoff()# 返回最后一个被触发节点的输出print(fGenerated fun fact:{result})结构一目了然start()标记入口listen(某方法)声明依赖边方法返回值沿边流动self.state全局共享。这就是一张显式图——和第57节ADK的edges、第44节LangGraph的StateGraph同属一个物种只是语法外衣是装饰器。对应的图结构返回值作为事件self.state写入kickoffgenerate_citystartgenerate_fun_factlisten(generate_city)最终输出state含自动UUIDflow.plot()还能把这张图渲染成HTML可视化——CrewAI终于有了自己的draw_mermaid时刻第56节给LangGraph加分的那个能力。核心API三个装饰器打天下Flows的API面小得可爱三个装饰器一个基类覆盖了图工程的全部基本需求。①start()入口节点。标记Flow的起点kickoff()时执行。可以标记多个方法它们并行启动——对Flows天生支持多起点扇出classMultiSourceFlow(Flow):start()deffetch_rss(self):...# 这两个start方法start()deffetch_db(self):...# 在kickoff时并行执行②listen()依赖边。监听另一个方法的输出触发执行。四种姿势按需取用listen(generate_city)# 姿势一监听方法本身推荐重构安全defa(self,city):...listen(generate_city)# 姿势二监听方法名字符串defb(self,city):...listen(or_(fetch_rss,fetch_db))# 姿势三任一完成即触发并行的OR汇合defc(self,whichever_output):...listen(and_(check_a,check_b))# 姿势四全部完成才触发AND汇合JoinNodedefd(self,last_output):...or_/and_这两个组合器就是图工程的Fan-out/Join在装饰器语法下的对应物——第58节ADK的JoinNode语义这里一行装饰器搞定。③router()条件边。监听上游输出返回一个标签字符串下游用start(标签)或listen(router方法, 标签)接住router(analyze_sentiment)# 监听analyze_sentiment的输出defroute_by_sentiment(self,score):returnpositiveifscore0elsenegative# 返回路由标签start(positive)# 标签命中才启动defhappy_path(self):...start(negative)defescalation(self):...对比第57节ADK的Event(routeRUN_TASK_B) 字典路由同一套Router拓扑两家语法。学到这里你应该有感觉了——图工程的模式语言是通用货币框架只是汇率不同的钱包。状态管理从黑盒到pydantic类型化传统Crew的状态是黑盒框架内部的执行记录Flows把状态拉到你的代码里而且给了两种用法。用法一默认dict状态。self.state就是个字典自动带一个唯一idUUID方便追踪。写起来随意适合快速原型self.state[city]巴黎用法二pydantic类型化状态我强烈推荐——继承Flow时传入状态模型IDE补全、类型检查、字段约束全部到位frompydanticimportBaseModel,FieldclassContentState(BaseModel):topic:stroutline:str|NoneNonedraft:str|NoneNonereview_passed:boolFalserevision_count:intField(default0,ge0,le5)# 约束最多5轮修订classContentFlow(Flow[ContentState]):# 泛型参数声明状态类型start()defresearch(self):self.state.topicGraph Engineeringreturnreadylisten(research)defwrite(self):print(self.state.topic)# IDE全程补全...这就是第27节状态Schema设计的老结论再次应验TypedDict/pydantic/自定义的选择题每个框架都用一次答案都一样——生产用pydantic。另外两个状态特性记住就好Flow(persistTrue)可以把状态持久化到SQLite进程重启后能接着跑断点续跑的轻量版flow.restart()从头重置状态。Crew与Flow混编两代范式的合体Flows最实用的能力Crew可以作为一个普通的被listen触发的方法调用——返回crew.kickoff()的结果就行。于是两代范式合体Flow层声明系统结构图Crew层封装角色协作子图。classIntelFlow(Flow):start()defcollect(self):returnfetch_all_sources()# 确定性代码想怎么写怎么写listen(collect)defanalyze(self,raw_data):resultanalysis_crew.kickoff(# CrewA分析师审查员的角色协作inputs{data:raw_data})returnresult.rawlisten(analyze)defpublish(self,insight):ifself.state.revision_count5:# 确定性逻辑重试上限控制returnfallback_publish(insight)returnwrite_report(insight)对照着ADK的嵌套Workflow第58节看Crew之于Flow就是子Workflow之于父图。角色协作这种模糊的、提示词驱动的工作打包成Crew黑盒系统级的精确的、代码驱动的结构用Flow显式声明——分层封装各得其所。这个结构就是我做企业项目的主力架构第65、66节会把它推到完整形态。迁移实战三人Crew改造为Flow迁移案例来了。把第62节的分析师文案编辑Crew改造成Flow迁移过程中暴露的正是传统Crew的四面墙。需求升级审核不过自动打回重写最多2轮、最终报告带修订轮数、分析简报超长自动截断。# flow_version.py —— 三人内容Crew迁移为Flow重试/条件/循环全部显式化fromcrewaiimportAgent,Crew,Process,Taskfromcrewai.flow.flowimportFlow,listen,router,startfrompydanticimportBaseModel,Fieldfromdotenvimportload_dotenv load_dotenv()# ---------- 类型化状态全流程单一真理源 ----------classContentState(BaseModel):topic:strbrief:str|NoneNone# 分析简报draft:str|NoneNone# 文案稿final:str|NoneNone# 终稿revisions:intField(default0,le2)# 修订上限写进约束review_comments:list[str][]# ---------- 三个角色原样搬过来角色定义是框架无关资产 ----------analystAgent(role资深市场分析师,goal从产品方向提炼最能打动用户的卖点洞察,backstory十年消费品市场分析擅长痛点→价值框架反感空洞形容词。)copywriterAgent(role资深营销文案,goal把洞察转化成有记忆点的传播文案,backstory写过500产品文案信奉一句话让人记住一个产品。)editorAgent(role挑剔的审核编辑,goal确保文案事实准确、无夸大表述,backstory前4A文案总监对绝对化表述零容忍意见具体到句子。)# ---------- 确定性节点传统Crew里没有位置Flow里随便写 ----------deftrim_brief(brief:str,limit:int800)-str:简报超长截断——纯Python逻辑插在两道AI工序之间returnbrief[:limit](\n...(已截断)iflen(brief)limitelse)# ---------- Flow系统结构显式声明 ----------classContentFlow(Flow[ContentState]):start()defresearch(self):节点1分析师Crew工序这里用单Task示意复杂场景可整Crew挂载tTask(descriptionf围绕{self.state.topic}输出卖点简报用户画像/痛点/3个卖点,expected_output300字内Markdown简报含3个卖点及支撑理由,agentanalyst)crewCrew(agents[analyst],tasks[t],processProcess.sequential)briefcrew.kickoff().raw self.state.brieftrim_brief(brief)# 确定性后处理随手就插returnself.state.brieflisten(research)defwrite(self,brief:str):节点2文案起草tTask(descriptionf基于简报写营销文案主标题/正文三句/行动号召简报\n{brief},expected_output按主标题/正文/行动号召三段结构120字内,agentcopywriter)crewCrew(agents[copywriter],tasks[t],processProcess.sequential)self.state.draftcrew.kickoff().rawreturnself.state.draftrouter(write)# 条件边审核结果决定走向defreview(self,draft:str):节点3编辑审核 → 路由tTask(descriptionf审核文案是否可发布卖点一致/无夸大\n{draft},expected_output输出JSON: {pass: true/false, comments: [修改意见]},agenteditor)crewCrew(agents[editor],tasks[t],processProcess.sequential)importjson verdictjson.loads(crew.kickoff().raw)self.state.review_commentsverdict.get(comments,[])# 修订超限强制放行防止无限打回——第32节兜底策略的Flow版ifverdict[pass]orself.state.revisions2:returnapprovedself.state.revisions1returnreviselisten(review,revise)# 标签路由打回重写defrevise(self):tTask(descriptionf按审核意见修改文案第{self.state.revisions}轮修订 原稿{self.state.draft}意见{; .join(self.state.review_comments)},expected_output修订后的完整文案保持三段结构,agentcopywriter)crewCrew(agents[copywriter,editor],tasks[t],processProcess.sequential)self.state.draftcrew.kickoff().rawreturnself.state.draftlisten(revise)defre_review(self,draft:str):修订稿回到审核收敛循环的回路returnself.review(draft)# 复用审核router形成环listen(review,approved)# 标签路由通过收尾deffinalize(self):self.state.finalself.state.draftreturnf[终稿通过 | 修订{self.state.revisions}轮]\n{self.state.final}if__name____main__:flowContentFlow()resultflow.kickoff(inputs{topic:AI代码审查机器人})print(result)注意write → review(router) → revise → re_review → review这条环Convergent Loop第39节让迭代有终点在CrewAI里的第一次原生表达——修订上限写在State的pydantic约束里兜底逻辑写在router的判断里循环的出口有两个通过/超限放行。这在传统Crew里需要外层for循环手动拼inputs才能模拟现在全是声明式的图结构。控制台输出[2026-09-27 14:20:01][INFO]: Flow started: ContentFlow (id: 3f2a8b1e-...) [research] 分析师产出简报 412字 [write] 文案起草完成 [review] 审核未通过: [秒杀人工审查为绝对化表述, 行动号召缺少具体动作] [revise] 第1轮修订完成 [re_review] 复审中... [review] 审核通过 [finalize] 终稿入库 [终稿通过 | 修订1轮] 【主标题】让AI替你盯住每一行代码 【正文】它读过你仓库的每一次提交比新人更懂你的代码风格。上班摸鱼两小时AI替你审完两百个PR。误报率低于行业均值值得托付。 【行动号召】本周期内测名额100个扫码即抢。flow.plot()生成的图上能清楚看到那个修订环write→review→revise→re_review→review。结构即文档新人看图就能懂这个系统的行为逻辑——传统Crew时代这得靠口口相传。迁移前后对比五处质变维度传统CrewFlow版质变流程结构框架黑盒靠工单顺序隐式表达start/listen/router显式声明结构可审查、可plot可视化确定性逻辑无处安放散在胶水代码任意两个AI工序之间随意插入AI与代码混合编排成为一等公民重试与循环不支持router标签方法环原生表达修订环上限兜底失败可收敛状态tasks_output事后翻pydantic类型化全程单一真理源IDE补全字段约束追踪UUID多Crew协作外部胶水代码串联Crew作为Flow节点挂载系统级结构统一在一处声明迁移成本坦白讲角色定义Agent和工单描述Task原样带走零改动——再一次验证第56节和第61节的结论提示词和角色资产是框架无关的真正要重写的只有编排层。这次迁移我花了不到两小时其中一半时间在调试router的标签拼写标签是字符串拼错了不会报错、只会静默不触发——这是个坑盯紧。什么时候该迁移工单3张以内、无重试需求传统Crew完全够用别为了图而图一旦出现重试、条件分支、多Crew、确定性工序混排中的任何一条上Flow别犹豫。小结Flows是CrewAI向图工程的正式靠拢start入口多起点并行、listen依赖边or_/and_组合实现Fan-out/Join、router条件边标签路由——三个装饰器覆盖图工程基本表达力。状态两档用法dict快速原型pydantic生产必选Flow[StateModel]泛型声明字段约束persistTrue可持久化断点续跑。Crew与Flow是分层关系不是替代关系Flow声明系统结构图Crew封装角色协作子图黑盒——对应ADK的嵌套Workflow、LangGraph的子图。迁移实测角色与工单资产零改动带走编排层重写约两小时修订环、截断逻辑、上限兜底这些传统Crew写不出来的东西全部显式化。一个新坑router标签是字符串拼错静默不触发——建议标签常量化管理。下节预告有了三角架构、三种流程、Flows图编排CrewAI的单系统战斗力已经齐了。下一节进入企业级多Crew协作模式、三层记忆系统Short-term/Long-term/Entity Memory、自定义工具集成并用一个销售Crew技术Crew法务Crew协作处理客户方案的完整案例把企业模式串起来。多Agent系统的组织设计学才是CrewAI区别于图框架的真正主场。互动区你手头的Crew项目里有没有那种用for循环包着crew.kickoff()模拟重试的胶水代码或者两个Crew的产出在main函数里手工搬运的拼接现场评论区晒出你最别扭的一段编排胶水我们一起看看用Flow改造后会是什么形状——点赞最高的案例我画成before/after双图放进下一节。如果觉得本文对你有帮助欢迎点赞、收藏、关注三连本系列持续更新中关注不迷路~
阅读完成 · 觉得有帮助？
