# CrewAI Flows — Event-Driven Multi-Agent Orchestration

> CrewAI Flows is the event-driven orchestration layer on top of Crews. Decorators @start, @listen, @router build a typed state machine for multi-agent.

## Install

Copy the content below into your project:

## Quick Use

1. `pip install 'crewai[tools]'>=0.70`
2. `from crewai.flow.flow import Flow, start, listen`
3. Define a Flow subclass with `@start` and `@listen` methods, call `flow.kickoff()`

---

## Intro

CrewAI Flows is the event-driven orchestration layer that complements CrewAI's role-based Crews. Decorators (`@start`, `@listen`, `@router`) build a typed state machine — perfect for production agents where Crews alone become hard to reason about. Best for: multi-step agent pipelines, branching workflows, anything beyond "3 agents talk in a loop". Works with: CrewAI 0.70+, Python 3.10+. Setup time: 2 minutes.

---

### Hello-flow

```python
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):
        # use a Crew of search-specialist agents
        return search_crew.kickoff({"topic": topic})

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

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

### Branching with @router

```python
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()
```

### State persistence

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

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

State survives across method calls and persists to disk if you set `persistence=SQLiteFlowPersistence()`. Useful for long-running agents that need recovery.

---

### FAQ

**Q: Crew vs Flow — when to pick which?**
A: Crew is best for natural agent-to-agent collaboration (researcher + writer + editor). Flow is best for explicit sequencing, branching, and recovery — when you'd otherwise hand-roll a state machine.

**Q: Is CrewAI Flows free?**
A: Yes — CrewAI is open-source under MIT license. The framework, Crews, and Flows are all free. CrewAI also offers an enterprise hosted platform with monitoring and UI.

**Q: Can a Flow contain Crews and vice versa?**
A: Yes — Flow methods commonly call `crew.kickoff()` to delegate sub-tasks to a Crew. The reverse (Crew calling a Flow) is less common but possible by exposing the Flow as a tool.

---

## Source & Thanks

> Built by [crewAIInc](https://github.com/crewAIInc). Licensed under MIT.
>
> [crewAIInc/crewAI](https://github.com/crewAIInc/crewAI) — ⭐ 30,000+

---

<!-- ZH -->

## 快速使用

1. `pip install 'crewai[tools]'>=0.70`
2. `from crewai.flow.flow import Flow, start, listen`
3. 定义 Flow 子类，写 `@start` 和 `@listen` 方法，调 `flow.kickoff()`

---

## 简介

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

---

### Hello-flow

```python
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 做分支

```python
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()
```

### 状态持久化

```python
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 暴露成工具来实现。

---

## 来源与感谢

> Built by [crewAIInc](https://github.com/crewAIInc). Licensed under MIT.
>
> [crewAIInc/crewAI](https://github.com/crewAIInc/crewAI) — ⭐ 30,000+


---
Source: https://tokrepo.com/en/workflows/crewai-flows-event-driven-multi-agent-orchestration
Author: CrewAI