CrewAI Multi-Agent Cheat Sheet

Installation & Setup

# Install CrewAI
pip install crewai crewai-tools

# With extras for additional tools
pip install 'crewai[tools]'

# Basic imports
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

Create an Agent

from crewai import Agent

researcher = Agent(
    role='Senior Research Analyst',
    goal='Uncover cutting-edge developments in AI',
    backstory="""You are a senior research analyst at a leading
    tech think tank. Your expertise lies in identifying emerging
    trends and technologies in AI.""",
    verbose=True,
    allow_delegation=False,
    tools=[search_tool, scrape_tool],
    llm=ChatOpenAI(model="gpt-4")
)

# Key Parameters:
# - role: Agent's function
# - goal: What agent aims to achieve
# - backstory: Context for decision-making
# - verbose: Print detailed logs
# - allow_delegation: Can delegate to other agents
# - tools: List of available tools
# - llm: Language model to use

Define Tasks

from crewai import Task

research_task = Task(
    description="""Conduct a comprehensive analysis of the latest
    advancements in AI in 2025. Identify key trends, breakthrough
    technologies, and potential industry impacts.""",
    expected_output="""A detailed report with at least 10 key findings,
    including statistics and examples for each trend.""",
    agent=researcher,
    async_execution=False
)

writing_task = Task(
    description="""Using the research provided, develop an engaging
    blog post highlighting the most significant AI advancements.""",
    expected_output="""A 4-paragraph blog post formatted in markdown,
    with an engaging introduction, key points, and conclusion.""",
    agent=writer,
    context=[research_task]  # Depends on research_task output
)

Create a Crew

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential,  # or Process.hierarchical
    verbose=2,  # 0, 1, or 2
    memory=True,  # Enable memory
    embedder={
        "provider": "openai",
        "config": {"model": "text-embedding-3-small"}
    }
)

# Execute the crew
result = crew.kickoff()
print(result)

Process Types

Built-in Tools

from crewai_tools import (
    SerperDevTool,       # Google search
    ScrapeWebsiteTool,   # Web scraping
    FileReadTool,        # Read files
    DirectoryReadTool,   # Read directories
    CodeInterpreterTool, # Execute code
    WebsiteSearchTool,   # Search websites
    SeleniumScrapingTool # Browser automation
)

# Initialize tools
search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()
file_tool = FileReadTool(file_path='./data.txt')

# Assign to agent
agent = Agent(
    role='Data Analyst',
    tools=[search_tool, file_tool],
    # ... other params
)

Custom Tools

from crewai_tools import BaseTool

class CustomCalculatorTool(BaseTool):
    name: str = "Calculator"
    description: str = "Performs mathematical calculations"

    def _run(self, expression: str) -> str:
        try:
            result = eval(expression)
            return f"Result: {result}"
        except Exception as e:
            return f"Error: {str(e)}"

# Use custom tool
calculator = CustomCalculatorTool()

agent = Agent(
    role='Financial Analyst',
    tools=[calculator],
    # ... other params
)

Agent Collaboration Patterns

# Sequential: Each agent works after previous
research → analysis → writing → editing

# Parallel with Merge: Multiple agents work simultaneously
researcher1 ┐
researcher2 ├─→ synthesizer → final_report
researcher3 ┘

# Hierarchical: Manager coordinates
manager
  ├─→ researcher
  ├─→ analyst
  └─→ writer

# Code Example: Parallel Research
crew = Crew(
    agents=[researcher1, researcher2, researcher3, synthesizer],
    tasks=[
        Task(description="Research AI", agent=researcher1, async_execution=True),
        Task(description="Research ML", agent=researcher2, async_execution=True),
        Task(description="Research Data", agent=researcher3, async_execution=True),
        Task(description="Synthesize findings", agent=synthesizer,
             context=[task1, task2, task3])
    ],
    process=Process.sequential
)

Memory & Context

# Enable crew memory
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    memory=True,  # Enables all memory types
    verbose=True
)

# Memory types automatically enabled:
# - Short-term: Current execution context
# - Long-term: Previous crew executions
# - Entity: Facts about entities

# Access memory programmatically
crew.memory.short_term.search("previous findings")
crew.memory.long_term.search("past projects")

# Clear memory
crew.memory.reset()

Task Dependencies

# Context-based dependencies
task1 = Task(description="Research", agent=researcher)
task2 = Task(description="Analyze", agent=analyst, context=[task1])
task3 = Task(description="Write", agent=writer, context=[task1, task2])

# Task2 receives task1 output
# Task3 receives both task1 and task2 outputs

# Async execution with dependencies
task_a = Task(description="Research AI", async_execution=True)
task_b = Task(description="Research ML", async_execution=True)
task_c = Task(description="Combine", context=[task_a, task_b])

# task_a and task_b run in parallel
# task_c waits for both to complete

Hierarchical Process

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[task1, task2, task3],
    process=Process.hierarchical,
    manager_llm=ChatOpenAI(model="gpt-4"),  # Manager's LLM
    verbose=True
)

# CrewAI creates a manager agent that:
# 1. Delegates tasks to appropriate agents
# 2. Validates task completion
# 3. Ensures quality standards
# 4. Coordinates overall workflow

# No need to define manager agent manually

Callbacks & Monitoring

def step_callback(output):
    print(f"Step completed: {output.description}")
    print(f"Output: {output.result}")

def task_callback(output):
    print(f"Task '{output.description}' completed")

crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    step_callback=step_callback,
    task_callback=task_callback
)

result = crew.kickoff()

Error Handling

from crewai import Agent, Task, Crew

try:
    crew = Crew(
        agents=[agent1, agent2],
        tasks=[task1, task2],
        max_rpm=20,  # Rate limiting
        verbose=True
    )

    result = crew.kickoff()

except Exception as e:
    print(f"Crew execution failed: {e}")

# Agent-level error handling
agent = Agent(
    role='Researcher',
    max_iter=15,  # Max iterations before stopping
    max_rpm=10,   # API rate limit
    # ... other params
)

Output Formats

# JSON output
task = Task(
    description="Extract structured data",
    expected_output="JSON with keys: title, summary, tags",
    output_json=True,
    agent=analyst
)

# File output
task = Task(
    description="Generate report",
    expected_output="Comprehensive analysis",
    output_file="report.md",
    agent=writer
)

# Pydantic model output
from pydantic import BaseModel

class Report(BaseModel):
    title: str
    findings: list[str]
    conclusion: str

task = Task(
    description="Create structured report",
    expected_output="Report with title, findings, conclusion",
    output_pydantic=Report,
    agent=analyst
)

Best Practices

Example: Content Creation Crew

from crewai import Agent, Task, Crew

# Define agents
researcher = Agent(
    role='Content Researcher',
    goal='Find accurate, up-to-date information on topics',
    backstory='Expert at finding reliable sources',
    tools=[search_tool],
    verbose=True
)

writer = Agent(
    role='Content Writer',
    goal='Create engaging, well-written content',
    backstory='Award-winning content creator',
    verbose=True
)

editor = Agent(
    role='Editor',
    goal='Ensure quality and accuracy',
    backstory='20 years editing experience',
    verbose=True
)

# Define tasks
research = Task(
    description='Research AI trends in 2025',
    expected_output='List of 10 key trends with sources',
    agent=researcher
)

write = Task(
    description='Write blog post about AI trends',
    expected_output='1000-word blog post in markdown',
    agent=writer,
    context=[research]
)

edit = Task(
    description='Edit and polish the blog post',
    expected_output='Final polished blog post',
    agent=editor,
    context=[write]
)

# Create and run crew
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research, write, edit],
    process=Process.sequential,
    memory=True
)

result = crew.kickoff()
print(result)