CrewAI

Expert guidance for building multi-agent AI systems.

Installation

pip install crewai crewai-tools

Basic Concepts

Agents

from crewai import Agent, LLM

Define LLM

llm = LLM( model="openai/gpt-4o", temperature=0.7 )

Create agent

researcher = Agent( role="Senior Research Analyst", goal="Uncover cutting-edge developments in AI and data science", backstory="""You work at a leading tech think tank. Your expertise lies in identifying emerging trends and technologies. You have a knack for dissecting complex data and presenting actionable insights.""", verbose=True, allow_delegation=False, llm=llm, tools=[search_tool, scrape_tool] )

writer = Agent( role="Tech Content Writer", goal="Write compelling content about AI advancements", backstory="""You are a renowned content writer specializing in technology and AI. You transform complex concepts into engaging narratives that educate and inspire.""", verbose=True, allow_delegation=True, llm=llm )

Tasks

from crewai import Task

research_task = Task( description="""Conduct comprehensive research on the latest developments in AI agents technology in 2024. Focus on identifying key trends, breakthrough technologies, and potential industry impacts.""", expected_output="""A detailed 3-paragraph report covering: 1. Main AI agent frameworks and their capabilities 2. Key use cases and applications 3. Future predictions and challenges""", agent=researcher, tools=[search_tool] )

write_task = Task( description="""Using the research findings, write a blog post about AI agents that is engaging and accessible to a technical audience. The post should be informative yet engaging.""", expected_output="A 4-paragraph blog post in markdown format", agent=writer, context=[research_task] # Depends on research task )

Crew

from crewai import Crew, Process

crew = Crew( agents=[researcher, writer], tasks=[research_task, write_task], process=Process.sequential, # or Process.hierarchical verbose=True )

Run the crew

result = crew.kickoff() print(result)

Tools

Built-in Tools

from crewai_tools import ( SerperDevTool, ScrapeWebsiteTool, FileReadTool, DirectoryReadTool, CodeInterpreterTool, PDFSearchTool, YoutubeVideoSearchTool )

Search tool

search_tool = SerperDevTool()

Web scraping

scrape_tool = ScrapeWebsiteTool()

File operations

file_tool = FileReadTool(file_path="data.txt") dir_tool = DirectoryReadTool(directory="./documents")

Code execution

code_tool = CodeInterpreterTool()

PDF search

pdf_tool = PDFSearchTool(pdf="report.pdf")

Custom Tools

from crewai.tools import BaseTool from pydantic import Field

class CalculatorTool(BaseTool): name: str = "Calculator" description: str = "Useful for performing mathematical calculations"

def _run(self, expression: str) -> str:
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"Error: {str(e)}"

class DatabaseTool(BaseTool): name: str = "Database Query" description: str = "Query the database for information" connection_string: str = Field(default="")

def _run(self, query: str) -> str:
    # Execute database query
    return execute_query(self.connection_string, query)

Use tools

calculator = CalculatorTool() db_tool = DatabaseTool(connection_string="postgresql://...")

Function-based Tools

from crewai.tools import tool

@tool("Weather Lookup") def weather_tool(city: str) -> str: """Look up current weather for a city.""" # API call to weather service return f"Weather in {city}: 72°F, Sunny"

@tool("Stock Price") def stock_tool(symbol: str) -> str: """Get current stock price for a symbol.""" # API call to stock service return f"{symbol}: $150.00"

Advanced Patterns

Hierarchical Process

from crewai import Crew, Process, Agent

manager = Agent( role="Project Manager", goal="Coordinate the team to deliver high-quality results", backstory="Experienced manager who ensures efficient collaboration", allow_delegation=True )

crew = Crew( agents=[researcher, writer, editor], tasks=[research_task, write_task, edit_task], process=Process.hierarchical, manager_agent=manager, # Or use manager_llm verbose=True )

Task Callbacks

def task_callback(output): print(f"Task completed with output: {output.raw}") # Save to database, send notification, etc.

task = Task( description="Research AI trends", expected_output="Research report", agent=researcher, callback=task_callback )

Memory

from crewai import Crew

crew = Crew( agents=[agent1, agent2], tasks=[task1, task2], memory=True, # Enable memory embedder={ "provider": "openai", "config": {"model": "text-embedding-3-small"} } )

Async Execution

import asyncio from crewai import Crew

async def run_crew(): crew = Crew( agents=[researcher, writer], tasks=[research_task, write_task], verbose=True )

result = await crew.kickoff_async()
return result

Run

result = asyncio.run(run_crew())

Configuration

YAML Configuration

agents.yaml

researcher: role: "Senior Research Analyst" goal: "Uncover cutting-edge developments" backstory: "Expert at identifying trends" tools: - search_tool - scrape_tool

writer: role: "Tech Content Writer" goal: "Write compelling content" backstory: "Renowned content writer"

tasks.yaml

research_task: description: "Research latest AI developments" expected_output: "Detailed research report" agent: researcher

write_task: description: "Write blog post from research" expected_output: "Blog post in markdown" agent: writer context: - research_task

from crewai import Crew from crewai.project import CrewBase, agent, crew, task

@CrewBase class MyCrew: agents_config = "config/agents.yaml" tasks_config = "config/tasks.yaml"

@agent
def researcher(self) -> Agent:
    return Agent(config=self.agents_config["researcher"])

@task
def research_task(self) -> Task:
    return Task(config=self.tasks_config["research_task"])

@crew
def crew(self) -> Crew:
    return Crew(
        agents=self.agents,
        tasks=self.tasks,
        process=Process.sequential
    )

LLM Configuration

from crewai import LLM

OpenAI

openai_llm = LLM(model="openai/gpt-4o", temperature=0.7)

Anthropic

claude_llm = LLM(model="anthropic/claude-3-5-sonnet-20241022")

Ollama (local)

ollama_llm = LLM( model="ollama/llama3.1", base_url="http://localhost:11434" )

Azure OpenAI

azure_llm = LLM( model="azure/gpt-4o", api_key="your-api-key", base_url="https://your-resource.openai.azure.com" )

Example: Research Team

from crewai import Agent, Task, Crew, Process from crewai_tools import SerperDevTool, ScrapeWebsiteTool

Tools

search = SerperDevTool() scrape = ScrapeWebsiteTool()

Agents

researcher = Agent( role="Research Analyst", goal="Find and analyze information", backstory="Expert researcher with attention to detail", tools=[search, scrape], verbose=True )

analyst = Agent( role="Data Analyst", goal="Analyze data and extract insights", backstory="Expert in data analysis and visualization", verbose=True )

writer = Agent( role="Report Writer", goal="Create comprehensive reports", backstory="Skilled technical writer", verbose=True )

Tasks

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

analyze = Task( description="Analyze the research findings", expected_output="Key insights and trends", agent=analyst, context=[research] )

report = Task( description="Write final report", expected_output="Professional report document", agent=writer, context=[research, analyze] )

Crew

crew = Crew( agents=[researcher, analyst, writer], tasks=[research, analyze, report], process=Process.sequential, verbose=True )

Execute

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

Resources

• CrewAI Documentation

• CrewAI GitHub

• CrewAI Tools