CrewAI is a cutting-edge framework for orchestrating role-playing, autonomous AI agents. It enables the creation of collaborative multi-agent systems where specialized agents work together as a cohesive team to accomplish complex tasks.
Each agent has a specific role, backstory, and goals that define their behavior and expertise
Agents work together, delegate tasks, and share information to achieve common objectives
Agents can use various tools like web search, file operations, and custom functions
Sequential, hierarchical, or custom process flows for different use cases
# Install CrewAI
pip install crewai
pip install 'crewai[tools]'
# Set up environment variables
export OPENAI_API_KEY="your-api-key-here"
# Optional: For other LLM providers
export ANTHROPIC_API_KEY="your-claude-key"
export GROQ_API_KEY="your-groq-key"
# Install additional tools
pip install langchain-community
pip install duckduckgo-search
pip install wikipedia-api
Agents are the building blocks of CrewAI. Each agent is an autonomous unit with specific attributes:
from crewai import 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.
You have a knack for dissecting complex data and presenting
actionable insights.""",
verbose=True,
allow_delegation=False,
tools=[search_tool, scrape_tool]
)
writer = Agent(
role='Content Strategist',
goal='Craft compelling content on tech advancements',
backstory="""You are a renowned Content Strategist,
known for your insightful and engaging articles.
You transform complex concepts into compelling narratives.""",
verbose=True,
allow_delegation=True,
tools=[writing_tool]
)
Tasks define what needs to be accomplished. They can be assigned to specific agents:
from crewai import Task
research_task = Task(
description="""Conduct a comprehensive analysis of the latest
AI advancements in 2024. Identify key trends, breakthrough
technologies, and potential industry impacts.""",
expected_output="Full analysis report in bullet points",
agent=researcher
)
writing_task = Task(
description="""Using the insights provided, develop an engaging
blog post that highlights the most significant AI advancements.
Make it accessible yet informative.""",
expected_output="A 750-word blog post",
agent=writer,
context=[research_task] # This task depends on research_task
)
A Crew represents a collaborative group of agents working together:
from crewai import Crew, Process
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, writing_task],
process=Process.sequential, # Tasks executed sequentially
verbose=2,
memory=True # Enable memory for better context retention
)
# Execute the crew
result = crew.kickoff()
print(result)
from crewai import Crew, Process
from langchain_openai import ChatOpenAI
# Create a manager LLM (can be different from agent LLMs)
manager_llm = ChatOpenAI(model="gpt-4", temperature=0.7)
# Create crew with hierarchical process
crew = Crew(
agents=[researcher, analyst, writer, reviewer],
tasks=[research_task, analysis_task, writing_task, review_task],
manager_llm=manager_llm,
process=Process.hierarchical,
verbose=True
)
# The manager will automatically delegate tasks to appropriate agents
result = crew.kickoff()
from crewai import Process
class CustomProcess(Process):
def execute(self, agents, tasks):
# Custom logic for task execution
results = []
# Parallel execution for research tasks
research_tasks = [t for t in tasks if "research" in t.description.lower()]
for task in research_tasks:
result = task.execute()
results.append(result)
# Sequential execution for dependent tasks
remaining_tasks = [t for t in tasks if t not in research_tasks]
for task in remaining_tasks:
task.context = results # Pass previous results as context
result = task.execute()
results.append(result)
return results
crew = Crew(
agents=agents,
tasks=tasks,
process=CustomProcess(),
verbose=True
)
from crewai_tools import (
SerperDevTool,
WebsiteSearchTool,
FileReadTool,
DirectoryReadTool,
CodeDocsSearchTool,
YoutubeVideoSearchTool
)
# Web search tool
search_tool = SerperDevTool()
# Website content tool
website_tool = WebsiteSearchTool(
website="https://example.com",
)
# File operations
file_tool = FileReadTool(file_path="./data/report.pdf")
# Code documentation search
docs_tool = CodeDocsSearchTool(
docs_url="https://docs.crewai.com/",
)
# Assign tools to agents
agent = Agent(
role="Research Specialist",
goal="Gather comprehensive information",
tools=[search_tool, website_tool, file_tool],
verbose=True
)
from crewai_tools import BaseTool
class CustomDatabaseTool(BaseTool):
name: str = "Database Query Tool"
description: str = "Query internal database for customer information"
def _run(self, query: str) -> str:
# Your custom database logic here
import sqlite3
conn = sqlite3.connect('customers.db')
cursor = conn.cursor()
cursor.execute(query)
results = cursor.fetchall()
conn.close()
return str(results)
# Create and use custom tool
db_tool = CustomDatabaseTool()
analyst = Agent(
role="Data Analyst",
goal="Analyze customer data",
tools=[db_tool],
verbose=True
)
A complete multi-agent system for researching, writing, and optimizing blog content.
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool
from langchain_openai import ChatOpenAI
# Initialize LLM
llm = ChatOpenAI(model="gpt-4", temperature=0.7)
# Tools
search_tool = SerperDevTool()
web_tool = WebsiteSearchTool()
# 1. Topic Researcher
topic_researcher = Agent(
role='Topic Research Specialist',
goal='Find trending topics and gather comprehensive information',
backstory='Expert in identifying viral content opportunities',
tools=[search_tool, web_tool],
llm=llm,
verbose=True
)
# 2. SEO Analyst
seo_analyst = Agent(
role='SEO Optimization Expert',
goal='Optimize content for search engines',
backstory='Specialist in keyword research and SEO best practices',
tools=[search_tool],
llm=llm,
verbose=True
)
# 3. Content Writer
content_writer = Agent(
role='Senior Content Writer',
goal='Create engaging and informative blog posts',
backstory='Award-winning writer with expertise in tech content',
llm=llm,
verbose=True
)
# 4. Editor
editor = Agent(
role='Content Editor',
goal='Polish and perfect the final content',
backstory='Meticulous editor ensuring quality and consistency',
llm=llm,
verbose=True
)
# Tasks
research_task = Task(
description="""Research the topic: 'Future of AI Agents in 2025'
Find the latest trends, statistics, and expert opinions.
Identify 3-5 key points to cover.""",
expected_output="Research summary with key points and sources",
agent=topic_researcher
)
seo_task = Task(
description="""Analyze SEO requirements for the blog post.
Identify primary and secondary keywords.
Suggest meta description and title tags.""",
expected_output="SEO optimization report",
agent=seo_analyst
)
writing_task = Task(
description="""Write a 1000-word blog post on 'Future of AI Agents in 2025'.
Include the research findings and SEO keywords naturally.
Make it engaging and informative.""",
expected_output="Complete blog post draft",
agent=content_writer,
context=[research_task, seo_task]
)
editing_task = Task(
description="""Review and edit the blog post.
Check for grammar, flow, and factual accuracy.
Ensure SEO guidelines are followed.""",
expected_output="Final polished blog post",
agent=editor,
context=[writing_task]
)
# Create and run the crew
blog_crew = Crew(
agents=[topic_researcher, seo_analyst, content_writer, editor],
tasks=[research_task, seo_task, writing_task, editing_task],
process=Process.sequential,
memory=True,
cache=True,
max_rpm=10,
verbose=True
)
# Execute
result = blog_crew.kickoff()
print("Final Blog Post:")
print(result)
from crewai.memory import ShortTermMemory, LongTermMemory, EntityMemory
# Configure crew with advanced memory
crew = Crew(
agents=[...],
tasks=[...],
memory=True, # Enable memory
memory_config={
"short_term": ShortTermMemory(),
"long_term": LongTermMemory(
storage_path="./crew_memory"
),
"entity": EntityMemory(
storage_path="./entity_memory"
)
},
embedder={
"provider": "openai",
"config": {
"model": "text-embedding-3-small"
}
}
)
# Memory persists across crew executions
result1 = crew.kickoff(inputs={"topic": "AI Safety"})
result2 = crew.kickoff(inputs={"topic": "AI Ethics"}) # Can reference previous context
# Optimized crew configuration
crew = Crew(
agents=agents,
tasks=tasks,
process=Process.sequential,
memory=True,
cache=True, # Cache results
max_rpm=10, # Rate limiting
max_iter=5, # Max iterations for tasks
step_callback=lambda step: print(f"Step: {step}"), # Progress tracking
task_callback=lambda task: print(f"Task completed: {task}"),
share_crew=True, # Share crew execution telemetry
output_log_file="crew_output.log" # Log outputs
)
# Async execution for better performance
import asyncio
async def run_crew_async():
result = await crew.kickoff_async()
return result
# Run multiple crews in parallel
async def run_multiple_crews():
crews = [crew1, crew2, crew3]
results = await asyncio.gather(*[c.kickoff_async() for c in crews])
return results
| Feature | CrewAI | AutoGen | LangChain Agents |
|---|---|---|---|
| Multi-Agent Focus | ✅ Core feature | ✅ Core feature | ⚠️ Limited |
| Role-Based Design | ✅ Excellent | ✅ Good | ⚠️ Basic |
| Learning Curve | Moderate | Steep | Moderate |
| Tool Integration | ✅ Extensive | ✅ Good | ✅ Excellent |
| Memory Management | ✅ Built-in | ⚠️ Basic | ✅ Flexible |
| Production Ready | ✅ Yes | ✅ Yes | ✅ Yes |