The short version: CrewAI Crews let a team of agents self-organize; Flows are the opposite tool — an event-driven graph where you pin down exactly what runs when. And you express the entire graph with five decorators. Once you know @start, @listen, @router, and the and_/or_ combinators, you can fan work out to run in parallel, join it back together, and branch on a decision — without writing a single line of threading, async plumbing, or a hand-rolled state machine. Here's the whole vocabulary with code you can paste.

The mental model: methods that listen for each other#

A Flow is a class. Each step is a method. You annotate a method to say what makes it run, and CrewAI schedules the rest — threading self.state between steps and deciding execution order from the annotations.

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

class GreetFlow(Flow):
    @start()
    def fetch_name(self):
        self.state["name"] = "Ada"
        return "fetched"

    @listen(fetch_name)
    def greet(self, _):
        return f"Hello, {self.state['name']}"

GreetFlow().kickoff()

@start() marks the entry point. @listen(fetch_name) says run greet when fetch_name emits. That's the atom every other pattern is built from.

Fan-out: parallel steps with zero threading code#

Here's the part that saves the most time. To run two steps at the same time, point both at the same upstream step. CrewAI sees two listeners on one emitter and schedules them concurrently — you write no asyncio, no ThreadPoolExecutor, no locks.

@start()
def load(self):
    self.state["doc"] = fetch_document()

@listen(load)
def summarize(self, _):
    self.state["summary"] = llm_summary(self.state["doc"])

@listen(load)
def extract_entities(self, _):
    self.state["entities"] = llm_entities(self.state["doc"])

summarize and extract_entities both depend only on load, so they run in parallel. Two independent LLM calls that would otherwise run back-to-back now overlap — often halving wall-clock latency for the kind of "do several things to one input" step that's everywhere in agent pipelines.

Join: and_ waits for all, or_ waits for any#

Parallel branches are only useful if you can bring them back together, and CrewAI gives you two combinators for exactly that. and_() is the join you reach for most often: it makes a method wait until every listed upstream step has finished before it runs, and even then it runs only once. or_() is its mirror image, firing the moment any one of several upstream steps completes. Here is the join in code:

from crewai.flow.flow import listen, and_, or_

@listen(and_(summarize, extract_entities))
def build_report(self, _):
    # runs ONLY after BOTH branches finish
    return compose(self.state["summary"], self.state["entities"])

and_() is the join: build_report fires exactly once, after every listed step has completed. Use it whenever a synthesis step needs the output of multiple parallel branches.

or_() is the mirror image — it fires as soon as any listed step emits:

@listen(or_(finished_from_cache, finished_from_api))
def deliver(self, result):
    # runs as soon as EITHER path completes
    return result

Reach for or_() when several different paths should converge on the same next step — a cache hit or a fresh fetch both lead to "deliver," and you don't want to duplicate that method.

Branch: @router for conditional paths#

Fan-out and join handle parallel work. @router handles choices. A router method returns a string label, and downstream methods listen for that specific label — so the flow takes one branch or another based on a runtime decision.

from crewai.flow.flow import router, listen

@router(classify)
def route_by_intent(self):
    if self.state["intent"] == "refund":
        return "refund"
    return "question"

@listen("refund")
def handle_refund(self):
    ...

@listen("question")
def handle_question(self):
    ...

This is the clean alternative to if/else smeared across your steps: the branching decision lives in one place (route_by_intent), and each branch is its own named, testable method. A router can return different labels on different runs, which is how you get an agent to decide its own path rather than following a fixed script.

State: one shared object, optionally typed#

Every step reads and writes self.state, so you never thread return values from one method to the next by hand — the state object is the shared blackboard every step draws on. For a quick prototype, leave it an untyped dict and move fast. For anything you intend to run in production, pass a Pydantic model as the flow's type parameter and get typed fields, validation, and editor autocomplete across every step of the flow:

from pydantic import BaseModel
from crewai.flow.flow import Flow, start

class ReportState(BaseModel):
    doc: str = ""
    summary: str = ""
    entities: list[str] = []

class ReportFlow(Flow[ReportState]):
    @start()
    def load(self):
        self.state.doc = fetch_document()   # typed, validated

Structured state also carries an auto-generated id field, which is what makes the next feature possible.

Persist: survive a crash and resume#

A long flow that dies three steps in shouldn't restart from zero. Add @persist and CrewAI checkpoints self.state (SQLite by default) after each step; on restart it rehydrates and continues from where it stopped. We walk the full recovery path — thread IDs, checkpointer, and resuming a half-finished run — in how to resume a crashed CrewAI Flow.

Putting it together#

The five decorators compose into any topology you need: @start in, fan out with parallel @listens, @router to pick a branch, and_/or_ to join, @persist to make it durable. That's the appeal of Flows over a free-form Crew — when the orchestration has to be exact and inspectable, you get a real control-flow graph without hand-writing the scheduler. Pair it with pluggable memory backends and the flow becomes both deterministic and stateful — the two things a fragile agent script usually lacks.