CrewAI - Multi-Agent Framework

CrewAI is a Python framework for creating teams of AI agents that collaborate on solving complex tasks. It is like assembling a team of specialists, where each one has their role!

Installation and Basics

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

Creating Agents

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)
12
13# Writer agent
14writer = Agent(
15    role="Content Writer",
16    goal="Create engaging and educational content about technology",
17    backstory="""You are a talented technical writer with a passion for
18    translating complex concepts into simple language. Your articles
19    are always well-organized and enjoyable to read.""",
20    verbose=True,
21    allow_delegation=True,
22    llm=llm
23)
24
25# Editor agent
26editor = Agent(
27    role="Senior Editor",
28    goal="Ensure the highest quality of published content",
29    backstory="""You are an experienced editor with an eye for detail.
30    You care about style consistency, factual accuracy, and clarity of message.
31    You are not afraid to suggest changes that will improve the text.""",
32    verbose=True,
33    allow_delegation=False,
34    llm=llm
35)

Defining Tasks

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)
15
16# Writing task
17writing_task = Task(
18    description="""Based on the research report, write a blog article.
19
20    Requirements:
21    1. Engaging introduction
22    2. Clear structure with headings
23    3. Practical examples
24    4. Summary with key takeaways
25    5. Length: 1000-1500 words
26
27    Use accessible language for technical readers.""",
28    expected_output="Complete blog article ready for publication",
29    agent=writer,
30    context=[research_task]  # Uses the result of the previous task
31)
32
33# Editing task
34editing_task = Task(
35    description="""Review and edit the article.
36
37    Check:
38    1. Grammar and style correctness
39    2. Argument consistency
40    3. Message clarity
41    4. Engaging tone
42
43    Make necessary corrections.""",
44    expected_output="Edited, final article",
45    agent=editor,
46    context=[writing_task]
47)

Creating and Running a Crew

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)

Tools for Agents

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)
22
23# Custom tool
24from crewai_tools import BaseTool
25from pydantic import BaseModel, Field
26
27class DatabaseQueryInput(BaseModel):
28    query: str = Field(description="SQL query to execute")
29
30class DatabaseQueryTool(BaseTool):
31    name: str = "Database Query"
32    description: str = "Executes SQL queries on the database"
33    args_schema: type[BaseModel] = DatabaseQueryInput
34
35    def _run(self, query: str) -> str:
36        # Query implementation
37        return f"Query result: {query}"

Hierarchical Process

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)

Memory and Context

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)

Callbacks and Monitoring

1from crewai.callbacks import CrewCallback
2
3class ProgressCallback(CrewCallback):
4    """Callback for tracking progress."""
5
6    def on_task_start(self, task: Task):
7        print(f"Starting: {task.description[:50]}...")
8
9    def on_task_end(self, task: Task, output: str):
10        print(f"Completed: {task.description[:50]}")
11        print(f"   Output: {output[:100]}...")
12
13    def on_agent_action(self, agent: Agent, action: str):
14        print(f"{agent.role}: {action}")
15
16# Usage
17crew = Crew(
18    agents=[researcher, writer],
19    tasks=[research_task, writing_task],
20    callbacks=[ProgressCallback()],
21    verbose=True
22)

Asynchronous Execution

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())

Example: Content Pipeline

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()
29
30# Run
31article = create_content_pipeline("Machine Learning in 2025")

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 enterprise!