CrewAI flows let you wrap autonomous agent crews in explicit, code-defined pipelines while keeping the messy creative work inside the agents. This guide gives an ordered path to building reliable multi-agent systems with CrewAI flows, showing where structure helps and where it fights you.
Why Combine Flows and Agents
A bare Crew executes a set of tasks with delegated autonomy. That works for open-ended research or brainstorming, but production systems usually need guardrails: fetch input from a queue, run a research crew, validate the output, then hand off to a writer crew. CrewAI flows give you that skeleton without forcing you to hand-roll asyncio orchestration.
The core tradeoff is visibility versus flexibility. Flows make data movement explicit; agents keep the unstructured reasoning where it belongs. Used together, they reduce the “what just happened” problem that pure multi-agent loops create.
Setting Up Your First Flow
Install the latest CrewAI (0.30+ for stable flow decorators). A flow is a class inheriting from Flow with decorated methods.
Defining Start and Listen Steps
The @start() decorator marks the entrypoint. @listen() wires a method to a previous step’s return value.
from crewai import Flow, start, listen
class SimpleFlow(Flow):
@start()
def get_query(self):
return "Compare inference gateways"
@listen(get_query)
def log_query(self, query: str) -> str:
# deterministic step, no LLM call
print(f"Received: {query}")
return query.upper()Run it with SimpleFlow().kickoff(). The return of log_query is the final output. This is pure structure—no agents yet.
Adding a Crew as a Step
Drop a Crew inside a listen step when you need autonomy. Keep the crew scoped to one job.
from crewai import Agent, Task, Crew
class ResearchFlow(Flow):
@start()
def seed(self):
return "LLM routing layers"
@listen(seed)
def run_crew(self, topic: str) -> str:
researcher = Agent(
role="Analyst",
goal=f"Summarize current thinking on {topic}",
backstory="Senior systems engineer",
verbose=False,
)
task = Task(
description=f"Write a tight brief on {topic}",
agent=researcher,
expected_output="3 paragraph brief",
)
crew = Crew(agents=[researcher], tasks=[task])
result = crew.kickoff()
return result.rawThe crew’s kickoff() blocks until completion. Treat the returned string as a opaque blob unless you define output_json or output_pydantic on the task.
Conditional Routing with @router
Not every branch should run the same next step. Use @router to return a string that selects a listener.
from crewai import router
class GatedFlow(Flow):
@start()
def seed(self):
return "draft"
@listen(seed)
def generate(self, topic):
# pretend we call a crew
return "DRAFT TEXT"
@router(generate)
def check_quality(self, text):
if len(text) < 20:
return "rewrite"
return "publish"
@listen("rewrite")
def rewrite(self, text):
return text + " (expanded)"
@listen("publish")
def publish(self, text):
return textRouter methods must return a string matching a @listen tag. If no listener matches, the flow raises at runtime—fail loud, not silent.
Managing State and Passing Context
For flows longer than three steps, pass shared data through self.state instead of threading return values. State is a pydantic model attached to the flow instance.
from pydantic import BaseModel
class FlowState(BaseModel):
topic: str = ""
attempts: int = 0
class StatefulFlow(Flow):
state: FlowState = FlowState()
@start()
def set_topic(self):
self.state.topic = "edge caching for LLMs"
return self.state.topic
@listen(set_topic)
def increment_and_run(self, topic):
self.state.attempts += 1
# crew call omitted
return f"attempt {self.state.attempts}"Use state when a later step needs data from two earlier steps. Otherwise, return values are cleaner.
Model Configuration and Reliability
Each Agent takes an llm argument. In a flow that calls several crews with different model tiers, centralizing the endpoint cuts config sprawl. Pointing agents at an OpenAI-compatible gateway like n4n.ai gives you one endpoint for 240+ models and automatic fallback when a provider is degraded, which stabilizes CrewAI flows that mix cheap and premium models in one pipeline.
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="gpt-4o-mini",
base_url="https://api.n4n.ai/v1",
api_key="YOUR_KEY",
)
agent = Agent(role="x", goal="y", backstory="z", llm=llm)If you self-host multiple providers, set base_url per agent and keep a config map. Do not hardcode keys in flow files—inject from environment.
Common Pitfalls and Tradeoffs
Over-structuring Autonomous Work
Engineers new to CrewAI flows often model every agent turn as a flow step. That kills autonomy. A crew should own its internal task loop; the flow should only bracket it. If you find yourself piping agent A’s output token-by-token into step B, you’ve collapsed the abstraction.
Hidden Coupling Between Steps
Listen decorators create implicit ordering. Rename a method and the string-based router breaks. Keep a single flows.py module per pipeline and grep for the string tags after refactors. Typed state reduces this but doesn’t eliminate router strings.
Error Handling and Retries
Flows do not auto-retry failed crews. Wrap crew calls in try/except and route to a fallback step.
@listen(seed)
def safe_crew(self, topic):
try:
return run_my_crew(topic)
except Exception as e:
self.state.last_error = str(e)
return "ERROR"Then router on safe_crew can send "ERROR" to a notification step. Without this, a provider 429 takes down the whole flow.
A Complete Example
Below is a compact but realistic flow: seed topic, research crew, quality gate, conditional expand, final report.
from crewai import Flow, start, listen, router, Agent, Task, Crew
class Pipeline(Flow):
@start()
def seed(self):
return "vector search tradeoffs"
@listen(seed)
def research(self, topic):
a = Agent(role="Researcher", goal=f"Research {topic}", backstory="DB engineer")
t = Task(description=f"Brief on {topic}", agent=a, expected_output="Brief")
return Crew(agents=[a], tasks=[t]).kickoff().raw
@router(research)
def gate(self, text):
return "ok" if len(text) > 50 else "short"
@listen("short")
def expand(self, text):
a = Agent(role="Editor", goal="Expand text", backstory="Writer")
t = Task(description=f"Expand: {text}", agent=a, expected_output="Expanded")
return Crew(agents=[a], tasks=[t]).kickoff().raw
@listen("ok")
def keep(self, text):
return text
@listen(expand)
@listen(keep)
def report(self, text):
return {"final": text}
if __name__ == "__main__":
print(Pipeline().kickoff())The report step listens to both branches, merging them. This pattern—multiple listeners on one method—is the cleanest way to rejoin conditional paths.
When to Avoid CrewAI Flows
If your problem is fully open-ended—“monitor this stream and react when something interesting happens”—a flow’s start/end shape adds friction. Use a long-running crew or a custom loop. Flows shine when you have a known sequence with one or two autonomous islands.
Pick flows when you need audit trails, deterministic handoffs, or human-in-the-loop checkpoints. Skip them when the only structure is “agent decides everything.” Combining structure with autonomy is a design choice, not a default; CrewAI flows just make that choice cheap to express.