CrewAI Flows: Decorator Orchestration for Production Pipelines

Crews vs Flows

A Crew is a frozen plan: list of agents, list of tasks, one process. A Flow is a graph of methods that decides at runtime what comes next. Use Crews for "research → write" pipelines; reach for Flows when you have branching, retries, human pauses, or several crews chained together.

The Flow Class

Subclass Flow[StateType] to get typed Pydantic state. kickoff() runs synchronously on the flow; kickoff_async() returns a coroutine for async hosts.

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

class State(BaseModel):
    topic: str = ""
    draft: str = ""

class WriteFlow(Flow[State]):
    @start()
    def begin(self):
        self.state.topic = "AI agents"
        return self.state.topic

    @listen(begin)
    def draft(self, topic):
        self.state.draft = f"Article on {topic}"
        return self.state.draft

flow = WriteFlow()
print(flow.kickoff().raw)

Control Primitives

• ▸@start() — entry point. Multiple starts allowed; conditional via a state key argument.
• ▸@listen(method) — fires after the upstream completes. Combine and_() for fan-in joins or or_() when the first branch to finish should trigger downstream work.
• ▸@router(method) — return a string tag; downstream listeners listen on the tag value.
• ▸@persist() — save state to SqliteProvider (default) or JsonProvider. Pair with CheckpointConfig.

State Management

Two flavors:

• •Structured — Flow[MyPydanticModel]. Recommended for production: typed, validated, serializable.
• •Unstructured — dict-like self.state["key"]. Quick, fragile.

Persistence & Resumability

Apply @persist() to the class. State is keyed by an automatic state ID; pass it back into flow.kickoff() to resume. Override the backend via CheckpointConfig(storage=JsonProvider()).

Streaming

Pass stream=True to Flow constructor. Listen to LLMStreamChunkEvent and LLMThinkingChunkEvent via a BaseEventListener to forward chunks to your UI.

Adding Crews to Flows

Flows can call Crews as just another step:

@listen(begin)
def research(self, topic):
    crew = ResearchCrew().crew()
    result = crew.kickoff(inputs={"topic": topic})
    self.state.draft = result.raw
    return result.raw

Plotting

Call flow.plot("my_flow") to render an HTML diagram of starts/listeners/routers. Great for documentation and reviews.

Reference

For cross-run facts outside typed Flow state, use the unified memory APIs self.remember() and self.recall() so agents retrieve semantically instead of stuffing large blobs into state.

API details: Flow, @start, @listen, @router, @persist.

Notes

CrewAI Flows FAQ