CrewAI Flows: The Structured Way to Build Multi-Step AI Pipelines

Crews vs Flows: What Is the Difference?

A Crew is good at open-ended, role-based collaboration where agents decide how to divide and complete work. A Flow is good at structured pipelines where you need deterministic control over execution order, branching, and state.

	Crew	Flow
Execution model	Agents collaborate, tasks assigned by process mode	Explicit steps connected by decorators
Control flow	Emergent from agent decisions	Deterministic — you define the DAG
State management	Via task context	Typed pydantic state shared across all steps
Best for	Research, writing, analysis with human-like collaboration	ETL pipelines, approval chains, multi-stage generation
Mix with Python?	Limited	Yes — any step can be pure Python

The key insight: Flows let you call Crews as steps within a larger structured pipeline. You get the best of both — deterministic orchestration with agentic execution inside each step.

Your First Flow

from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel
 
class ContentState(BaseModel):
    topic: str = ""
    outline: str = ""
    draft: str = ""
    final: str = ""
 
class ContentFlow(Flow[ContentState]):
 
    @start()
    def set_topic(self):
        """Entry point — runs first."""
        self.state.topic = "The impact of AI on software engineering"
        print(f"Topic set: {self.state.topic}")
 
    @listen(set_topic)
    def generate_outline(self):
        """Runs after set_topic completes."""
        # In a real flow: call an LLM or a Crew here
        self.state.outline = "1. Introduction\n2. Code generation\n3. Testing\n4. Conclusions"
        print("Outline ready")
 
    @listen(generate_outline)
    def write_draft(self):
        """Runs after outline is ready."""
        self.state.draft = f"Draft article about: {self.state.topic}\n{self.state.outline}"
        print("Draft written")
 
    @listen(write_draft)
    def finalise(self):
        self.state.final = self.state.draft + "\n\n[Reviewed and approved]"
        return self.state.final
 
flow = ContentFlow()
result = flow.kickoff()
print(result)

Calling a Crew Inside a Flow Step

This is the most powerful pattern — use a Crew for the complex, agentic parts and Flow for the structure around them:

from crewai import Agent, Task, Crew
from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel
from langchain_anthropic import ChatAnthropic
 
llm = ChatAnthropic(model="claude-haiku-4-5-20251001")
 
class ResearchState(BaseModel):
    query: str = ""
    research_findings: str = ""
    summary: str = ""
 
class ResearchFlow(Flow[ResearchState]):
 
    @start()
    def intake(self):
        self.state.query = "Latest developments in AI agent memory systems"
 
    @listen(intake)
    def run_research_crew(self):
        """Hand off to a Crew for the research step."""
        researcher = Agent(
            role="Research Specialist",
            goal="Find accurate, detailed information on the given topic",
            backstory="Expert at synthesising technical research",
            llm=llm, verbose=False
        )
        task = Task(
            description=f"Research this topic thoroughly: {self.state.query}",
            agent=researcher,
            expected_output="Detailed research findings in bullet points"
        )
        crew = Crew(agents=[researcher], tasks=[task])
        self.state.research_findings = str(crew.kickoff())
 
    @listen(run_research_crew)
    def summarise(self):
        """A simple Python step — no agent needed."""
        lines = self.state.research_findings.split("\n")
        self.state.summary = "\n".join(lines[:10])  # first 10 lines
        return self.state.summary
 
flow = ResearchFlow()
result = flow.kickoff()

Conditional Routing with @router

Use @router to branch the flow based on state — different paths for different conditions:

from crewai.flow.flow import Flow, listen, start, router
from pydantic import BaseModel
 
class ReviewState(BaseModel):
    content: str = ""
    word_count: int = 0
    quality_score: float = 0.0
 
class QualityFlow(Flow[ReviewState]):
 
    @start()
    def generate_content(self):
        self.state.content = "This is a sample article..." * 50
        self.state.word_count = len(self.state.content.split())
 
    @listen(generate_content)
    def score_content(self):
        # Simulate quality scoring (in production: call an LLM evaluator)
        self.state.quality_score = 0.85 if self.state.word_count > 100 else 0.4
 
    @router(score_content)
    def route_by_quality(self):
        """Return the name of the next method to run."""
        if self.state.quality_score >= 0.7:
            return "publish"
        else:
            return "revise"
 
    @listen("publish")
    def publish(self):
        print(f"Publishing content (score: {self.state.quality_score:.0%})")
        return "published"
 
    @listen("revise")
    def revise(self):
        print(f"Content needs revision (score: {self.state.quality_score:.0%})")
        self.state.content += " [REVISED]"
        return "revised"
 
flow = QualityFlow()
result = flow.kickoff()

The string returned by a @router method must exactly match the string passed to @listen() on the branch methods. A typo here causes the flow to silently skip that branch.

Parallel Steps with @listen on Multiple Events

When a step should only run after multiple predecessors complete, list them in @listen:

from crewai.flow.flow import Flow, listen, start, and_
from pydantic import BaseModel
 
class ParallelState(BaseModel):
    seo_keywords: list = []
    market_research: str = ""
    final_brief: str = ""
 
class ContentBriefFlow(Flow[ParallelState]):
 
    @start()
    def fetch_seo_keywords(self):
        self.state.seo_keywords = ["AI agents", "LLM frameworks", "automation"]
 
    @start()
    def run_market_research(self):
        self.state.market_research = "Target audience: developers building AI apps"
 
    # Only runs after BOTH start steps complete
    @listen(and_(fetch_seo_keywords, run_market_research))
    def compile_brief(self):
        self.state.final_brief = (
            f"Keywords: {self.state.seo_keywords}\n"
            f"Research: {self.state.market_research}"
        )
        return self.state.final_brief
 
flow = ContentBriefFlow()
result = flow.kickoff()  # fetch_seo_keywords and run_market_research run in parallel

Passing Initial State to kickoff()

flow = ContentFlow()
 
# Pass initial state values at kickoff time
result = flow.kickoff(inputs={
    "topic": "Building production-ready RAG systems"
})

When to Use Flows vs Crews

Use a Flow when you need deterministic execution order — step 1 must always precede step 2.
Use a Flow when you need to branch based on outputs — quality gates, approval routing, conditional processing.
Use a Flow when you want to mix LLM steps with regular Python logic without forcing everything through an agent.
Use a Crew alone when the task is open-ended and benefits from agent collaboration without rigid sequencing.
Use both together (Crew inside a Flow step) for complex pipelines that need both structure and agentic flexibility.