CrewAI - Multi-Agent Framework

In the previous lesson we wrote the agent class, the orchestrator and the message chain ourselves. It worked, but every new feature, such as tools, memory or task delegation, would have to be added by hand. When the expedition grows, it is better to use a ready-made camp rulebook.

CrewAI is a Python framework for creating teams of AI agents that collaborate to solve complex tasks. It is like assembling an expedition crew where everyone has their own role. The whole framework rests on three concepts: Agent (who), Task (what) and Crew (how together).

Installation and Basics

We install two packages and import the core classes. The language model is configured with the LLM class from CrewAI itself:

1# pip install crewai crewai-tools
2
3from crewai import Agent, Task, Crew, Process, LLM
4
5# LLM model
6llm = LLM(model="gpt-4o-mini", temperature=0.7)

Watch out for a change in the library: older examples online use LangChain's ChatOpenAI here, but current CrewAI versions do not depend on LangChain and reject such an object when validating the agent. LLM(model="gpt-4o-mini") uses the key from the OPENAI_API_KEY variable under the hood.

Creating Agents

An agent in CrewAI is three sentences about itself: a role (role), a goal (goal) and a backstory (backstory). The model receives them as the description of the character it should play. Here is the researcher:

1# Researcher agent
2researcher = Agent(
3    role="Senior Research Analyst",
4    goal="Conduct thorough research and provide reliable information",
5    backstory="""You are an experienced analyst with 15 years of experience
6    in researching technology trends. You specialize in AI and ML analysis.
7    You always verify facts and look for reliable sources.""",
8    verbose=True,
9    allow_delegation=False,
10    llm=llm
11)

verbose=True prints the agent's reasoning to the console, and allow_delegation=False forbids it from passing work to others. In current versions False is the default anyway.

The writer and the editor look similar, only the character description changes:

1# Writer agent
2writer = Agent(
3    role="Content Writer",
4    goal="Create engaging and educational content about technology",
5    backstory="""You are a talented technical writer with a passion for
6    translating complex concepts into simple language. Your articles
7    are always well-organized and enjoyable to read.""",
8    verbose=True,
9    allow_delegation=True,
10    llm=llm
11)
12
13# Editor agent
14editor = Agent(
15    role="Senior Editor",
16    goal="Ensure the highest quality of published content",
17    backstory="""You are an experienced editor with an eye for detail.
18    You care about style consistency, factual accuracy, and clarity of message.
19    You are not afraid to suggest changes that will improve the text.""",
20    verbose=True,
21    allow_delegation=False,
22    llm=llm
23)

The writer has allow_delegation=True, so when it does not know something, it can ask a crewmate. That is the only technical difference, the rest is character written into backstory.

Defining Tasks

A Task describes the work to be done. Every task has a description, an expected output (expected_output) and an assigned agent:

1# Research task
2research_task = Task(
3    description="""Conduct thorough research on the topic: {topic}
4
5    Your task:
6    1. Find the latest information and trends
7    2. Identify key players and technologies
8    3. Analyze potential applications
9    4. Assess future development directions
10
11    Deliver a detailed research report.""",
12    expected_output="Detailed research report with citations and sources",
13    agent=researcher
14)

The {topic} notation in the description is a placeholder for a variable. CrewAI fills it with the value from the inputs parameter when the crew starts.

The writing task uses the research result. That is what the context parameter is for:

1# Writing task
2writing_task = Task(
3    description="""Based on the research report, write a blog article.
4
5    Requirements:
6    1. Engaging introduction
7    2. Clear structure with headings
8    3. Practical examples
9    4. Summary with key takeaways
10    5. Length: 1000-1500 words
11
12    Use accessible language for technical readers.""",
13    expected_output="Complete blog article ready for publication",
14    agent=writer,
15    context=[research_task]  # Uses the result of the previous task
16)

context=[research_task] hands the researcher's report to the writer, so we do not have to glue texts manually the way our orchestrator did.

The editor receives the writer's article as context:

1# Editing task
2editing_task = Task(
3    description="""Review and edit the article.
4
5    Check:
6    1. Grammar and style correctness
7    2. Argument consistency
8    3. Message clarity
9    4. Engaging tone
10
11    Make necessary corrections.""",
12    expected_output="Edited, final article",
13    agent=editor,
14    context=[writing_task]
15)

Description, expected output, agent and context are the four elements you build every task from.

Creating and Running the Crew

A Crew connects agents and tasks. The process parameter decides how the crew works:

1# Create the team
2content_crew = Crew(
3    agents=[researcher, writer, editor],
4    tasks=[research_task, writing_task, editing_task],
5    process=Process.sequential,  # or Process.hierarchical
6    verbose=True
7)
8
9# Run with inputs
10result = content_crew.kickoff(inputs={"topic": "RAG in enterprise AI"})
11print(result)

Process.sequential runs the tasks one after another, in list order. CrewAI supports two process types: sequential and hierarchical. kickoff returns a CrewOutput object whose text is in the raw field, and print shows it directly.

Tools for Agents

An agent without tools knows only what the model learned. The crewai-tools package provides ready-made tools for search, websites and files:

1from crewai_tools import (
2    SerperDevTool,      # Google search
3    WebsiteSearchTool,  # Website search
4    FileReadTool,       # File reading
5    PDFSearchTool,      # PDF search
6)
7
8# Configure tools
9search_tool = SerperDevTool()
10web_tool = WebsiteSearchTool()
11file_tool = FileReadTool()
12
13# Agent with tools
14researcher_with_tools = Agent(
15    role="Research Analyst",
16    goal="Gather and analyze information from various sources",
17    backstory="Expert in information retrieval...",
18    tools=[search_tool, web_tool, file_tool],
19    verbose=True,
20    llm=llm
21)

SerperDevTool needs a key in the SERPER_API_KEY variable. The agent decides by itself when to reach for which tool, based on their descriptions.

A custom tool is a class that inherits from BaseTool. Today we import it from crewai.tools, not from crewai_tools as in older examples:

1# Custom tool
2from crewai.tools import BaseTool
3from pydantic import BaseModel, Field
4
5class DatabaseQueryInput(BaseModel):
6    query: str = Field(description="SQL query to execute")
7
8class DatabaseQueryTool(BaseTool):
9    name: str = "Database Query"
10    description: str = "Executes SQL queries on the database"
11    args_schema: type[BaseModel] = DatabaseQueryInput
12
13    def _run(self, query: str) -> str:
14        # Query implementation
15        return f"Query result: {query}"

args_schema is a Pydantic model describing the arguments, and the _run method contains the actual logic. The language model reads name and description, so write them clearly.

Hierarchical Process

In a hierarchical process, a manager distributes the tasks. We create it like an ordinary agent:

1# Coordinating manager
2manager = Agent(
3    role="Project Manager",
4    goal="Coordinate teamwork and ensure timely delivery",
5    backstory="Experienced PM with AI team management skills",
6    verbose=True,
7    llm=llm
8)
9
10# Crew with hierarchy
11hierarchical_crew = Crew(
12    agents=[researcher, writer, editor],
13    tasks=[research_task, writing_task, editing_task],
14    process=Process.hierarchical,
15    manager_agent=manager,  # or manager_llm=llm
16    verbose=True
17)

The manager does not go into the agents list but into manager_agent. Instead of your own manager you can pass manager_llm, and CrewAI creates one itself. One of the two is required in a hierarchical process.

Memory and Context

The crew can remember earlier findings. We enable memory with the memory=True flag, and embedder points to the embedding model CrewAI uses to store memories:

1# Crew with memory
2crew_with_memory = Crew(
3    agents=[researcher, writer],
4    tasks=[research_task, writing_task],
5    process=Process.sequential,
6    memory=True,  # Enable memory
7    embedder={
8        "provider": "openai",
9        "config": {"model": "text-embedding-3-small"}
10    },
11    verbose=True
12)
13
14# Agent with memory
15agent_with_memory = Agent(
16    role="Assistant",
17    goal="Help the user",
18    backstory="...",
19    memory=True,
20    llm=llm
21)

These are the same embeddings we learned two lessons ago, only used to store the crew's memories instead of documents.

Callbacks and Monitoring

With longer tasks you want to know what is going on. Crew accepts two callback functions: task_callback after every task and step_callback after every agent step:

1from crewai.tasks.task_output import TaskOutput
2
3def on_task_end(output: TaskOutput) -> None:
4    """Called after each task finishes."""
5    print(f"Completed: {output.description[:50]}")
6    print(f"   Output: {output.raw[:100]}...")
7
8def on_step(step) -> None:
9    """Called after each agent step."""
10    print(f"Agent step: {str(step)[:100]}")
11
12# Usage
13crew = Crew(
14    agents=[researcher, writer],
15    tasks=[research_task, writing_task],
16    task_callback=on_task_end,
17    step_callback=on_step,
18    verbose=True
19)

TaskOutput carries the task description and the raw result in the raw field. Do not look for a CrewCallback class or a crewai.callbacks module: CrewAI has neither, and you pass callbacks exactly through these two parameters.

Asynchronous Execution

You can run several independent crews at the same time. That is what kickoff_async is for:

1import asyncio
2
3async def run_crew_async(topic: str) -> str:
4    """Asynchronous crew execution."""
5    crew = Crew(
6        agents=[researcher, writer],
7        tasks=[research_task, writing_task],
8        process=Process.sequential
9    )
10
11    result = await crew.kickoff_async(inputs={"topic": topic})
12    return result
13
14# Run multiple crews in parallel
15async def run_multiple_crews():
16    topics = ["AI in medicine", "AI in finance", "AI in education"]
17
18    tasks = [run_crew_async(topic) for topic in topics]
19    results = await asyncio.gather(*tasks)
20
21    return dict(zip(topics, results))
22
23results = asyncio.run(run_multiple_crews())

asyncio.gather waits for all three crews at once, and dict(zip(...)) pairs the topics with the results. In production, create separate agent and task objects for each crew, because here they all share the same research_task and writing_task.

Example: Content Pipeline

Finally we put everything together into one pipeline with an extra SEO specialist:

1def create_content_pipeline(topic: str) -> str:
2    """Complete content creation pipeline."""
3
4    # Agents
5    seo_analyst = Agent(
6        role="SEO Specialist",
7        goal="Optimize content for search engines",
8        backstory="SEO expert with content marketing experience",
9        tools=[SerperDevTool()],
10        llm=llm
11    )
12
13    # Tasks
14    seo_task = Task(
15        description=f"Analyze keywords for the topic: {topic}",
16        expected_output="List of 10 keywords with their potential",
17        agent=seo_analyst
18    )
19
20    # Crew
21    pipeline = Crew(
22        agents=[seo_analyst, researcher, writer, editor],
23        tasks=[seo_task, research_task, writing_task, editing_task],
24        process=Process.sequential,
25        verbose=True
26    )
27
28    return pipeline.kickoff(inputs={"topic": topic})
29
30# Run
31article = create_content_pipeline("Machine Learning in 2025")

The seo_task description uses an f-string, but the researcher's and writer's tasks still contain {topic}, which is why kickoff receives inputs. Without it, the agents would see the literal text "{topic}". I recommend starting with the sequential process and turning on the hierarchy only for a bigger crew, because the manager means extra model calls.

CrewAI is a powerful framework for building AI teams. In the next lesson you will learn about production RAG systems - how to deploy these technologies in the enterprise!

Remember: in CrewAI an agent is a crew member, a task is their assignment, and the crew is the whole expedition.