CrewAI: Multi-Agent Orchestration Framework for Python

1. What Is CrewAI?

CrewAI is an open-source Python framework for orchestrating autonomous AI agents that collaborate as a crew. Created by João Moura and released in late 2023, it models multi-agent systems as teams of role-playing specialists -- each agent has a role, a goal, and a backstory that shapes its behavior. Agents work together on tasks organized into sequential or hierarchical processes, producing structured outputs that flow through the pipeline.

The core design philosophy is role-based collaboration over rigid programming. Instead of writing explicit control flow for every decision, you define agents with natural-language roles and goals, assign them tasks with expected outputs, and let the framework orchestrate their collaboration. This makes CrewAI particularly effective for knowledge work automation: research, content creation, analysis, planning, and any workflow where multiple specialists contribute to a shared outcome.

As of April 2026, CrewAI has over 25,000 GitHub stars and is one of the most widely adopted multi-agent frameworks in the Python ecosystem. The framework supports any LLM provider (OpenAI, Anthropic Claude, Google Gemini, Ollama, Groq, Azure OpenAI, and more), includes a built-in memory system, integrates with MCP servers and LangChain tools, and offers CrewAI+ -- a managed platform for production deployment with monitoring, testing, and enterprise features.

3. Agent Design

Agent design is the most important part of building effective crews. Each agent is defined by three natural-language fields that shape its behavior: role (what the agent is), goal (what it is trying to achieve), and backstory (context that gives the agent personality and expertise). These fields are injected into the system prompt, so crafting them well is critical for agent performance.

Beyond the core identity fields, agents accept configuration for LLM assignment (llm parameter), tool access (tools list), delegation behavior (allow_delegation), maximum iteration limits (max_iter), maximum execution time (max_execution_time), and memory settings. You can assign different LLMs to different agents -- use a powerful model for complex reasoning tasks and a fast, cheap model for simple data extraction.

from crewai import Agent, LLM

# Agent with specific LLM and configuration
analyst = Agent(
    role="Senior Financial Analyst",
    goal="Produce accurate financial analysis with actionable "
         "investment recommendations for {company}",
    backstory="You are a CFA charterholder with 20 years of experience "
              "in equity research. You specialize in technology sector "
              "analysis and are known for your rigorous, data-driven "
              "approach to valuation.",
    llm=LLM(model="anthropic/claude-sonnet-4-20250514", temperature=0.1),
    tools=[financial_data_tool, sec_filing_tool],
    allow_delegation=False,
    max_iter=15,
    max_execution_time=300,  # 5 minutes
    verbose=True
)

# YAML-based agent definition (config/agents.yaml)
# researcher:
#   role: "Senior Research Analyst"
#   goal: "Find comprehensive information on {topic}"
#   backstory: "Expert researcher with deep domain knowledge."
#   llm: openai/gpt-4o
#   max_iter: 10

4. Task Definition

Tasks are the units of work in CrewAI. Each task has a description (what to do), an expected_output (what the result should look like), and an agent (who does it). The description and expected_output support {variable} interpolation, so you can parameterize tasks at runtime via crew.kickoff(inputs={...}).

The context parameter creates data dependencies between tasks. When a task lists other tasks in its context, it receives their outputs as additional input. This is how information flows through a crew: the researcher's output feeds the writer, the writer's output feeds the editor. Context creates a DAG (directed acyclic graph) of task dependencies that CrewAI resolves automatically.

from crewai import Task
from pydantic import BaseModel
from typing import List

# Structured output model
class ResearchReport(BaseModel):
    topic: str
    key_findings: List[str]
    recommendations: List[str]
    confidence_score: float

# Task with structured output and context
research_task = Task(
    description="Research {topic} and produce a structured analysis. "
                "Focus on: current state, key trends, major players, "
                "and technical challenges. Use available search tools.",
    expected_output="A structured research report with key findings, "
                    "actionable recommendations, and confidence score.",
    agent=researcher,
    output_pydantic=ResearchReport
)

# Async tasks that run in parallel
market_task = Task(
    description="Analyze market trends for {topic}.",
    expected_output="Market analysis with growth projections.",
    agent=market_analyst,
    async_execution=True
)

tech_task = Task(
    description="Analyze technical landscape for {topic}.",
    expected_output="Technical analysis with architecture comparisons.",
    agent=tech_analyst,
    async_execution=True
)

# Task with callback
def save_report(output):
    with open("report.md", "w") as f:
        f.write(output.raw)
    print(f"Report saved: {len(output.raw)} characters")

final_task = Task(
    description="Synthesize all research into a final report.",
    expected_output="A comprehensive report combining all analyses.",
    agent=writer,
    context=[market_task, tech_task],
    callback=save_report
)

5. Process Types

The process type determines how CrewAI orchestrates task execution. Choosing the right process depends on your workflow's coordination needs: predictable pipelines use sequential, complex delegation uses hierarchical, and collaborative decision-making uses consensual.

from crewai import Crew, Process

# Sequential: tasks run in listed order
sequential_crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential,
    verbose=True
)

# Hierarchical: manager delegates to agents
hierarchical_crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.hierarchical,
    manager_llm="openai/gpt-4o",  # LLM for the auto-created manager
    verbose=True
)

# Hierarchical with custom manager
manager = Agent(
    role="Project Manager",
    goal="Coordinate the team to produce the highest quality output",
    backstory="Experienced PM who excels at delegating tasks to the "
              "right specialists and ensuring quality standards.",
    allow_delegation=True
)

custom_manager_crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.hierarchical,
    manager_agent=manager,
    verbose=True
)

7. Memory System

CrewAI includes a multi-layer memory system that gives agents the ability to learn and retain information across tasks and crew executions. Memory is disabled by default and enabled with memory=True on the Crew. When enabled, agents can recall previous interactions, build up entity knowledge, and improve their performance over time.

from crewai import Crew

# Enable memory with default storage
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    memory=True,           # enables all memory layers
    verbose=True
)

# Custom memory configuration
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    memory=True,
    embedder={
        "provider": "openai",
        "config": {
            "model": "text-embedding-3-small"
        }
    },
    long_term_memory=LongTermMemory(
        storage=LTMSQLiteStorage(db_path="./crew_memory.db")
    ),
    short_term_memory=ShortTermMemory(
        storage=RAGStorage(
            embedder_config={
                "provider": "openai",
                "config": {"model": "text-embedding-3-small"}
            }
        )
    ),
    entity_memory=EntityMemory(
        storage=RAGStorage(
            embedder_config={
                "provider": "openai",
                "config": {"model": "text-embedding-3-small"}
            }
        )
    )
)

8. Flows API

The Flows API is CrewAI's higher-level abstraction for building event-driven, stateful workflows that coordinate multiple crews and external operations. While Crews handle multi-agent collaboration on a set of tasks, Flows handle the orchestration of multiple crews, conditional logic, state management, and integration with external systems in complex pipelines.

Flows use a decorator-based syntax: @start() marks the entry point, @listen() connects methods that react to events (outputs from other methods), and @router() implements conditional branching based on state or results. The Flow maintains typed state across all steps, making it easy to build complex, multi-stage workflows with clear data flow.

from crewai.flow.flow import Flow, listen, start, router
from pydantic import BaseModel

class ContentPipelineState(BaseModel):
    topic: str = ""
    research: str = ""
    draft: str = ""
    quality_score: float = 0.0
    final_content: str = ""

class ContentPipeline(Flow[ContentPipelineState]):

    @start()
    def gather_topic(self):
        self.state.topic = "Multi-Agent AI Orchestration"
        return self.state.topic

    @listen(gather_topic)
    def research_phase(self, topic):
        # Run a research crew
        research_crew = Crew(
            agents=[researcher],
            tasks=[research_task],
            verbose=True
        )
        result = research_crew.kickoff(inputs={"topic": topic})
        self.state.research = result.raw
        return result.raw

    @listen(research_phase)
    def writing_phase(self, research):
        # Run a writing crew
        writing_crew = Crew(
            agents=[writer, editor],
            tasks=[writing_task, editing_task],
            process=Process.sequential
        )
        result = writing_crew.kickoff(
            inputs={"research": research}
        )
        self.state.draft = result.raw
        return result.raw

    @router(writing_phase)
    def quality_check(self, draft):
        # Evaluate quality and decide next step
        score = evaluate_quality(draft)
        self.state.quality_score = score
        if score >= 0.8:
            return "publish"
        return "revise"

    @listen("publish")
    def publish(self):
        self.state.final_content = self.state.draft
        return f"Published: {len(self.state.final_content)} chars"

    @listen("revise")
    def revise(self):
        # Re-run writing with feedback
        return self.writing_phase(self.state.research)

# Run the flow
flow = ContentPipeline()
result = flow.kickoff()

11. Framework Comparison

The multi-agent framework landscape in 2026 includes several major players. Each framework makes different trade-offs between abstraction level, flexibility, and production readiness. The right choice depends on your workflow complexity, team expertise, and deployment requirements.

CrewAI excels when you need structured team collaboration with clear roles -- it has the highest abstraction level, making it the fastest to prototype multi-agent workflows. AutoGen (Microsoft) is strongest for multi-turn conversational patterns where agents debate and refine outputs. LangGraph offers the most control with explicit graph-based state machines, ideal for complex workflows that need checkpointing and time-travel debugging. Claude Agent SDK has the deepest system access with built-in file and shell tools. OpenAI Agents SDK provides the simplest path to multi-agent with minimal abstractions and sandbox execution.

12. Production Deployment

Moving CrewAI from prototyping to production requires addressing reliability, monitoring, error handling, cost control, and scaling. CrewAI+ (the managed platform) handles many of these concerns, but self-hosted deployments need explicit infrastructure around the framework.

# Production deployment with error handling and monitoring
import logging
from crewai import Crew
from tenacity import retry, stop_after_attempt, wait_exponential

logger = logging.getLogger("crewai_production")

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=60)
)
def run_crew_with_retries(crew, inputs):
    try:
        result = crew.kickoff(inputs=inputs)
        logger.info(
            f"Crew completed: {result.token_usage}"
        )
        return result
    except Exception as e:
        logger.error(f"Crew execution failed: {e}")
        raise

# Deploy as API endpoint (FastAPI example)
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()

@app.post("/api/analyze")
async def analyze(topic: str, background_tasks: BackgroundTasks):
    crew = Crew(
        agents=[researcher, analyst, writer],
        tasks=[research_task, analysis_task, report_task],
        process=Process.sequential,
        memory=True,
        max_rpm=30  # rate limit across all agents
    )
    background_tasks.add_task(
        run_crew_with_retries, crew, {"topic": topic}
    )
    return {"status": "processing", "topic": topic}

# CLI deployment with CrewAI+
# crewai deploy    -- deploys to CrewAI+ platform
# crewai monitor   -- real-time crew monitoring
# crewai logs      -- view execution logs

Getting Started

CrewAI requires Python 3.10 or newer. Install with pip and create your first crew in minutes.

# Install CrewAI and tools
pip install crewai crewai-tools

# Create a new project with the CLI
crewai create crew my_project
cd my_project

# Configure agents in config/agents.yaml
# Configure tasks in config/tasks.yaml
# Run the crew
crewai run

# Minimal crew -- your first multi-agent workflow
from crewai import Agent, Task, Crew

researcher = Agent(
    role="Researcher",
    goal="Find accurate information on {topic}",
    backstory="Expert research analyst."
)

writer = Agent(
    role="Writer",
    goal="Write clear, engaging content based on research",
    backstory="Skilled technical writer."
)

research = Task(
    description="Research {topic} thoroughly.",
    expected_output="Comprehensive research brief.",
    agent=researcher
)

article = Task(
    description="Write an article based on the research.",
    expected_output="A polished 1000-word article.",
    agent=writer,
    context=[research]
)

crew = Crew(agents=[researcher, writer], tasks=[research, article])
result = crew.kickoff(inputs={"topic": "CrewAI multi-agent orchestration"})
print(result)

For production use, configure your LLM API keys as environment variables, enable memory for learning across runs, and use the Flows API for complex multi-crew orchestration. The official documentation at docs.crewai.com covers all concepts, integrations, and advanced patterns in depth.