CrewAI Flows – Orchestrating End‑to‑End Workflows

CrewAI Flows are the orchestration layer that controls when and how your code, LLM calls, agents, and crews run together in a structured, event‑driven workflow. Where Crews are “teams of agents,” Flows are the workflow engine that chains steps, manages state, and routes execution.

1. What Flows Are (and Why They Exist)

A Flow is a Python class that defines an AI workflow as connected methods, decorated with @start, @listen, and optional routing / control decorators.
Flows let you mix: raw code, direct LLM calls, Agents, and full Crews, all under one event‑driven control structure.
They add “software‑engineering discipline” on top of multi‑agent systems: clear steps, explicit data passing, and testable logic.

Key Flow capabilities:

Simplified workflow creation (chain many steps and crews)
State management across steps
Event‑driven architecture (steps react to prior outputs)
Flexible control flow (branching, fan‑out, fan‑in, human gates)

2. Core Building Blocks: Flow, @start, @listen

2.1 Flow class

You define a Flow by subclassing Flow:

# python
from crewai.flow.flow import Flow, listen, start
from litellm import completion

class ExampleFlow(Flow):
    model = "gpt-4o-mini"

    @start()
    def generate_city(self):
        print("Starting flow")
        print(f"Flow State ID: {self.state['id']}")

        response = completion(
            model=self.model,
            messages=[{"role": "user", "content": "Return the name of a random city in the world."}],
        )

        random_city = response["choices"]["message"]["content"]
        self.state["city"] = random_city
        print(f"Random City: {random_city}")
        return random_city

    @listen(generate_city)
    def generate_fun_fact(self, random_city):
        response = completion(
            model=self.model,
            messages=[{"role": "user", "content": f"Tell me a fun fact about {random_city}"}],
        )

        fun_fact = response["choices"]["message"]["content"]
        self.state["fun_fact"] = fun_fact
        return fun_fact

flow = ExampleFlow()
flow.plot()
result = flow.kickoff()
print(f"Generated fun fact: {result}")

What’s happening:

@start() marks entry methods that fire when the Flow begins.
@listen(generate_city) runs when generate_city finishes and receives its output as an argument.
self.state carries shared data (city , fun_fact) across steps.

2.2 @start()

Defines entry points into your Flow.
You can have:
- Simple start: @start()
- Start gated on an event/label: @start("method_or_label")
- Start with a callable condition to control when it fires.

Multiple @start() methods can run in parallel when the Flow begins or resumes.

2.3 @listen()

@listen() makes a method react to the output of another step.
You can reference the source step by:
- Name: @listen("generate_city")
- Function: @listen(generate_city)

The listener’s parameters correspond to the upstream output, so the data flow is explicit and typed.

3. Flow Output and Running Flows

3.1 Final output

The final output of a Flow is the return value of the last method that completes.
flow.kickoff() returns that final value directly.

Example:

class OutputExampleFlow(Flow):
    @start()
    def first_method(self):
        return "Output from first_method"

    @listen(first_method)
    def second_method(self, first_output):
        return f"Second method received: {first_output}"

flow = OutputExampleFlow()
flow.plot("my_flow_plot")
final_output = flow.kickoff()
print("---- Final Output ----")
print(final_output)  # "Second method received: Output from first_method"

3.2 Running and streaming

You can run flows in several ways:

Programmatically:

flow = ExampleFlow()
result = flow.kickoff()

With streaming for real‑time updates:

class StreamingFlow(Flow):
    stream = True

    @start()
    def research(self):
        ...

flow = StreamingFlow()
streaming = flow.kickoff()

for chunk in streaming:
    print(chunk.content, end="", flush=True)

result = streaming.result  # final output

From the CLI (recommended):

crewai run           # auto-detects flow if type="flow" in pyproject.toml

# or (older style)
crewai flow kickoff

4. State Management in Flows

State is the memory of your Flow: it keeps context between steps and across restarts.

4.1 State lifecycle

The state follows a predictable lifecycle:

Initialization – created as an empty dict or Pydantic model.
Modification – methods read and update it.
Transmission – passed automatically across methods.
(Optional) Persistence – can be saved to storage and reloaded.
Completion – final state reflects all updates.

4.2 Unstructured state (dict)

self.state is a dictionary; you add keys dynamically.
Flow automatically adds a unique id (UUID) to track executions.

Example:

class UnstructuredExampleFlow(Flow):
    @start()
    def first_method(self):
        print(f"State ID: {self.state['id']}")
        self.state["counter"] = 0
        self.state["message"] = "Hello from unstructured flow"

    @listen(first_method)
    def second_method(self):
        self.state["counter"] += 1
        self.state["message"] += " - updated"

    @listen(second_method)
    def third_method(self):
        self.state["counter"] += 1
        self.state["message"] += " - updated again"
        print(f"State after third_method: {self.state}")

Use when you want maximum flexibility and are still exploring your schema.

4.3 Structured state (Pydantic model)

You define a BaseModel that represents state.
Flow[YourStateModel] enforces types and gives IDE auto‑completion.
An id field is automatically added and managed.

Example:

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

class ExampleState(BaseModel):
    counter: int = 0
    message: str = ""

class StateExampleFlow(Flow[ExampleState]):
    @start()
    def first_method(self):
        print(f"State ID: {self.state.id}")
        self.state.message = "Hello from structured flow"

    @listen(first_method)
    def second_method(self):
        self.state.counter += 1
        self.state.message += " - updated"
        return self.state.message

flow = StateExampleFlow()
flow.plot("my_flow_plot")
final_output = flow.kickoff()
print(f"Final Output: {final_output}")
print("Final State:", flow.state)

Use structured state when you want reliability, validation, and clear contracts.

5. Persisting Flow State (@persist)

Persistence lets flows survive restarts and long‑running processes.

@persist at class level: persist all flow method states (default backend: SQLite).
@persist at method level: persist only that method’s state (fine‑grained control).

Example – class‑level:

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

class MyState(BaseModel):
    counter: int = 0

@persist  # SQLiteFlowPersistence by default
class MyFlow(Flow[MyState]):
    @start()
    def initialize_flow(self):
        self.state.counter = 1
        print("Initialized flow. State ID:", self.state.id)

    @listen(initialize_flow)
    def next_step(self):
        self.state.counter += 1
        print("Flow state is persisted. Counter:", self.state.counter)

Key points:

Each state has a unique UUID that is preserved across saves/loads.
SQLite is default; you can plug in custom persistence backends.
Strong error handling and validation around save/load.

Use persistence for long‑running, resumable, or production workflows.

6. Flow Control: or_, and_, router, and Human Feedback

Flows support rich control‑flow primitives beyond simple chains.

6.1 or_ – listen to any of multiple sources

Triggers when any of the specified methods emit output.

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

class OrExampleFlow(Flow):
    @start()
    def start_method(self):
        return "Hello from the start method"

    @listen(start_method)
    def second_method(self):
        return "Hello from the second method"

    @listen(or_(start_method, second_method))
    def logger(self, result):
        print(f"Logger: {result}")

6.2 and_ – wait for all sources

Triggers only when all specified methods have emitted outputs.

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

class AndExampleFlow(Flow):
    @start()
    def start_method(self):
        self.state["greeting"] = "Hello from the start method"

    @listen(start_method)
    def second_method(self):
        self.state["joke"] = "What do computers eat? Microchips."

    @listen(and_(start_method, second_method))
    def logger(self):
        print("---- Logger ----")
        print(self.state)

6.3 @router – conditional routing

Routes based on method output (like a switch/case).

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

class ExampleState(BaseModel):
    success_flag: bool = False

class RouterFlow(Flow[ExampleState]):
    @start()
    def start_method(self):
        print("Starting the structured flow")
        self.state.success_flag = random.choice([True, False])

    @router(start_method)
    def second_method(self):
        return "success" if self.state.success_flag else "failed"

    @listen("success")
    def third_method(self):
        print("Third method running")

    @listen("failed")
    def fourth_method(self):
        print("Fourth method running")

6.4 @human_feedback – human‑in‑the‑loop gates

Requires CrewAI ≥ 1.8.0.

Pauses flow execution to collect human feedback.
Optionally maps free‑form feedback into discrete outcomes like "approved" / "rejected" using an LLM, then routes to matching listeners.

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

class ReviewFlow(Flow):
    @start()
    @human_feedback(
        message="Do you approve this content?",
        emit=["approved", "rejected", "needs_revision"],
        llm="gpt-4o-mini",
        default_outcome="needs_revision",
    )
    def generate_content(self):
        return "Content to be reviewed..."

    @listen("approved")
    def on_approval(self, result: HumanFeedbackResult):
        print(f"Approved! Feedback: {result.feedback}")

    @listen("rejected")
    def on_rejection(self, result: HumanFeedbackResult):
        print(f"Rejected. Reason: {result.feedback}")

You can also use it without routing just to capture comments and access them via self.last_human_feedback or self.human_feedback_history.

7. Integrating Agents and Crews into Flows

Flows can orchestrate Agents directly or full Crews, turning them into steps in a larger application.

7.1 Using Agents inside a Flow

Example: market research with structured output.

from typing import Any, Dict, List
from pydantic import BaseModel, Field
from crewai.agent import Agent
from crewai_tools import SerperDevTool
from crewai.flow.flow import Flow, listen, start

class MarketAnalysis(BaseModel):
    key_trends: List[str] = Field(description="List of identified market trends")
    market_size: str = Field(description="Estimated market size")
    competitors: List[str] = Field(description="Major competitors in the space")

class MarketResearchState(BaseModel):
    product: str = ""
    analysis: MarketAnalysis | None = None

class MarketResearchFlow(Flow[MarketResearchState]):
    @start()
    def initialize_research(self) -> Dict[str, Any]:
        print(f"Starting market research for {self.state.product}")
        return {"product": self.state.product}

    @listen(initialize_research)
    async def analyze_market(self) -> Dict[str, Any]:
        analyst = Agent(
            role="Market Research Analyst",
            goal=f"Analyze the market for {self.state.product}",
            backstory="You are an experienced market analyst...",
            tools=[SerperDevTool()],
            verbose=True,
        )

        query = f"""
        Research the market for {self.state.product}. Include:
        1. Key market trends
        2. Market size
        3. Major competitors
        Format your response according to the specified structure.
        """
        result = await analyst.kickoff_async(query, response_format=MarketAnalysis)

        return {"analysis": result.pydantic}

    @listen(analyze_market)
    def present_results(self, analysis) -> None:
        ...

Flows are the macro‑orchestrator,

Agents are micro‑problem solvers.

7.2 Using Crews inside a Flow

Flows can also call entire crews, using Flow state as inputs and capturing outputs.

Folder scaffold (from crewai create flow name_of_flow):

Path

name_of_flow/ Root flow project.

crews/poem_crew/ A sample crew with its own config.

crews/poem_crew/config/ agents.yaml, tasks.yaml.

crews/poem_crew/poem_crew.py Crew definition (agents, tasks, crew).

tools/ Custom tools.

main.py Main flow definition and entry point.

Example: connecting a PoemCrew:

from random import randint
from pydantic import BaseModel
from crewai.flow.flow import Flow, listen, start
from .crews.poem_crew.poem_crew import PoemCrew

class PoemState(BaseModel):
    sentence_count: int = 1
    poem: str = ""

class PoemFlow(Flow[PoemState]):
    @start()
    def generate_sentence_count(self):
        print("Generating sentence count")
        self.state.sentence_count = randint(1, 5)

    @listen(generate_sentence_count)
    def generate_poem(self):
        print("Generating poem")
        result = PoemCrew().crew().kickoff(
            inputs={"sentence_count": self.state.sentence_count}
        )
        print("Poem generated", result.raw)
        self.state.poem = result.raw

    @listen(generate_poem)
    def save_poem(self):
        print("Saving poem")
        with open("poem.txt", "w") as f:
            f.write(self.state.poem)

This is the signature pattern: Flows orchestrate multiple domain crews into one cohesive business process.

8. Building Your First Flow Project (CLI)

The First Flow guide shows a typical project setup.

Create a Flow project:

# shell commands

crewai create flow guide_creator_flow
cd guide_creator_flow

Add a specialized crew (e.g., content writer):

# shell commands

crewai flow add-crew content-crew

This generates a new crew folder with agents.yaml, tasks.yaml, and a crew Python file wired for use inside your flow.

Edit main.py to define your Flow and connect it to that crew (similar to the Poem example).

This scaffold naturally separates concerns:

Flow logic in main.py
Specialized crews in crews/
Tools in tools/

Here is a mental model: Flows = “orchestrator app”; Crews = “specialist teams”.

9. Visualizing Flows with plot()

Plots help you and your students see the workflow.

Call flow.plot("my_flow_plot") on a Flow instance to generate my_flow_plot.html with an interactive graph.
Or use:

# shell commands

crewai flow plot

in a structured project.

Plots show:

Nodes = methods (@start, @listen, routers, human_feedback stops)
Edges = data / control‑flow connections

10. When to Reach for Flows

Use Flows when:

They need multi‑step orchestration (not just a single crew run).
They want explicit state management across steps (and maybe persistence).
They must integrate multiple crews, agents, tools, and raw code into one system.
They need branching, fan‑out, human approval, or complex routing.

In Summary:

-> Crews turn LLMs into specialist teams. -> Flows turn those teams into a product‑grade application with memory, control‑flow, and observability.

CrewAI Flows – Orchestrating End‑to‑End Workflows

On this page