CrewAI

Overview

CrewAI is an open-source framework for orchestrating autonomous AI agents working collaboratively to accomplish complex tasks. The framework emphasizes role-based coordination where each agent has specific roles, goals, and backstories, enabling specialized expertise and task delegation. CrewAI provides structured workflows through multiple process types, event-driven execution via Flows, and comprehensive memory systems for maintaining context across agent interactions.

The framework addresses the problem of coordinating multiple specialized agents through a crew orchestration model where agents, tasks, and execution processes are explicitly defined and managed. CrewAI supports Python and integrates with various LLM providers.

Key technical components covered:

• Core architecture (agents, tasks, crews, flows)
• Process types and execution models
• Memory system implementation
• Agent delegation and communication
• Tools and function calling
• Event-driven workflows with Flows
• Template system and configuration
• Version history and breaking changes

Core Architecture

CrewAI’s architecture comprises four fundamental components:

Agents are autonomous entities with defined roles (job functions), goals (objectives to achieve), backstories (context for personality and decision-making), and tools (external functions available for use). Agents utilize LLMs for reasoning and can manage short-term and long-term memory for stateful execution. Each agent operates independently but can communicate and delegate tasks to other agents.

Tasks represent discrete units of work with description (what needs to be done), expected_output (format and content of results), and agent (responsible agent). Tasks can execute sequentially or in parallel depending on process configuration. Task output serves as context for subsequent tasks in the workflow.

Crews form the orchestration layer grouping agents and tasks together. A crew determines execution order, manages agent collaboration, and controls workflow patterns. Crews can operate in different process modes (sequential, hierarchical, or planned consensual) defining how tasks are assigned and executed.

Flows provide event-driven workflow management combining tasks, crews, and code into multi-step processes. Flows enable control over state, execution paths, conditional logic, and persistence. Built with decorators like @start(), @listen(), and @router() defining event triggers and state transitions.

Process Types and Execution Models

CrewAI implements three process types for task orchestration:

Sequential Process executes tasks one after another in predefined order. Each task’s output automatically becomes context for the next task, ensuring logical information flow. Requires explicit agent assignment for each task via the agent field. Suitable for projects where tasks depend on completion of previous ones.

from crewai import Crew, Process, Agent, Task
 
researcher = Agent(role='Researcher', goal='Conduct research')
analyst = Agent(role='Analyst', goal='Analyze data')
 
research_task = Task(description='Gather data...', agent=researcher, expected_output='Raw Data')
analysis_task = Task(description='Analyze data...', agent=analyst, expected_output='Insights')
 
crew = Crew(
    agents=[researcher, analyst],
    tasks=[research_task, analysis_task],
    process=Process.sequential
)

Hierarchical Process introduces a manager agent who oversees task execution, planning, delegation, and validation. The manager dynamically assigns tasks to agents based on context and capabilities, emulating corporate hierarchy. The manager can be user-specified or automatically created by CrewAI. Requires defining manager_llm or manager_agent parameter. Suitable for complex projects requiring dynamic task allocation and oversight.

from crewai import Crew, Process
from langchain_openai import ChatOpenAI
 
crew = Crew(
    agents=[researcher, analyst],
    tasks=[research_task, analysis_task],
    process=Process.hierarchical,
    manager_llm=ChatOpenAI(model="gpt-4")
)

Consensual Process (planned) aims to enable collaborative decision-making among agents on task execution. Introduces democratic approach to task management. Not yet implemented in codebase but reflects ongoing development roadmap.

Parallel Execution can be enabled within sequential or hierarchical processes by setting async_execution=True on individual tasks, allowing independent tasks to run concurrently.

Memory System Implementation

CrewAI implements a structured memory system with four components using ChromaDB (vector database) and SQLite3 (relational database):

Short-Term Memory stores recent interactions and outcomes using ChromaDB with RAG (Retrieval-Augmented Generation). Enables agents to recall information relevant to current context during ongoing execution. Utilizes vector embeddings for similarity search to retrieve contextually relevant information.

Long-Term Memory preserves insights and learnings from past executions using SQLite3 for persistent storage. Stores task results across sessions, enabling agents to build and refine knowledge over time. Provides cross-session continuity for agent learning.

Entity Memory captures and organizes information about entities (people, places, concepts) encountered during tasks using ChromaDB with RAG. Facilitates deeper understanding and relationship mapping between entities. Enables agents to maintain structured knowledge graphs.

Contextual Memory combines short-term, long-term, and entity memories to maintain coherence across task sequences. Provides unified context for agent decision-making by aggregating relevant information from all memory types.

Memory data resides in platform-specific directories (e.g., ~/Library/Application Support/CrewAI/{project_name}/ on macOS). The system supports custom storage backends like Qdrant or Mem0 for enhanced performance and features through pluggable memory providers.

Agent Delegation and Communication

CrewAI agents support autonomous inter-agent delegation enabling collaborative problem-solving:

Agent delegation is managed through the allow_delegation attribute. When set to True, an agent can assign tasks to peer agents. The allowed_agents parameter defines a controlled delegation hierarchy specifying which agents can receive delegated tasks. This hierarchical structure ensures delegation occurs within predefined boundaries.

Built-in collaboration tools facilitate autonomous communication:

• Delegate work to coworker tool allows agents to assign tasks to designated colleagues
• Ask question to coworker tool enables agents to seek information or clarification from peers

These tools support dynamic collaboration where agents leverage each other’s expertise and resources. Agents automatically use these tools when allow_delegation=True without requiring explicit tool registration.

Implementation pattern:

from crewai import Agent
 
manager = Agent(
    role="Project Manager",
    goal="Coordinate team efforts",
    backstory="Experienced in delegation and oversight",
    allow_delegation=True,
    verbose=True
)
 
researcher = Agent(
    role="Researcher",
    goal="Conduct analysis",
    backstory="Expert in data research",
    allow_delegation=False
)

The manager agent can delegate tasks to the researcher agent, but the researcher cannot delegate further. This creates clear delegation hierarchies preventing circular dependencies.

Tools and Function Calling

CrewAI integrates external functionality through tools defined using the BaseTool class:

Tool definition requires specifying name, description, and args_schema (Pydantic model defining input parameters):

from crewai.tools import BaseTool
from pydantic import BaseModel, Field
 
class ExampleToolInput(BaseModel):
    param1: str = Field(..., description="Description of param1")
    param2: int = Field(..., description="Description of param2")
 
class ExampleTool(BaseTool):
    name: str = "Example Tool"
    description: str = "Description of what the tool does"
    args_schema: Type[BaseModel] = ExampleToolInput
 
    def _run(self, param1: str, param2: int) -> str:
        return f"Processed {param1} and {param2}"

Tool integration assigns tools to agents via the tools parameter:

agent = Agent(
    role="Example Role",
    goal="Achieve objective",
    backstory="Agent backstory",
    tools=[ExampleTool()],
    verbose=True
)

Function calling is implemented for LLMs supporting this capability. The agent’s logic parses responses to detect tool calls, executes them, and continues the conversation with results. Tool calls are handled through a message loop appending tool results to the conversation history for the LLM to process.

Event listeners can monitor tool usage through the event system:

from crewai.utilities.events.base_event_listener import BaseEventListener
from crewai.utilities.events.task_events import TaskStartedEvent, TaskCompletedEvent
 
class MyEventListener(BaseEventListener):
    def setup_listeners(self, crewai_event_bus):
        @crewai_event_bus.on(TaskStartedEvent)
        def on_task_started(source, event: TaskStartedEvent):
            print(f"Task started: {source}")

Event-Driven Workflows with Flows

CrewAI Flows provide structured, event-driven framework for orchestrating workflows with precise control over execution paths, state management, and conditional logic.

Event-driven architecture uses decorators to define method triggers:

• @start(): Entry point method triggered when flow begins
• @listen(): Method triggered by completion of another method or event
• @router(): Method implementing conditional branching based on state

State management offers two approaches:

Unstructured state uses dictionaries for flexibility:

from crewai.flow.flow import Flow, start, listen
 
class MyFlow(Flow):
    @start()
    def first_method(self):
        self.state["data"] = "some value"
        return "completed"
    
    @listen(first_method)
    def second_method(self):
        data = self.state.get("data")
        return f"processed {data}"

Structured state uses Pydantic models for type safety and validation:

from pydantic import BaseModel
 
class MyState(BaseModel):
    data: str
    count: int = 0
 
class MyFlow(Flow[MyState]):
    @start()
    def first_method(self):
        self.state.data = "value"
        self.state.count += 1

Conditional logic enables branching workflows through router methods:

@router(some_method)
def decide_next(self):
    if self.state.count > 5:
        return "high_count_handler"
    else:
        return "low_count_handler"

Crew integration allows flows to orchestrate crews within workflow steps, combining autonomous agent collaboration with structured flow control.

Template System and Configuration

CrewAI allows customization of agent behavior through template parameters:

system_template defines the initial system message setting agent’s role, goal, and backstory:

system_template = """You are {role}. {backstory}
Your goal is: {goal}
 
Respond naturally and conversationally."""

prompt_template structures user prompts the agent responds to:

prompt_template = """Task: {input}
 
Please complete this task carefully.
 
Additional context:
- Use professional tone
- Maintain consistency
- Ensure clarity"""

response_template formats assistant’s response:

response_template = """Thought: {thought}
Final Answer: {answer}"""

Agent configuration with custom templates:

agent = Agent(
    role="Research Assistant",
    goal="Assist with research",
    backstory="Knowledgeable assistant with resources",
    system_template=system_template,
    prompt_template=prompt_template,
    response_template=response_template,
    verbose=True
)

Template variables use placeholders like {role}, {goal}, {backstory}, {input} automatically populated during execution. These enable dynamic content injection without hardcoding agent-specific information.

For model files (e.g., with Ollama), templates can be defined within the model file using SYSTEM, PARAMETER, and other directives for centralized model behavior configuration.

Version History and Breaking Changes

v0.14.0 (February 22, 2024) removed crewai_tools dependency. Added support for exporting outputs in JSON and Pydantic formats for open-source models. v0.14.4 (February 24, 2024) set agents to memoryless by default to reduce token usage, affecting users relying on default memory settings.

v0.16.0 (February 28, 2024) removed remaining crewai_tools dependency. Added initial support for input interpolation and tracking for tool usage, tool errors, format errors, and token usage.

v0.19.0 (March 4, 2024) enhanced tool usage efficiency resulting in 1023.21% increase. Improved delegation functionality. Allowed setting function_calling_llm on Agent or Crew. Enabled passing inputs during kickoff via crew.kickoff(inputs={'key': 'value'}).

v0.22.0 (March 11, 2024) introduced initial CLI with crewai create command. Allowed defining agents and tasks as dictionaries. Enhanced agent logging. Fixed maximum recursion depth exceeded bug.

v0.41.0 (July 19, 2024) introduced type-safe output where all crews and tasks return proper TaskOutput and CrewOutput objects (breaking change). Added planning feature for crews via planning=True on Crew instance allowing pre-action planning. Introduced replay feature via CLI command to list tasks from last run and replay from specific point.

CrewAI maintains rapid development pace with frequent releases adding features and occasionally introducing breaking changes requiring migration of existing implementations.