TopolearnPicking the right structure early saves rewrites when an app moves from prototype to production.
Inside a flow step you can call a crew like any other code:
@listen(plan)
def write(self):
result = WritingCrew().crew().kickoff(inputs={"topic": self.state.topic})
self.state.draft = result.rawcrewai create flow NAME scaffolds a flow project: main.py holds the flow, and crews/ contains one folder per crew. crewai run detects that the project is a flow and runs it. A JSON-first crew can be loaded inside a step with load_crew(path).
When to use what (following CrewAI's own guidance):
- Low complexity, low precision (brainstorming, simple drafts): one simple crew.
- Low complexity, high precision (extraction, classification): a flow with direct LLM or single-agent calls and structured outputs.
- High complexity, low precision (research reports, content pipelines): a crew with several specialised agents.
- High complexity, high precision (regulated or mission-critical work): a flow that orchestrates crews with validation steps between them.
Crews are quicker to prototype. Flows are easier to test, debug and maintain because the control flow is ordinary code. Start with the simplest design that works and add structure as requirements become clear.
A common production shape: a flow validates input, calls a research crew, checks the result with code or a guardrail, routes to a human when needed, and saves the outcome. Each crew stays small and focused.
Example
from pydantic import BaseModel
from crewai import Agent, Crew, Task
from crewai.flow import Flow, listen, start
def writing_crew() -> Crew:
writer = Agent(role="Blog writer", goal="Write clear posts", backstory="Writes for engineers.")
post = Task(description="Write a 150-word post about {topic}.", expected_output="Markdown.", agent=writer)
return Crew(agents=[writer], tasks=[post])
class PostState(BaseModel):
topic: str = ""
draft: str = ""
class PostFlow(Flow[PostState]):
@start()
def validate(self):
if not self.state.topic.strip():
raise ValueError("topic is required")
@listen(validate)
def write(self):
self.state.draft = writing_crew().kickoff(inputs={"topic": self.state.topic}).raw
@listen(write)
def publish(self): return f"{len(self.state.draft.split())} words ready"
print(PostFlow().kickoff(inputs={"topic": "Python virtual environments"}))Remember
- Flow steps call
crew.kickoff(...)and store results in state crewai create flow NAMEscaffolds main.py plus crews/- Crews for open-ended work; flows for control, validation and precision
Watch out for
- Building one huge crew with many agents for a process that is mostly fixed steps; a flow with small crews is cheaper and easier to debug.