flows

CrewAI Flows Guide

Overview

Flows provide event-driven orchestration with precise control over execution paths, state management, and conditional branching. Use Flows when you need more control than Crews provide.

When to Use Flows vs Crews

| Scenario | Use Crews | Use Flows | |----------|-----------|-----------| | Simple multi-agent collaboration | ✅ | | | Sequential/hierarchical tasks | ✅ | | | Conditional branching | | ✅ | | Complex state management | | ✅ | | Event-driven workflows | | ✅ | | Hybrid (Crews inside Flow steps) | | ✅ |

Flow Basics

Creating a Flow

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

# Define state model
class MyState(BaseModel):
    counter: int = 0
    data: str = ""
    results: list = []

# Create flow with typed state
class MyFlow(Flow[MyState]):

    @start()
    def initialize(self):
        """Entry point - runs first"""
        self.state.counter = 1
        return {"initialized": True}

    @listen(initialize)
    def process(self, data):
        """Runs after initialize completes"""
        self.state.counter += 1
        return f"Processed: {data}"

# Run flow
flow = MyFlow()
result = flow.kickoff()
print(flow.state.counter)  # Access final state

Flow Decorators

@start() - Entry Point

@start()
def begin(self):
    """First method(s) to execute"""
    return {"status": "started"}

# Multiple start points (run in parallel)
@start()
def start_a(self):
    return "A"

@start()
def start_b(self):
    return "B"

@listen() - Event Trigger

# Listen to single method
@listen(initialize)
def after_init(self, result):
    """Runs when initialize completes"""
    return process(result)

# Listen to string name
@listen("high_confidence")
def handle_high(self):
    """Runs when router returns 'high_confidence'"""
    pass

@router() - Conditional Branching

@router(analyze)
def decide_path(self):
    """Returns string to route to specific listener"""
    if self.state.confidence > 0.8:
        return "high_confidence"
    elif self.state.confidence > 0.5:
        return "medium_confidence"
    return "low_confidence"

@listen("high_confidence")
def handle_high(self):
    pass

@listen("medium_confidence")
def handle_medium(self):
    pass

@listen("low_confidence")
def handle_low(self):
    pass

or_() and and_() - Conditional Combinations

from crewai.flow.flow import or_, and_

# Triggers when EITHER condition is met
@listen(or_("success", "partial_success"))
def handle_any_success(self):
    pass

# Triggers when BOTH conditions are met
@listen(and_(task_a, task_b))
def after_both_complete(self):
    pass

State Management

Pydantic State Model

from pydantic import BaseModel, Field
from typing import Optional

class WorkflowState(BaseModel):
    # Required fields
    input_data: str

    # Optional with defaults
    processed: bool = False
    confidence: float = 0.0
    results: list = Field(default_factory=list)
    error: Optional[str] = None

    # Nested models
    metadata: dict = Field(default_factory=dict)

class MyFlow(Flow[WorkflowState]):
    @start()
    def init(self):
        # Access state
        print(self.state.input_data)

        # Modify state
        self.state.processed = True
        self.state.results.append("item")
        self.state.metadata["timestamp"] = "2025-01-01"

State Initialization

# Initialize with inputs
flow = MyFlow()
result = flow.kickoff(inputs={"input_data": "my data"})

# Or set state before kickoff
flow.state.input_data = "my data"
result = flow.kickoff()

Integrating Crews in Flows

Crew as Flow Step

from crewai import Crew, Agent, Task, Process
from crewai.flow.flow import Flow, listen, start

class ResearchFlow(Flow[ResearchState]):

    @start()
    def gather_requirements(self):
        return {"topic": self.state.topic}

    @listen(gather_requirements)
    def run_research_crew(self, requirements):
        # Define crew
        researcher = Agent(
            role="Researcher",
            goal="Research {topic}",
            backstory="Expert researcher"
        )

        research_task = Task(
            description="Research {topic} thoroughly",
            expected_output="Detailed findings",
            agent=researcher
        )

        crew = Crew(
            agents=[researcher],
            tasks=[research_task],
            process=Process.sequential
        )

        # Execute crew within flow
        result = crew.kickoff(inputs=requirements)
        self.state.research_output = result.raw
        return result

    @listen(run_research_crew)
    def process_results(self, crew_result):
        # Process crew output
        return {"summary": self.state.research_output[:500]}

Multiple Crews in Flow

class MultiCrewFlow(Flow[MultiState]):

    @start()
    def init(self):
        return {"ready": True}

    @listen(init)
    def research_phase(self, data):
        return research_crew.kickoff(inputs={"topic": self.state.topic})

    @listen(research_phase)
    def writing_phase(self, research):
        return writing_crew.kickoff(inputs={"research": research.raw})

    @listen(writing_phase)
    def review_phase(self, draft):
        return review_crew.kickoff(inputs={"draft": draft.raw})

Complex Flow Patterns

Parallel Execution

class ParallelFlow(Flow[ParallelState]):

    @start()
    def init(self):
        return {"ready": True}

    # These run in parallel after init
    @listen(init)
    def branch_a(self, data):
        return crew_a.kickoff()

    @listen(init)
    def branch_b(self, data):
        return crew_b.kickoff()

    @listen(init)
    def branch_c(self, data):
        return crew_c.kickoff()

    # Waits for all branches
    @listen(and_(branch_a, branch_b, branch_c))
    def merge_results(self):
        return