CrewAI

Role-based, process-driven multi-agent framework. Optimized for structured team workflows.

Core Concepts

Agents = Roles

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="Senior Research Analyst",
    goal="Find cutting-edge information on {topic}",
    backstory="""You're an expert researcher with 10 years experience
    in technology trend analysis. You know how to find the signal
    in the noise.""",
    tools=[search_tool, scrape_tool],
    llm="gpt-4o",
    verbose=True,
    allow_delegation=True,
    max_iter=5,
    memory=True
)

writer = Agent(
    role="Technical Writer",
    goal="Create clear, engaging content from research",
    backstory="""You transform complex research into accessible
    technical articles. You excel at structure and clarity.""",
    tools=[],
    llm="gpt-4o",
    verbose=True
)

Tasks = Work Items

research_task = Task(
    description="Research the latest developments in {topic}",
    expected_output="Comprehensive report with 5 key findings, sources, and quotes",
    agent=researcher,
    async_execution=False,  # Sequential by default
    context=["previous_research"]  # Dependencies
)

writing_task = Task(
    description="Write a technical blog post based on research",
    expected_output="2000-word blog post with intro, sections, conclusion",
    agent=writer,
    context=[research_task]  # Depends on research
)

Crews = Teams + Processes

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,  # or Process.hierarchical
    verbose=2,
    memory=True,
    cache=True,
    max_rpm=100,  # Rate limiting
    embedder={
        "provider": "openai",
        "model": "text-embedding-3-small"
    }
)

result = crew.kickoff(inputs={"topic": "AI Agents in 2024"})

Processes

Sequential (Default)

Task 1 → Task 2 → Task 3

• Each task waits for previous
• Context passed automatically
• Simple, predictable

Hierarchical

Manager Agent
    ├── Worker 1 → Task 1
    ├── Worker 2 → Task 2
    └── Worker 3 → Task 3

• Manager agent decomposes, delegates, reviews
• Workers execute in parallel where possible
• Manager reviews and approves

crew = Crew(
    agents=[manager, worker1, worker2],
    tasks=[task1, task2],
    process=Process.hierarchical,
    manager_llm="gpt-4o",  # Separate LLM for manager
    manager_agent=manager  # Or explicit manager agent
)

Process Comparison

Agent Configuration

Full Agent Parameters

Agent(
    role="Role Title",
    goal="Objective",
    backstory="Character context",

    # Tools
    tools=[tool1, tool2],
    tool_call_limit=10,

    # LLM
    llm="gpt-4o",
    function_calling_llm="gpt-4o-mini",  # For tools

    # Behavior
    verbose=True,
    allow_delegation=True,  # Can delegate to other agents
    max_iter=15,  # Max reasoning loops
    max_rpm=60,   # Rate limit

    # Memory
    memory=True,
    embedder={"provider": "openai", "model": "text-embedding-3-small"},

    # Callbacks
    step_callback=lambda step: log_step(step),

    # Constraints
    max_execution_time=300,  # Seconds
    respect_context_window=True
)

Tools

Built-in Tools

from crewai_tools import (
    SerperDevTool,      # Google search
    ScrapeWebsiteTool,  # Web scraping
    FileReadTool,       # Read files
    FileWriteTool,      # Write files
    DirectoryReadTool,  # List directory
    CSVSearchTool,      # Query CSV
    JSONSearchTool,     # Query JSON
    EXASearchTool,      # Exa.ai search
    YouTubeChannelTool, # YouTube data
    GithubSearchTool,   # GitHub search
)

Custom Tools

from crewai.tools import BaseTool
from pydantic import BaseModel, Field

class MyToolInput(BaseModel):
    query: str = Field(description="Search query")

class MyTool(BaseTool):
    name: str = "my_custom_tool"
    description: str = "Description for LLM"
    args_schema: type[BaseModel] = MyToolInput

    def _run(self, query: str) -> str:
        # Implementation
        return result

# Use
agent = Agent(tools=[MyTool()], ...)

Tool Calling

• Uses OpenAI function calling format
• Supports parallel tool calls
• Automatic retry on failure

Memory

Short-term (Conversation)

• Automatic within crew execution
• Passed via task context

Long-term (Embeddings)

crew = Crew(
    agents=[...],
    tasks=[...],
    memory=True,
    embedder={
        "provider": "openai",
        "model": "text-embedding-3-small",
        "config": {"api_key": "..."}
    }
)
# Stores: task outputs, agent reasoning, tool results
# Retrieves: relevant context for new tasks

Entity Memory

# Tracks entities across tasks
entity_memory = True  # Remembers people, companies, concepts

Flows (Structured Processes)

from crewai import Flow, Start, Listen, Router

class ResearchFlow(Flow):
    @Start()
    def start_research(self):
        return {"topic": self.inputs["topic"]}

    @Listen("start_research")
    def deep_research(self, topic):
        # Parallel research agents
        return {"findings": parallel_research(topic)}

    @Router("deep_research")
    def route_by_finding(self, findings):
        if findings["needs_deep_dive"]:
            return "deep_dive"
        return "write"

    @Listen("deep_dive")
    def deep_dive(self, findings):
        return {"enhanced": deep_research(findings)}

    @Listen("write")
    def write_report(self, findings):
        return {"report": write(findings)}

flow = ResearchFlow()
result = flow.kickoff(inputs={"topic": "AI Agents"})

Human-in-the-Loop

Human Tools

from crewai_tools import HumanInputTool

human_tool = HumanInputTool()

agent = Agent(
    role="Analyst",
    tools=[human_tool],
    # When agent needs human input, calls tool
)

Task-Level Approval

task = Task(
    description="...",
    agent=analyst,
    human_input=True  # Pauses for human review before execution
)

Flows (CrewAI Flows - New)

from crewai.flow import Flow, Start, Listen, Router

class MyFlow(Flow):
    @Start()
    def start(self):
        return {"data": "initial"}

    @Listen("start")
    def step1(self, state):
        return {"result": process(state["data"])}

    @Router("step1")
    def route(self, state):
        if state["result"].needs_review:
            return "review"
        return "finalize"

    @Listen("review")
    def human_review(self, state):
        return {"approved": human_approve(state["result"])}

    @Listen("finalize")
    def finalize(self, state):
        return {"output": finalize(state["result"])}

flow = MyFlow()
result = flow.kickoff()

Deployment

CrewAI Enterprise

# Hosted platform with:
# - RBAC, audit logs
# - Scaling, queue management
# - Analytics, cost tracking
# - SSO, compliance

from crewai import Crew
crew = Crew(..., enterprise=True)

Self-Hosted

FROM python:3.11-slim
RUN pip install crewai crewai-tools
COPY . .
CMD ["python", "main.py"]

Kubernetes

# Deploy as job or deployment
# Use Redis for shared memory/cache
# Use PostgreSQL for persistence

Best Practices

Agent Design

1. Specific roles — "Senior Python Dev" > "Developer"
2. Clear goals — Measurable, time-bound
3. Rich backstories — Drives persona consistency
4. Minimal tools — Only what's needed
5. Test prompts — Iterate on backstory/goal

Task Design

1. Explicit expected_output — Enables validation
2. Single responsibility — One task, one outcome
3. Context dependencies — Use context=[task] not implicit
4. Async where possible — async_execution=True for independent

Crew Design

1. Start sequential — Move to hierarchical only when needed
2. Limit agents — 3-5 typical, >7 gets unwieldy
3. Monitor costs — max_rpm, token tracking
4. Enable memory — For multi-session continuity

Debugging

Verbose Levels

crew = Crew(..., verbose=0)  # Silent
crew = Crew(..., verbose=1)  # Task starts/ends
crew = Crew(..., verbose=2)  # Full reasoning + tool calls

Step Callbacks

def log_step(step):
    print(f"Agent: {step.agent.role}")
    print(f"Action: {step.action}")
    print(f"Result: {step.result}")

agent = Agent(..., step_callback=log_step)

Debug Mode

import os
os.environ["CREWAI_DEBUG"] = "1"
# Logs all prompts, responses, tool calls

Related Notes

• Framework Comparison — vs LangGraph, AutoGen
• MultiAgent Patterns — Role-based patterns
• Coordination Patterns — Sequential, hierarchical
• LangGraph — Alternative for stateful
• AutoGen — Alternative for conversational
• Human in the Loop — Human tools
• Memory Systems — Memory patterns
• Observability — Monitoring

Sources: CrewAI docs (docs.crewai.com), CrewAI GitHub, CrewAI Enterprise docs, CrewAI Flows docs, crewai-tools package, community examples, production case studies.