πŸŽ‰ 75% of content is free forever β€” Unlock Premium from $10/mo β†’
CW
Search courses…
πŸ’Ό Servicesℹ️ Aboutβœ‰οΈ ContactView Pricing Plansfrom $10

Building Multi-Agent Systems with CrewAI

AI AgentsMulti-Agent Systems🟒 Free Lesson

Advertisement

Building Multi-Agent Systems with CrewAI

Multi-Agent System ArchitectureOrchestratorResearcherWriterReviewerShared MemoryTask QueueTool RegistryCrewAI Execution Engine

What is Multi-Agent Systems?

Multi-agent systems coordinate multiple specialized agents to solve complex tasks that exceed single-agent capabilities. Each agent has distinct roles, tools, and expertise, collaborating through shared memory and message passing.

The key advantage is decomposition: complex problems are broken into manageable subproblems assigned to domain experts. A researcher gathers information, a writer synthesizes it, and a reviewer ensures quality. This mirrors human team dynamics and produces superior results for multifaceted tasks.

Multi-agent architectures include hierarchical (manager assigns to workers), sequential (pipeline of agents), and parallel (independent agents). CrewAI implements a flexible model supporting all three patterns with built-in delegation and task routing.

Project Overview

We will build a content research team with 4 specialized agents:

  • Researcher: Gathers information from multiple sources
  • Writer: Synthesizes findings into coherent articles
  • Editor: Reviews and improves quality
  • SEO Specialist: Optimizes for search engines

Expected outcome: A multi-agent system that produces publication-ready articles from a topic prompt.

Difficulty: Advanced (requires understanding of agent design, task decomposition, and inter-agent communication)

Architecture

CrewAI Multi-Agent PipelineResearch Agentweb_search, paper_searchWriter Agentcontent_generationEditor Agentgrammar_check, styleSEO Agentkeyword_optShared Context MemoryTask Dependency GraphFinal Output Pipeline

Tools & Setup

ToolVersionPurpose
Python3.11+Core language
CrewAI0.30+Multi-agent framework
OpenAI1.0+LLM backbone
duckduckgo-search4.0+Web search tool
pydantic2.0+Data validation

Step 1: Environment Setup

python -m venv venv
source venv/bin/activate
pip install crewai[tools] duckduckgo-search openai
export OPENAI_API_KEY="sk-your-key"

Step 2: Project Structure

Architecture Diagram
multi-agent/
β”œβ”€β”€ agents/
β”‚   β”œβ”€β”€ __init__.py
β”‚   β”œβ”€β”€ researcher.py
β”‚   β”œβ”€β”€ writer.py
β”‚   β”œβ”€β”€ editor.py
β”‚   └── seo_specialist.py
β”œβ”€β”€ tasks/
β”‚   β”œβ”€β”€ __init__.py
β”‚   └── content_tasks.py
β”œβ”€β”€ tools/
β”‚   β”œβ”€β”€ __init__.py
β”‚   └── search_tools.py
β”œβ”€β”€ crew.py
β”œβ”€β”€ config.py
└── main.py

Step 3: Custom Tools

# tools/search_tools.py
from crewai import Tool
from duckduckgo_search import DDGS
from typing import List, Dict

def search_web(query: str, num_results: int = 5) -> List[Dict]:
    with DDGS() as ddgs:
        results = list(ddgs.text(query, max_results=num_results))
        return [
            {"title": r["title"], "snippet": r["body"], "url": r["href"]}
            for r in results
        ]

def get_research_tool() -> Tool:
    def _search(query: str) -> str:
        results = search_web(query, num_results=5)
        formatted = []
        for i, r in enumerate(results):
            formatted.append(
                f"[{i+1}] {r['title']}\n{r['snippet']}\nURL: {r['url']}"
            )
        return "\n\n".join(formatted)

    return Tool(
        name="Web Search",
        description="Search the web for current information on any topic",
        func=_search,
    )

def get_article_analysis_tool() -> Tool:
    def _analyze(url: str) -> str:
        import httpx
        try:
            response = httpx.get(url, follow_redirects=True, timeout=10)
            from bs4 import BeautifulSoup
            soup = BeautifulSoup(response.text, "html.parser")
            text = soup.get_text(separator="\n", strip=True)
            return text[:3000]
        except Exception as e:
            return f"Error analyzing article: {str(e)}"

    return Tool(
        name="Article Analyzer",
        description="Extract and analyze content from a URL",
        func=_analyze,
    )

Step 4: Agent Definitions

# agents/researcher.py
from crewai import Agent, Tool
from tools.search_tools import get_research_tool, get_article_analysis_tool

def create_researcher() -> Agent:
    tools = [get_research_tool(), get_article_analysis_tool()]
    return Agent(
        role="Senior Research Analyst",
        goal="Gather comprehensive, accurate information on the given topic",
        backstory="""You are an expert researcher with a talent for finding 
        relevant information quickly. You verify multiple sources and identify 
        key insights, trends, and data points. You always cite your sources.""",
        tools=tools,
        verbose=True,
        allow_delegation=False,
        max_iter=15,
        memory=True,
    )

# agents/writer.py
from crewai import Agent

def create_writer() -> Agent:
    return Agent(
        role="Content Writer",
        goal="Transform research into engaging, well-structured articles",
        backstory="""You are a skilled writer who creates compelling content.
        You structure information logically, use clear language, and maintain
        reader engagement throughout. You can adapt tone for different audiences.""",
        tools=[],
        verbose=True,
        allow_delegation=False,
        max_iter=10,
        memory=True,
    )

# agents/editor.py
from crewai import Agent

def create_editor() -> Agent:
    return Agent(
        role="Senior Editor",
        goal="Ensure content quality, accuracy, and readability",
        backstory="""You are a meticulous editor with exceptional attention to 
        detail. You check for grammar, clarity, flow, and factual accuracy.
        You improve content without changing the author's voice.""",
        tools=[],
        verbose=True,
        allow_delegation=False,
        max_iter=10,
        memory=True,
    )

# agents/seo_specialist.py
from crewai import Agent

def create_seo_specialist() -> Agent:
    return Agent(
        role="SEO Specialist",
        goal="Optimize content for search engine visibility",
        backstory="""You are an SEO expert who understands search algorithms
        and user intent. You optimize titles, headings, meta descriptions,
        and content structure for maximum search visibility.""",
        tools=[],
        verbose=True,
        allow_delegation=False,
        max_iter=8,
        memory=True,
    )

Step 5: Task Definitions and Crew

# tasks/content_tasks.py
from crewai import Task

def create_research_task(agent, topic: str) -> Task:
    return Task(
        description=f"""Research the topic: "{topic}"

1. Search for 5-7 authoritative sources
2. Extract key facts, statistics, and insights
3. Identify different perspectives and debates
4. Note any recent developments or trends
5. Compile a structured research summary with citations

Output format:
- Key Findings (5-7 bullet points)
- Supporting Evidence (with source URLs)
- Different Perspectives
- Recent Developments
- Conclusion""",
        expected_output="Structured research summary with citations",
        agent=agent,
    )

def create_writing_task(agent, topic: str) -> Task:
    return Task(
        description=f"""Write a comprehensive article on: "{topic}"

Using the research provided:
1. Create an engaging introduction with a hook
2. Structure content with clear H2/H3 headings
3. Include relevant examples and data points
4. Write 1500-2000 words
5. End with a strong conclusion and call to action

Style guidelines:
- Professional but accessible tone
- Short paragraphs (2-3 sentences)
- Use bullet points for lists
- Include transition sentences between sections""",
        expected_output="Complete 1500-2000 word article",
        agent=agent,
    )

def create_editing_task(agent) -> Task:
    return Task(
        description="""Review and edit the article for:

1. Grammar and spelling errors
2. Clarity and readability
3. Logical flow and structure
4. Factual accuracy (flag any unsupported claims)
5. Tone consistency
6. Engagement level

Provide specific improvements and a cleaned-up version.""",
        expected_output="Edited article with improvements noted",
        agent=agent,
    )

def create_seo_task(agent) -> Task:
    return Task(
        description="""Optimize the article for SEO:

1. Suggest primary keyword and 3-5 secondary keywords
2. Optimize title tag (under 60 characters)
3. Write meta description (under 160 characters)
4. Ensure proper heading hierarchy (H1, H2, H3)
5. Add internal linking suggestions
6. Optimize image alt text suggestions
7. Check keyword density (1-2%)

Provide the optimized version with SEO changes highlighted.""",
        expected_output="SEO-optimized article with metadata",
        agent=agent,
    )

# crew.py
from crewai import Crew, Process
from agents.researcher import create_researcher
from agents.writer import create_writer
from agents.editor import create_editor
from agents.seo_specialist import create_seo_specialist
from tasks.content_tasks import (
    create_research_task,
    create_writing_task,
    create_editing_task,
    create_seo_task,
)

class ContentCrew:
    def __init__(self, topic: str):
        self.topic = topic
        self.researcher = create_researcher()
        self.writer = create_writer()
        self.editor = create_editor()
        self.seo_specialist = create_seo_specialist()

    def run(self) -> dict:
        research_task = create_research_task(self.researcher, self.topic)
        writing_task = create_writing_task(self.writer, self.topic)
        editing_task = create_editing_task(self.editor)
        seo_task = create_seo_task(self.seo_specialist)

        crew = Crew(
            agents=[
                self.researcher,
                self.writer,
                self.editor,
                self.seo_specialist,
            ],
            tasks=[
                research_task,
                writing_task,
                editing_task,
                seo_task,
            ],
            process=Process.sequential,
            verbose=True,
            memory=True,
        )

        result = crew.kickoff()
        return {
            "topic": self.topic,
            "output": result,
            "usage_metrics": crew.usage_metrics,
        }

Mathematical Foundation

Task Dependency Scoring:

Where each parameter means:

  • , , β€” weight coefficients for scoring factors
  • β€” estimated difficulty (1-10)
  • β€” business importance (1-10)
  • β€” number of dependent tasks (higher = higher priority)

Intuition: Tasks are scored to determine execution order. High-priority, low-dependency tasks execute first, ensuring optimal parallelism.

Agent Utilization:

Intuition: Measures what percentage of time each agent is actively working vs idle. Higher utilization indicates better task assignment.

Advanced Features

# Hierarchical multi-agent with manager
from crewai import Agent, Crew, Process

def create_hierarchical_crew():
    manager = Agent(
        role="Project Manager",
        goal="Coordinate the team to deliver high-quality output",
        backstory="You are an experienced PM who delegates effectively.",
        allow_delegation=True,
        verbose=True,
    )

    researcher = create_researcher()
    writer = create_writer()
    editor = create_editor()

    return Crew(
        agents=[manager, researcher, writer, editor],
        tasks=[],
        process=Process.hierarchical,
        manager_agent=manager,
        verbose=True,
    )

Testing & Evaluation

import pytest
from crew import ContentCrew

def test_crew_initialization():
    crew = ContentCrew(topic="AI agents")
    assert crew.topic == "AI agents"
    assert crew.researcher is not None

def test_crew_run():
    crew = ContentCrew(topic="What are AI agents?")
    result = crew.run()
    assert "output" in result
    assert len(result["output"]) > 0

Performance Metrics

MetricValueNotes
Articles/Hour4-6With GPT-4 backbone
Avg Quality Score8.5/10Human evaluation
Task Completion95%+Successful crew runs
Agent Utilization70%+Well-balanced tasks
Cost per Article$0.50-2.00GPT-4 pricing

Deployment

# main.py
from crew import ContentCrew

def main():
    topic = input("Enter article topic: ")
    crew = ContentCrew(topic=topic)
    result = crew.run()

    print("\n" + "=" * 60)
    print("ARTICLE OUTPUT")
    print("=" * 60)
    print(result["output"])
    print(f"\nUsage: {result['usage_metrics']}")

if __name__ == "__main__":
    main()

Real-World Use Cases

  • Content Marketing: Automated blog post and article creation
  • Market Research: Multi-source competitive analysis
  • Report Generation: Financial and business reports
  • Documentation: Technical documentation creation
  • Social Media: Content creation and scheduling teams

Common Pitfalls & Solutions

PitfallSolution
Agent confusionDefine clear roles and backstories
Circular delegationSet allow_delegation=False for leaf agents
Context lossUse shared memory and task outputs
High API costsUse GPT-3.5 for simpler agents
Quality inconsistencyAdd validation steps between agents

Summary with Key Takeaways

  • Multi-agent systems decompose complex tasks across specialized agents
  • CrewAI provides built-in delegation, memory, and process orchestration
  • Sequential processes work well for content pipelines; hierarchical for complex projects
  • Agent backstories significantly impact output quality and style
  • Always include review/validation agents to catch errors before final output

Need Expert AI Agents Help?

Get personalized tutoring, project support, or professional consulting.

Advertisement