CrewAI organizes multiple agents into a crew that works through a shared task list, and a Flow wraps that crew in deterministic Python control flow. This guide builds a content crew: a researcher agent gathers points, a writer agent drafts a post, and an editor agent reviews it, with a Flow step that reruns the writer if the editor rejects the draft. Stack: Python, crewai. Budget 45-60 minutes.
| Fact | Value |
|---|---|
| Tool | CrewAI |
| Difficulty | Intermediate |
| Time required | 45-60 min |
| Compatibility | Python 3.10-3.13, crewai, an LLM API key (Anthropic or OpenAI), any OS |
| Last updated | November 2026 |
What you'll build
You'll build a three-agent content crew for a small marketing team. A researcher agent gathers talking points on a topic, a writer agent turns them into a short blog post, and an editor agent checks the draft against a style guideline and either approves it or sends back specific feedback. A Flow wraps all of this: if the editor rejects the draft, the Flow reruns the writer with the editor's feedback instead of giving up after one pass.
In CrewAI, Agent instances carry role, goal, and backstory, which shape how each one reasons and writes, separately from Task instances that describe the actual work and which agent owns it. A Crew bundles a list of agents and tasks with a Process (sequential here, so tasks run in the order you listed them) and kickoff(inputs=...) runs the whole thing, returning a CrewOutput with .raw text and .tasks_output per task. A Flow sits one level up: its @start() and @listen() methods are plain Python, so the retry-on-rejection logic lives in code you can read and test, not in another layer of agent reasoning.
Choose CrewAI when the work naturally splits into named roles handing off a single artifact (research into draft into edit), and you want that handoff and retry logic to be explicit Python rather than one agent deciding everything. If your task is really one agent calling tools in a loop, LangChain's create_agent is simpler; if you need an explicit stateful graph with pausable human review, LangGraph is the better fit.
Prerequisites
- Python 3.10-3.13 and a terminal
- An Anthropic or OpenAI API key for the LLM calls
- pip install crewai
- Basic comfort with Python classes and decorators
Step-by-step guide
Install CrewAI and set your API key
Install the crewai package and export the key for whichever model provider you're using; CrewAI reads standard provider environment variables like ANTHROPIC_API_KEY. No separate account or dashboard is required to run a crew locally.
1pip install crewai2export ANTHROPIC_API_KEY="sk-ant-..."Expected result: pip install finishes with no errors and `python -c "import crewai"` runs silently.
Define the three agents
Each Agent gets a role (what it's called and does), a goal (what it's trying to achieve), and a backstory (context that shapes its voice and judgment calls). verbose=True prints each agent's reasoning while you're developing; turn it off once the crew is stable.
1from crewai import Agent23researcher = Agent(4 role="Content Researcher",5 goal="Gather accurate, specific talking points on the given topic",6 backstory="You research technical topics for a developer-focused blog. You prefer concrete numbers and real examples over vague claims.",7 verbose=True,8)9writer = Agent(10 role="Blog Writer",11 goal="Turn research into a clear, engaging short blog post",12 backstory="You write for busy developers. Short paragraphs, no fluff, one clear takeaway per section.",13 verbose=True,14)15editor = Agent(16 role="Style Editor",17 goal="Catch anything that violates the style guide and either approve the draft or explain exactly what to fix",18 backstory="You enforce a strict no-jargon, no-hype style guide. You reject vague superlatives like 'revolutionary' or 'game-changing'.",19 verbose=True,20)Expected result: Three Agent objects, each with a distinct role and backstory that will shape how it responds.
Define the tasks and chain them with context
Each Task needs a description, an expected_output describing the shape of a good result, and the agent that owns it. write_task's context=[research_task] feeds the researcher's output into the writer's prompt automatically; edit_task's context does the same with the writer's draft. {topic} and {style_notes} are filled in later from kickoff's inputs.
1from crewai import Task23research_task = Task(4 description="Research {topic} and list 5 specific, concrete talking points with real numbers where possible.",5 expected_output="A bullet list of 5 talking points.",6 agent=researcher,7)8write_task = Task(9 description="Write a 300-400 word blog post on {topic} using the research. Follow this style guide: {style_notes}",10 expected_output="A complete short blog post in markdown.",11 agent=writer,12 context=[research_task],13)14edit_task = Task(15 description="Review the draft against this style guide: {style_notes}. Respond with APPROVED or REJECTED: <specific feedback>.",16 expected_output="Either 'APPROVED' or 'REJECTED: ' followed by specific, actionable feedback.",17 agent=editor,18 context=[write_task],19)Expected result: Three Task objects where write_task and edit_task will automatically see the prior task's output in their prompt.
Assemble the Crew
Crew bundles the agents and tasks with a Process. Process.sequential runs research_task, then write_task, then edit_task, in that order, which matches the context=[...] chain you just built. kickoff(inputs=...) fills in {topic} and {style_notes} and runs the whole sequence, returning a CrewOutput.
1from crewai import Crew, Process23content_crew = Crew(4 agents=[researcher, writer, editor],5 tasks=[research_task, write_task, edit_task],6 process=Process.sequential,7)89result = content_crew.kickoff(inputs={10 "topic": "why prompt caching cuts agent costs",11 "style_notes": "no jargon, no hype words, concrete numbers only",12})13print(result.raw)Expected result: A full research -> write -> edit run printing either 'APPROVED' or 'REJECTED: ...' as the final task's output.
Wrap the crew in a Flow that retries on rejection
A Flow's @start() method kicks things off; @listen(step) methods run after step finishes, receiving its return value. This Flow runs the crew, checks whether the editor's output starts with REJECTED, and if so builds a second crew that reruns just write_task and edit_task with the feedback folded into the style notes, capping it at one retry so a stubborn editor doesn't loop forever.
1from crewai.flow import Flow, listen, start2from pydantic import BaseModel34class FlowState(BaseModel):5 topic: str = ""6 style_notes: str = ""7 result: str = ""8 retried: bool = False910class ContentFlow(Flow[FlowState]):11 @start()12 def run_crew(self):13 output = content_crew.kickoff(inputs={"topic": self.state.topic, "style_notes": self.state.style_notes})14 self.state.result = output.raw15 return output.raw1617 @listen(run_crew)18 def check_and_retry(self, editor_output: str):19 if editor_output.strip().upper().startswith("APPROVED") or self.state.retried:20 self.state.result = editor_output21 return22 self.state.retried = True23 feedback = editor_output.split(":", 1)[-1].strip()24 notes = f"{self.state.style_notes}. Also fix: {feedback}"25 output = content_crew.kickoff(inputs={"topic": self.state.topic, "style_notes": notes})26 self.state.result = output.rawExpected result: A flow that automatically reruns the writer and editor once on rejection, then stops regardless of the second outcome.
Run the flow and inspect the final state
kickoff() on the Flow instance runs run_crew then check_and_retry via the @listen chain. Read the final result from flow.state.result rather than the return value of kickoff(), since the Flow's own state is what check_and_retry actually updated.
1flow = ContentFlow()2flow.kickoff(inputs={3 "topic": "why prompt caching cuts agent costs",4 "style_notes": "no jargon, no hype words, concrete numbers only",5})6print(flow.state.result)Expected result: A final approved (or once-retried) blog post draft printed from flow.state.result.
Complete working example
1"""Content crew: researcher -> writer -> editor, wrapped in a Flow that retries once on rejection.2Run: python content_crew_flow.py (pip install crewai; set ANTHROPIC_API_KEY)3"""4from crewai import Agent, Crew, Process, Task5from crewai.flow import Flow, listen, start6from pydantic import BaseModel78researcher = Agent(9 role="Content Researcher",10 goal="Gather accurate, specific talking points on the given topic",11 backstory="You research technical topics for a developer-focused blog. You prefer concrete numbers and real examples over vague claims.",12 verbose=True,13)14writer = Agent(15 role="Blog Writer",16 goal="Turn research into a clear, engaging short blog post",17 backstory="You write for busy developers. Short paragraphs, no fluff, one clear takeaway per section.",18 verbose=True,19)20editor = Agent(21 role="Style Editor",22 goal="Catch anything that violates the style guide and either approve the draft or explain exactly what to fix",23 backstory="You enforce a strict no-jargon, no-hype style guide. You reject vague superlatives like 'revolutionary' or 'game-changing'.",24 verbose=True,25)2627research_task = Task(28 description="Research {topic} and list 5 specific, concrete talking points with real numbers where possible.",29 expected_output="A bullet list of 5 talking points.",30 agent=researcher,31)32write_task = Task(33 description="Write a 300-400 word blog post on {topic} using the research. Follow this style guide: {style_notes}",34 expected_output="A complete short blog post in markdown.",35 agent=writer,36 context=[research_task],37)38edit_task = Task(39 description="Review the draft against this style guide: {style_notes}. Respond with APPROVED or REJECTED: <specific feedback>.",40 expected_output="Either 'APPROVED' or 'REJECTED: ' followed by specific, actionable feedback.",41 agent=editor,42 context=[write_task],43)4445content_crew = Crew(46 agents=[researcher, writer, editor],47 tasks=[research_task, write_task, edit_task],48 process=Process.sequential,49)505152class FlowState(BaseModel):53 topic: str = ""54 style_notes: str = ""55 result: str = ""56 retried: bool = False575859class ContentFlow(Flow[FlowState]):60 @start()61 def run_crew(self):62 output = content_crew.kickoff(inputs={"topic": self.state.topic, "style_notes": self.state.style_notes})63 self.state.result = output.raw64 return output.raw6566 @listen(run_crew)67 def check_and_retry(self, editor_output: str):68 if editor_output.strip().upper().startswith("APPROVED") or self.state.retried:69 self.state.result = editor_output70 return71 self.state.retried = True72 feedback = editor_output.split(":", 1)[-1].strip()73 notes = f"{self.state.style_notes}. Also fix: {feedback}"74 output = content_crew.kickoff(inputs={"topic": self.state.topic, "style_notes": notes})75 self.state.result = output.raw767778if __name__ == "__main__":79 flow = ContentFlow()80 flow.kickoff(inputs={81 "topic": "why prompt caching cuts agent costs",82 "style_notes": "no jargon, no hype words, concrete numbers only",83 })84 print(flow.state.result)85Common mistakes
Why it's a problem: Writing one agent's backstory to cover every role in the crew
How to avoid: A shared backstory blurs each agent's judgment. Give researcher, writer, and editor distinct backstories that match what each should actually optimize for.
Why it's a problem: Forgetting context=[previous_task] on a task that needs the prior task's output
How to avoid: Without it, write_task has no idea what research_task found and will invent its own research instead of using it.
Why it's a problem: Letting the Flow retry unconditionally on any rejection
How to avoid: Cap retries in Flow state (like this guide's self.state.retried) so a stubborn editor-writer disagreement can't burn unlimited tokens.
Why it's a problem: Reading kickoff()'s return value instead of flow.state after running a Flow
How to avoid: A @listen method's return value isn't automatically the Flow's final output. Store what you need in self.state and read it from there.
Why it's a problem: Using Process.hierarchical when Process.sequential would do
How to avoid: Hierarchical process adds a manager agent making delegation decisions, which costs more tokens and adds nondeterminism. Reach for it only when task order genuinely can't be fixed in advance.
Best practices
- Keep each agent's role, goal, and backstory focused on one job; split a role in two rather than overloading one agent with unrelated responsibilities.
- Chain tasks with context=[...] instead of repeating the same information across multiple task descriptions.
- Put retry and branching logic in a Flow's plain Python methods, not in agent instructions, so it's testable and predictable.
- Read CrewOutput.tasks_output when you need an earlier task's result specifically, not just the crew's final .raw.
- Set expected_output specifically enough that an agent can self-check against it, like this guide's APPROVED/REJECTED format.
- Turn off verbose=True once a crew is stable; it's useful for debugging but adds noise and token overhead to every run.
Still stuck?
Copy one of these prompts to get a personalized, step-by-step explanation.
I have a CrewAI content crew (researcher, writer, editor agents chained with context=[...]) wrapped in a Flow that retries once on rejection (code below). Add a fourth agent, seo_checker, whose task runs after edit_task approves and adds a meta description, and chain it into the same Crew. [paste complete_code]
In Cursor, open content_crew_flow.py and ask: 'Add a @router method to ContentFlow that branches to a human_escalation step instead of a second retry when the editor rejects the draft twice.'
Frequently asked questions
What's the difference between a Crew and a Flow in CrewAI?
A Crew is a team of agents working through a task list via a Process (sequential or hierarchical); the agents' own reasoning drives most of the decisions. A Flow is plain Python with @start/@listen/@router methods that can run one or more crews as steps, giving you deterministic control over branching and retries around the agent reasoning.
Can agents in a crew talk to each other directly?
Not directly between arbitrary agents. Context flows through context=[...] on tasks (passing one task's output into another's prompt) or through allow_delegation=True on an agent, which lets it ask another agent a question mid-task in a hierarchical process.
How do I get a structured object out of a crew instead of raw text?
Set output_pydantic on a Task to a Pydantic model describing the shape you want; the task's result will include a populated instance of that model accessible via CrewOutput.pydantic (or .tasks_output[i].pydantic for a specific task).
Does CrewAI work with models other than OpenAI?
Yes. CrewAI agents accept any LiteLLM-supported model string; Anthropic, Gemini, and local Ollama models all work by setting llm= on the Agent or crewai's model-selection environment variables.
How many retries is reasonable before escalating to a human?
One or two is typical for a content-review loop like this guide's; beyond that, two AI agents disagreeing usually means the instructions are ambiguous, not that one more retry will fix it. Escalate to a human reviewer instead of looping further.
Can RapidDev build a CrewAI pipeline like this for our content team?
Yes. RapidDev builds multi-agent content, research, and ops crews on CrewAI (and the Flow-level retry and escalation logic most tutorials skip). A free 30-minute consultation is the fastest way to scope yours.
Talk to an Expert
Our team has built 1,000+ apps. Get personalized help with your project.
Book a free consultation