What You’ll Learn: Tasks are the work items you assign to your crew — they define what needs to be done, who does it, and what the expected output looks like. Processes control how tasks flow: sequentially (one after another) or hierarchically (a manager delegates). Think of tasks like tickets in a project management tool: each has a description, assignee, acceptance criteria, and dependencies.
1. Task Definition
Tasks are the atomic units of work in CrewAI. Each task has a description (what to do), an expected_output (success criteria), and an agent assignment (who does it). Tasks can also produce structured outputs, write to files, and pass context to downstream tasks.
Task Design Principle: Write
expected_output as if describing what a perfect deliverable looks like. The more specific you are about format, length, and content, the more consistent your results. Think of it as a rubric.
1.1 Tasks with Structured Output
from crewai import Agent, Task
from pydantic import BaseModel
from typing import List
# Define structured output schema
class ResearchReport(BaseModel):
title: str
summary: str
key_findings: List[str]
confidence_score: float
sources: List[str]
# Create agent
researcher = Agent(
role="Senior Research Analyst",
goal="Produce thorough research reports with cited sources",
backstory="You are meticulous about accuracy and structure."
)
# Task with Pydantic output
research_task = Task(
description="""Research the current state of quantum computing
in 2026. Focus on recent breakthroughs, commercial applications,
and remaining challenges.""",
expected_output="""A structured research report with title,
executive summary, 5-8 key findings, confidence score (0-1),
and source URLs.""",
agent=researcher,
output_pydantic=ResearchReport # Enforces schema
)
print(f"Task: {research_task.description[:50]}...")
print(f"Output type: {research_task.output_pydantic.__name__}")
1.2 Task Context — Passing Data Between Tasks
from crewai import Agent, Task
researcher = Agent(
role="Researcher",
goal="Find accurate information",
backstory="You are thorough and precise."
)
writer = Agent(
role="Writer",
goal="Write compelling content from research",
backstory="You transform data into engaging narratives."
)
# First task produces output
research_task = Task(
description="Research the latest trends in renewable energy.",
expected_output="A detailed brief with data points and sources.",
agent=researcher
)
# Second task receives context from first
writing_task = Task(
description="""Using the research provided, write a blog post
about renewable energy trends.""",
expected_output="A 1000-word blog post in markdown format.",
agent=writer,
context=[research_task], # Receives output from research_task
output_file="output/energy-blog.md"
)
print(f"Writing task context: {len(writing_task.context)} tasks")
print(f"Output file: {writing_task.output_file}")
2. Sequential Processes
Sequential is the simplest process type — tasks execute in order, each one receiving the output of the previous task as context. Best for linear workflows where each step builds on the last.
2.1 Sequential Crew with Data Flow
from crewai import Agent, Task, Crew, Process
# Define agents
analyst = Agent(
role="Market Analyst",
goal="Analyze market data and identify trends",
backstory="You are a quantitative analyst at a hedge fund."
)
strategist = Agent(
role="Investment Strategist",
goal="Develop investment strategies from analysis",
backstory="You translate market insights into actionable strategies."
)
editor = Agent(
role="Report Editor",
goal="Polish reports for executive audiences",
backstory="You make complex financial analysis accessible."
)
# Define sequential tasks
analysis_task = Task(
description="Analyze the current tech sector market conditions for {quarter}.",
expected_output="Market analysis with trends, risks, and opportunities.",
agent=analyst
)
strategy_task = Task(
description="Based on the market analysis, develop 3 investment strategies.",
expected_output="Three ranked strategies with risk/reward profiles.",
agent=strategist,
context=[analysis_task]
)
report_task = Task(
description="Edit and format the strategies into an executive briefing.",
expected_output="A polished 2-page executive briefing in markdown.",
agent=editor,
context=[strategy_task],
output_file="output/briefing.md"
)
# Assemble sequential crew
crew = Crew(
agents=[analyst, strategist, editor],
tasks=[analysis_task, strategy_task, report_task],
process=Process.sequential,
verbose=True
)
# Kickoff
result = crew.kickoff(inputs={"quarter": "Q2 2026"})
print(f"Final output length: {len(result.raw)} chars")
When to Use Sequential: Choose sequential when tasks have clear dependencies (A → B → C), when later tasks need the full output of earlier tasks, or when you need predictable execution order for debugging.
3. Hierarchical Processes
In hierarchical mode, a manager agent receives all tasks and dynamically delegates them to worker agents based on their roles and capabilities. The manager coordinates, reviews, and may reassign work.
3.1 Hierarchical Crew with Custom Manager
from crewai import Agent, Task, Crew, Process
# Worker agents
developer = Agent(
role="Senior Python Developer",
goal="Write clean, efficient, well-tested Python code",
backstory="You have 10 years of Python experience and love TDD."
)
tester = Agent(
role="QA Engineer",
goal="Find bugs and ensure code quality",
backstory="You are relentless in finding edge cases."
)
# Custom manager agent
tech_lead = Agent(
role="Technical Lead",
goal="Coordinate development for maximum quality and efficiency",
backstory="""You lead a team of developers. You understand task
dependencies, can break down complex requirements, and know
which team member is best suited for each task.""",
allow_delegation=True
)
# Tasks (manager decides assignment)
implement_task = Task(
description="Implement a REST API endpoint for user authentication with JWT tokens.",
expected_output="Complete Python code with FastAPI endpoint, JWT handling, and error responses."
)
test_task = Task(
description="Write comprehensive tests for the authentication endpoint.",
expected_output="Pytest test suite covering happy path, error cases, and edge cases."
)
# Hierarchical crew
crew = Crew(
agents=[developer, tester],
tasks=[implement_task, test_task],
process=Process.hierarchical,
manager_agent=tech_lead,
verbose=True
)
result = crew.kickoff()
print(f"Hierarchical result: {result.raw[:200]}...")
Due Diligence Automation
A VC firm uses CrewAI with hierarchical process for startup evaluation: a Manager Agent assigns research tasks to specialists (market analyst, technical reviewer, financial modeler). The hierarchy ensures thoroughness — the manager identifies gaps and assigns follow-up research. Result: due diligence time reduced from 2 weeks to 2 days.
4. Conditional Tasks
Conditional tasks execute only when a specified condition is met, enabling branching logic within your crew’s workflow.
4.1 Conditional Execution with Guard Functions
from crewai import Agent, Task, Crew, Process
from crewai.tasks.conditional_task import ConditionalTask
analyst = Agent(
role="Data Analyst",
goal="Analyze data quality and completeness",
backstory="You specialize in data validation."
)
cleaner = Agent(
role="Data Cleaner",
goal="Clean and normalize messy datasets",
backstory="You handle missing values, outliers, and formatting."
)
reporter = Agent(
role="Report Writer",
goal="Write clear data quality reports",
backstory="You communicate technical findings to stakeholders."
)
# First task: assess data quality
assess_task = Task(
description="Assess the quality of the {dataset} dataset. Report completeness percentage.",
expected_output="Data quality score (0-100) and list of issues found.",
agent=analyst
)
# Conditional: only clean if quality is below threshold
def needs_cleaning(context):
"""Only execute cleaning if quality score < 80."""
output = context.get("assess_task", "")
# Simple heuristic: check if "below 80" or low score mentioned
return "issues found" in output.lower() or "incomplete" in output.lower()
clean_task = ConditionalTask(
description="Clean the dataset: handle missing values, remove duplicates, normalize formats.",
expected_output="Cleaned dataset summary with changes made.",
agent=cleaner,
condition=needs_cleaning
)
# Always runs: final report
report_task = Task(
description="Write a summary report of the data pipeline results.",
expected_output="Executive summary of data quality and any cleaning performed.",
agent=reporter,
context=[assess_task]
)
crew = Crew(
agents=[analyst, cleaner, reporter],
tasks=[assess_task, clean_task, report_task],
process=Process.sequential,
verbose=True
)
result = crew.kickoff(inputs={"dataset": "customer_transactions_2026"})
print(f"Pipeline result: {result.raw[:200]}...")
Conditional Tasks vs Flows: For simple if/then branching, conditional tasks work well. For complex multi-path routing, loops, or state management, use Flows (covered in Part 5). Flows provide @router decorators for sophisticated control flow.
5. Agent Collaboration
When allow_delegation=True, agents can ask other agents for help during task execution. This enables emergent collaboration where agents recognize their limitations and seek expertise.
5.1 Delegation Patterns
from crewai import Agent, Task, Crew, Process
# Agent that CAN delegate
lead_researcher = Agent(
role="Lead Researcher",
goal="Produce comprehensive research by leveraging team expertise",
backstory="""You lead a research team. When you encounter topics
outside your core expertise, you delegate to specialists.""",
allow_delegation=True # Can ask other agents for help
)
# Specialist agents (available for delegation)
stats_expert = Agent(
role="Statistical Analyst",
goal="Provide rigorous statistical analysis",
backstory="You are a PhD statistician specialized in causal inference.",
allow_delegation=False
)
domain_expert = Agent(
role="Healthcare Domain Expert",
goal="Provide deep healthcare industry knowledge",
backstory="You are a physician with 20 years of clinical experience.",
allow_delegation=False
)
# Task assigned to lead — may delegate subtasks
research_task = Task(
description="""Research the effectiveness of AI diagnostics in
healthcare. Include statistical evidence and clinical perspectives.""",
expected_output="""A comprehensive report combining statistical
analysis of AI diagnostic accuracy with clinical expert opinions
on practical implementation challenges.""",
agent=lead_researcher
)
# All agents available in crew for delegation
crew = Crew(
agents=[lead_researcher, stats_expert, domain_expert],
tasks=[research_task],
process=Process.sequential,
verbose=True
)
result = crew.kickoff()
print(f"Collaborative result: {result.raw[:300]}...")
5.2 Common Collaboration Patterns
Collaboration Anti-Patterns: (1) Don’t give all agents
allow_delegation=True — creates infinite loops. (2) Don’t have more than one delegating agent per crew unless using hierarchical process. (3) Keep delegation chains shallow (max 2 levels deep).
Try It Yourself: Build a ‘competitive analysis’ workflow with 4 tasks in a sequential process: (1) identify 3 competitors, (2) research each competitor’s pricing, (3) analyze strengths/weaknesses, (4) write a comparison report. Then refactor to hierarchical process with a Manager agent that delegates. Compare the quality of outputs between sequential and hierarchical.
Next in the CrewAI SDK Track
In Part 4: Crews & Production Architecture, we’ll build production-grade crews with the @CrewBase decorator pattern, async and batch execution, the recommended Flow-wrapping-Crew architecture, and deployment patterns with FastAPI.