CrewAI Tutorial: Build Multi-Agent AI Systems That Work Together

pip install crewai crewai-tools langchain-openai

# Set your API key
export OPENAI_API_KEY="sk-..."

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

from crewai import Agent, Task, Crew

# Agent 1: Researcher
researcher = Agent(
    role="Market Research Analyst",
    goal="Find accurate, current information about a given topic",
    backstory="""You are an experienced market research analyst with 10 years 
    of experience. You excel at finding reliable information, identifying trends,
    and presenting findings clearly. You always cite your sources.""",
    tools=[SerperDevTool()],  # Can search the web
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0.1),
    verbose=True
)

# Agent 2: Writer
writer = Agent(
    role="Content Writer",
    goal="Write engaging, accurate articles based on research findings",
    backstory="""You are a professional content writer who specializes in making 
    complex topics accessible. You write in a clear, engaging style and always 
    ensure the content is accurate and well-structured.""",
    tools=[],  # Writer doesn't need tools — works with provided research
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0.7),
    verbose=True
)

# Task 1: Research
research_task = Task(
    description="""Research the current state of {topic}.
    Find:
    1. Key developments in the last 12 months
    2. Major players and their recent actions
    3. Market trends and predictions
    4. Expert opinions
    
    Compile findings with sources.""",
    expected_output="A structured research report with bullet points and source URLs",
    agent=researcher
)

# Task 2: Write article (uses research output)
writing_task = Task(
    description="""Based on the research provided, write a 800-word article about {topic}.
    
    The article should:
    - Have an engaging introduction that hooks the reader
    - Present key findings clearly with concrete examples
    - Include expert perspectives
    - End with a forward-looking conclusion
    
    Use the research findings as your source material.""",
    expected_output="An 800-word article in markdown format",
    agent=writer,
    context=[research_task]  # Access to research output
)

# Crew: orchestrates the workflow
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,  # Tasks run in order
    verbose=True
)

# Run
result = crew.kickoff(inputs={"topic": "AI agents in enterprise software"})
print(result.raw)

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

llm = ChatOpenAI(model="gpt-4o-mini")

# Agent definitions
researcher = Agent(
    role="Senior Research Analyst",
    goal="Conduct thorough research on topics to gather accurate, comprehensive information",
    backstory="Expert researcher with skills in finding reliable sources and synthesizing complex information",
    tools=[SerperDevTool(), ScrapeWebsiteTool()],
    llm=llm,
    max_iter=5,      # Max tool call iterations
    memory=True      # Remember across tasks
)

writer = Agent(
    role="Senior Content Writer",
    goal="Write engaging, SEO-optimized content based on research",
    backstory="Experienced writer who creates compelling content that ranks well and engages readers",
    tools=[],
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0.7),
    memory=True
)

editor = Agent(
    role="Content Editor",
    goal="Review and improve content for accuracy, clarity, and SEO",
    backstory="Expert editor with background in content marketing and SEO optimization",
    tools=[SerperDevTool()],  # Can verify facts
    llm=llm
)

# Tasks
research_task = Task(
    description="""Research '{keyword}' comprehensively:
    1. Search for recent articles, studies, and expert opinions
    2. Identify the key questions people ask about this topic
    3. Find statistics and data points that support key claims
    4. Note competitors ranking for this topic
    
    Provide a structured research brief.""",
    expected_output="Research brief with key findings, statistics, and sources",
    agent=researcher,
    output_file="research_output.md"  # Save intermediate output
)

writing_task = Task(
    description="""Write a comprehensive blog post about '{keyword}':
    - Target length: 1500-2000 words
    - Include H2 and H3 headers
    - Use the provided research findings
    - Write for both humans and search engines
    - Include a FAQ section
    - End with a clear call-to-action""",
    expected_output="Full blog post in markdown format",
    agent=writer,
    context=[research_task]
)

editing_task = Task(
    description="""Review and improve the blog post:
    1. Check all factual claims against the research
    2. Improve clarity and flow
    3. Ensure keyword '{keyword}' appears naturally
    4. Verify all headings follow H2/H3 structure
    5. Check that the FAQ section is relevant
    6. Rate the content quality 1-10 with specific improvements
    
    Return the improved article.""",
    expected_output="Edited article with quality score and change notes",
    agent=editor,
    context=[research_task, writing_task]
)

# Assemble crew
content_crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential,
    verbose=True,
    memory=True,         # Enable crew memory
    embedder={           # For memory storage
        "provider": "openai",
        "config": {"model": "text-embedding-3-small"}
    }
)

# Run with specific keyword
result = content_crew.kickoff(inputs={"keyword": "best practices for code review"})
print(result.raw)

# Run agents in parallel for independent tasks

# Multiple researchers working simultaneously
research_ai_task = Task(
    description="Research AI developments in healthcare in 2026",
    expected_output="List of 10 major developments with brief descriptions",
    agent=researcher
)

research_finance_task = Task(
    description="Research AI developments in financial services in 2026",
    expected_output="List of 10 major developments with brief descriptions",
    agent=researcher  # Same agent can run in parallel tasks
)

synthesis_task = Task(
    description="""Synthesize the research from both healthcare and finance AI reports.
    Create a comparative analysis: What themes appear in both? What's unique to each?
    What does this tell us about AI adoption patterns?""",
    expected_output="A 500-word comparative analysis with key insights",
    agent=writer,
    context=[research_ai_task, research_finance_task]
)

parallel_crew = Crew(
    agents=[researcher, writer],
    tasks=[research_ai_task, research_finance_task, synthesis_task],
    process=Process.sequential  # Use sequential for now
    # Note: True parallel execution requires: process=Process.hierarchical
    # and a manager_agent that coordinates parallelism
)

from crewai.tools import BaseTool
from pydantic import BaseModel, Field

class DatabaseQueryInput(BaseModel):
    query: str = Field(description="SQL query to execute")
    database: str = Field(description="Database name", default="production")

class DatabaseTool(BaseTool):
    name: str = "Database Query Tool"
    description: str = "Execute SQL queries against the company database"
    args_schema: type[BaseModel] = DatabaseQueryInput
    
    def _run(self, query: str, database: str = "production") -> str:
        """Execute a database query and return results."""
        # In production: connect to your actual database
        # For demo: return mock results
        return f"Query results for '{query}' on {database}: [10 rows returned]"

# Simple function-based tool
from crewai.tools import tool

@tool("Calculate Metrics Tool")
def calculate_metrics(data: str) -> str:
    """Calculate business metrics from provided data string."""
    # Process the data and return metrics
    return f"Metrics calculated: conversion_rate=3.2%, avg_revenue=$45.50"

# Assign custom tools to agents
data_analyst = Agent(
    role="Data Analyst",
    goal="Analyze company data to identify trends and opportunities",
    backstory="Expert data analyst with SQL and business intelligence skills",
    tools=[DatabaseTool(), calculate_metrics],
    llm=llm
)

# Manager agent coordinates the crew

manager = Agent(
    role="Project Manager",
    goal="Coordinate the team to deliver high-quality research reports efficiently",
    backstory="Experienced project manager who excels at breaking down complex projects and delegating appropriately",
    llm=ChatOpenAI(model="gpt-4o"),  # Use more capable model for manager
    allow_delegation=True   # Can delegate to other agents
)

hierarchical_crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.hierarchical,
    manager_agent=manager,  # Manager orchestrates workflow
    verbose=True
)

# Manager decides:
# - Which agent handles which task
# - When to retry failed tasks
# - How to handle ambiguous situations
# - When the final output meets quality standards