CrewAI Flows — Event-Driven Multi-Agent Orchestration

简介

CrewAI Flows 是 CrewAI 角色化 Crew 之上的事件驱动编排层。装饰器（@start / @listen / @router）拼出类型化状态机 —— 在生产里 Crew 变得难推理时正好补位。适合多步 agent 流水线、分支工作流、任何超出「3 个 agent 循环对话」的场景。需要 CrewAI 0.70+ 和 Python 3.10+。装机时间 2 分钟。

Hello-flow

from crewai.flow.flow import Flow, listen, start

class ResearchFlow(Flow):

    @start()
    def find_topic(self):
        return "open-source vector databases 2026"

    @listen(find_topic)
    def search_web(self, topic):
        # 调用 search-specialist agent 组成的 Crew
        return search_crew.kickoff({"topic": topic})

    @listen(search_web)
    def summarize(self, results):
        return summarizer_crew.kickoff({"results": results})

flow = ResearchFlow()
final = flow.kickoff()

用 @router 做分支

class BugFlow(Flow):

    @start()
    def classify(self):
        return classifier_crew.kickoff()

    @router(classify)
    def route(self, classification):
        return "fix" if classification.is_bug else "explain"

    @listen("fix")
    def fix_bug(self):
        return fixer_crew.kickoff()

    @listen("explain")
    def explain(self):
        return explainer_crew.kickoff()

状态持久化

class StatefulFlow(Flow[ResearchState]):
    initial_state = ResearchState(topic="", findings=[])

    @start()
    def init(self):
        self.state.topic = "agent frameworks"

状态跨方法调用保留，设 persistence=SQLiteFlowPersistence() 可落盘。适合需要恢复的长时 agent。

FAQ

Q: Crew 和 Flow 怎么选？ A: Crew 适合 agent 之间自然协作（researcher + writer + editor）。Flow 适合显式排序、分支、恢复 —— 否则你要自己手撸状态机。

Q: CrewAI Flows 免费吗？ A: 免费 —— CrewAI MIT 开源。框架、Crew、Flow 都免费。CrewAI 也提供带监控和 UI 的企业托管版。

Q: Flow 能包 Crew 反过来呢？ A: 能 —— Flow 方法常调 crew.kickoff() 把子任务委派给 Crew。反向（Crew 调 Flow）少见但可以把 Flow 暴露成工具来实现。