CrewAI Agent Examples: 6 Multi-Agent Crews

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

search_tool = SerperDevTool()
web_tool = WebsiteSearchTool()

researcher = Agent(
    role="Senior Content Researcher",
    goal="Find accurate, current information on {topic} from authoritative sources",
    backstory="You are an expert researcher who finds and evaluates sources critically.",
    tools=[search_tool, web_tool],
    verbose=True,
    llm="gpt-4o-mini"
)

analyst = Agent(
    role="Content Strategist",
    goal="Synthesize research into a structured content outline with key insights",
    backstory="You turn raw research into clear narrative structures for writers.",
    verbose=True,
    llm="gpt-4o-mini"
)

writer = Agent(
    role="SEO Content Writer",
    goal="Write a 1500-word article optimized for search based on the outline",
    backstory="You write engaging, well-structured articles that rank in search engines.",
    verbose=True,
    llm="gpt-4o"
)

research_task = Task(
    description="Research the topic: {topic}. Find 5 authoritative sources.",
    expected_output="A list of 5 sources with key findings from each.",
    agent=researcher
)

outline_task = Task(
    description="Create a detailed article outline based on the research findings.",
    expected_output="A structured H2/H3 outline with bullet points for each section.",
    agent=analyst,
    context=[research_task]
)

writing_task = Task(
    description="Write the full article following the outline. Include an introduction and conclusion.",
    expected_output="A 1500-word article in markdown format ready for publication.",
    agent=writer,
    context=[outline_task]
)

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, outline_task, writing_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff(inputs={"topic": "AI agent frameworks in 2025"})

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

scraper = ScrapeWebsiteTool()

scout = Agent(
    role="Competitive Intelligence Scout",
    goal="Scrape and collect recent updates from competitor websites: {competitors}",
    backstory="You methodically collect data from competitor sites without missing key changes.",
    tools=[scraper, WebsiteSearchTool()],
    llm="gpt-4o-mini"
)

analyst = Agent(
    role="Market Intelligence Analyst",
    goal="Identify strategic signals from competitor activity data",
    backstory="You detect pricing shifts, feature launches, and positioning changes that matter.",
    llm="gpt-4o"
)

reporter = Agent(
    role="Executive Communications Writer",
    goal="Produce a concise, actionable competitive intelligence brief",
    backstory="You distill complex competitive data into clear executive summaries.",
    llm="gpt-4o-mini"
)

collect_task = Task(
    description="Visit each competitor URL in {competitors} and collect: pricing changes, new features, blog posts from the last 7 days.",
    expected_output="Structured JSON with competitor name, URL, and list of changes.",
    agent=scout
)

analyze_task = Task(
    description="Analyze the collected data for strategic patterns, pricing moves, and positioning shifts.",
    expected_output="Analysis with 3-5 key strategic signals and their implications.",
    agent=analyst,
    context=[collect_task]
)

brief_task = Task(
    description="Write a one-page competitive intelligence brief suitable for a VP of Product.",
    expected_output="A markdown brief with sections: Key Moves, Threats, Opportunities, Recommended Actions.",
    agent=reporter,
    context=[analyze_task]
)

crew = Crew(
    agents=[scout, analyst, reporter],
    tasks=[collect_task, analyze_task, brief_task],
    process=Process.sequential
)

result = crew.kickoff(inputs={"competitors": ["competitor-a.com/pricing", "competitor-b.com/blog"]})

from crewai import Agent, Task, Crew, Process
from crewai_tools import CodeInterpreterTool

code_tool = CodeInterpreterTool()

security_reviewer = Agent(
    role="Security Engineer",
    goal="Identify security vulnerabilities in the provided code",
    backstory="You specialize in OWASP Top 10, injection attacks, and secrets leakage.",
    tools=[code_tool],
    llm="gpt-4o"
)

performance_reviewer = Agent(
    role="Performance Engineer",
    goal="Identify performance bottlenecks and inefficient patterns in the code",
    backstory="You optimize for time complexity, memory usage, and I/O efficiency.",
    tools=[code_tool],
    llm="gpt-4o"
)

style_reviewer = Agent(
    role="Code Quality Reviewer",
    goal="Evaluate code readability, maintainability, and adherence to best practices",
    backstory="You enforce clean code principles: DRY, single responsibility, clear naming.",
    llm="gpt-4o-mini"
)

lead_reviewer = Agent(
    role="Lead Code Reviewer",
    goal="Synthesize all review findings into a prioritized, actionable review report",
    backstory="You have 15 years of experience leading code reviews at top tech companies.",
    llm="gpt-4o"
)

# Tasks run in parallel, then synthesis runs last
security_task = Task(
    description="Review this code for security issues: {code}",
    expected_output="List of security findings with severity (Critical/High/Medium/Low) and line numbers.",
    agent=security_reviewer
)

performance_task = Task(
    description="Review this code for performance issues: {code}",
    expected_output="List of performance findings with estimated impact and suggested fixes.",
    agent=performance_reviewer
)

style_task = Task(
    description="Review this code for style and quality issues: {code}",
    expected_output="List of style issues with references to best practice guidelines.",
    agent=style_reviewer
)

synthesis_task = Task(
    description="Synthesize all review findings into a final pull request review.",
    expected_output="A structured PR review with sections: Must Fix, Should Fix, Suggestions, Approved/Rejected verdict.",
    agent=lead_reviewer,
    context=[security_task, performance_task, style_task]
)

crew = Crew(
    agents=[security_reviewer, performance_reviewer, style_reviewer, lead_reviewer],
    tasks=[security_task, performance_task, style_task, synthesis_task],
    process=Process.sequential  # CrewAI 0.30+ supports parallel task groups
)

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

researcher = Agent(
    role="Prospect Research Specialist",
    goal="Research the prospect company and contact: {prospect_name} at {company}",
    backstory="You find recent news, company challenges, and individual priorities quickly.",
    tools=[SerperDevTool()],
    llm="gpt-4o-mini"
)

pain_analyst = Agent(
    role="B2B Pain Point Analyst",
    goal="Identify the top 3 business pain points this prospect likely faces based on research",
    backstory="You translate company context into specific, relevant business challenges.",
    llm="gpt-4o"
)

copywriter = Agent(
    role="B2B Sales Copywriter",
    goal="Write a personalized cold email referencing specific research and pain points",
    backstory="You write concise, non-generic emails that get replies. No platitudes.",
    llm="gpt-4o"
)

research_task = Task(
    description="Research {prospect_name}, {title} at {company}. Find: recent company news, growth stage, known challenges.",
    expected_output="A 200-word prospect profile with company context and individual priorities.",
    agent=researcher
)

pain_task = Task(
    description="Based on the research, identify the 3 most relevant pain points for our {product_category} solution.",
    expected_output="3 pain points with a 1-sentence rationale for each.",
    agent=pain_analyst,
    context=[research_task]
)

email_task = Task(
    description="Write a 150-word cold email to {prospect_name}. Reference specific research. Propose a 20-minute call.",
    expected_output="Subject line + email body. Conversational tone. No buzzwords.",
    agent=copywriter,
    context=[research_task, pain_task]
)

crew = Crew(
    agents=[researcher, pain_analyst, copywriter],
    tasks=[research_task, pain_task, email_task],
    process=Process.sequential
)

result = crew.kickoff(inputs={
    "prospect_name": "Sarah Chen",
    "title": "VP of Engineering",
    "company": "Acme Corp",
    "product_category": "developer tooling"
})

from crewai import Agent, Task, Crew, Process
from crewai_tools import PDFSearchTool

pdf_tool = PDFSearchTool(pdf="{report_path}")

extractor = Agent(
    role="Financial Data Extractor",
    goal="Extract key financial metrics from the earnings report PDF",
    backstory="You extract structured data from unstructured financial documents precisely.",
    tools=[pdf_tool],
    llm="gpt-4o-mini"
)

analyst = Agent(
    role="Financial Analyst",
    goal="Interpret the extracted metrics: revenue growth, margins, guidance, vs. analyst expectations",
    backstory="You have 10 years of equity research experience. You focus on what moves the stock.",
    llm="gpt-4o"
)

risk_assessor = Agent(
    role="Risk Assessment Specialist",
    goal="Identify risks, red flags, and forward-looking concerns from the report language",
    backstory="You read between the lines of management commentary to spot warning signs.",
    llm="gpt-4o"
)

writer = Agent(
    role="Investment Research Writer",
    goal="Produce a concise investment summary with a clear Buy/Hold/Sell framework",
    backstory="You write institutional-quality research notes that are clear and actionable.",
    llm="gpt-4o"
)

crew = Crew(
    agents=[extractor, analyst, risk_assessor, writer],
    tasks=[
        Task(description="Extract revenue, EPS, gross margin, guidance from {report_path}.",
             expected_output="Structured JSON with financial metrics.", agent=extractor),
        Task(description="Analyze the metrics vs. prior quarter and analyst consensus.",
             expected_output="Narrative analysis with beat/miss assessment.", agent=analyst),
        Task(description="Identify 3-5 risk factors from the MD&A section language.",
             expected_output="Risk list with likelihood and impact ratings.", agent=risk_assessor),
        Task(description="Write a 400-word investment summary with recommendation.",
             expected_output="Research note with thesis, risks, and Buy/Hold/Sell.", agent=writer)
    ],
    process=Process.sequential
)

from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

manager_llm = ChatOpenAI(model="gpt-4o")

manager = Agent(
    role="Customer Success Manager",
    goal="Route and resolve customer tickets by delegating to the right specialist",
    backstory="You assess tickets, delegate to specialists, and ensure customers get complete answers.",
    llm="gpt-4o",
    allow_delegation=True
)

technical = Agent(
    role="Technical Support Specialist",
    goal="Resolve technical issues, API errors, and integration problems",
    backstory="You have deep product knowledge and can debug technical configurations.",
    llm="gpt-4o-mini",
    allow_delegation=False
)

billing = Agent(
    role="Billing Specialist",
    goal="Handle billing questions, invoices, upgrades, and refund requests",
    backstory="You know the billing system inside out and have authority to issue credits.",
    llm="gpt-4o-mini",
    allow_delegation=False
)

onboarding = Agent(
    role="Onboarding Specialist",
    goal="Guide new customers through setup, best practices, and first-value milestones",
    backstory="You help customers get productive quickly with personalized guidance.",
    llm="gpt-4o-mini",
    allow_delegation=False
)

resolution_task = Task(
    description="Resolve this customer ticket: {ticket_content}. Ensure the customer gets a complete, accurate answer.",
    expected_output="A complete, professional customer response that resolves the issue.",
    agent=manager
)

crew = Crew(
    agents=[manager, technical, billing, onboarding],
    tasks=[resolution_task],
    process=Process.hierarchical,
    manager_llm=manager_llm,
    verbose=True
)

result = crew.kickoff(inputs={"ticket_content": "I can't connect my Salesforce integration and I'm being charged for a plan I didn't upgrade to."})