Multi-Agent Crew Template

Agent

Production-ready template for building multi-agent systems with CrewAI, with role-based agents and task delegation.

Multi-Agent Crew Template

Overview

A production-ready template for building multi-agent systems with CrewAI. This template provides a complete, configurable foundation for creating collaborative AI agent teams that can work together to solve complex tasks.

Features

  • Role-based agent definition: Clear roles, goals, and backstories
  • Task delegation: Automatic task assignment based on agent capabilities
  • Process management: Sequential, hierarchical, or consensual processes
  • Tool integration: Built-in support for custom and pre-built tools
  • Memory and context: Agent memory and context sharing
  • Output validation: Structured output with Pydantic models
  • Error handling: Graceful error recovery and retry logic
  • Logging and monitoring: Comprehensive logging for debugging

Installation

pip install crewai crewai-tools

Template Structure

multi-agent-crew-template/
├── config/
│   ├── agents.yaml        # Agent definitions
│   ├── tasks.yaml         # Task definitions
│   └── crew.yaml          # Crew configuration
├── agents/
│   ├── __init__.py
│   ├── researcher.py      # Research agent
│   ├── writer.py          # Writing agent
│   ├── editor.py          # Editing agent
│   └── manager.py         # Manager agent
├── tools/
│   ├── __init__.py
│   ├── search_tool.py     # Web search tool
│   ├── file_tool.py       # File operations tool
│   └── custom_tool.py     # Custom tools
├── tasks/
│   ├── __init__.py
│   ├── research_tasks.py
│   ├── writing_tasks.py
│   └── editing_tasks.py
├── crew/
│   ├── __init__.py
│   └── my_crew.py         # Crew assembly
├── outputs/
│   └── results/           # Output directory
├── tests/
│   ├── test_agents.py
│   └── test_crew.py
├── main.py                # Entry point
└── requirements.txt

Quick Start

1. Define Agents

# config/agents.yaml
researcher:
  role: >
    Senior Research Analyst
  goal: >
    Uncover cutting-edge developments in {topic}
  backstory: >
    You're a senior research analyst with deep expertise in {topic}.
    You have a knack for identifying emerging trends and understanding
    complex technical concepts. Your work is used by industry leaders
    to make strategic decisions.
  verbose: true
  allow_delegation: false

writer:
  role: >
    Technical Content Writer
  goal: >
    Create compelling, accurate content about {topic}
  backstory: >
    You're a technical content writer specializing in {topic}.
    You excel at translating complex technical concepts into
    clear, engaging content that resonates with both technical
    and non-technical audiences.
  verbose: true
  allow_delegation: true

editor:
  role: >
    Content Editor and Quality Assurance
  goal: >
    Ensure content accuracy, clarity, and quality
  backstory: >
    You're a meticulous content editor with a keen eye for detail.
    You ensure all content meets the highest standards of accuracy,
    clarity, and engagement before publication.
  verbose: true
  allow_delegation: false

2. Define Tasks

# config/tasks.yaml
research_task:
  description: >
    Conduct comprehensive research on {topic}.
    Find the latest developments, key players, trends, and challenges.
    Gather information from at least 5 authoritative sources.
  expected_output: >
    A detailed research report with:
    - Executive summary
    - Key findings (at least 10)
    - Source citations
    - Identified trends and patterns
  agent: researcher

writing_task:
  description: >
    Write a comprehensive article about {topic} based on the research.
    Target audience: {audience}
    Tone: {tone}
    Length: approximately 2000 words
  expected_output: >
    A well-structured article with:
    - Engaging introduction
    - Clear sections with headings
    - Supporting evidence and examples
    - Actionable insights
    - Conclusion with next steps
  agent: writer

editing_task:
  description: >
    Review and edit the article for accuracy, clarity, and quality.
    Check for factual errors, improve flow, and ensure consistency.
  expected_output: >
    A polished, publication-ready article with:
    - All factual claims verified
    - Improved clarity and flow
    - Consistent tone and style
    - Proper formatting and structure
  agent: editor

3. Assemble Crew

# crew/my_crew.py
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
import yaml

# Load configuration
with open("config/agents.yaml") as f:
    agents_config = yaml.safe_load(f)

with open("config/tasks.yaml") as f:
    tasks_config = yaml.safe_load(f)

# Create agents
researcher = Agent(
    config=agents_config["researcher"],
    llm=ChatOpenAI(model="gpt-4o"),
    tools=[search_tool],
    verbose=True
)

writer = Agent(
    config=agents_config["writer"],
    llm=ChatOpenAI(model="gpt-4o"),
    verbose=True
)

editor = Agent(
    config=agents_config["editor"],
    llm=ChatOpenAI(model="gpt-4o"),
    verbose=True
)

# Create tasks
research_task = Task(
    config=tasks_config["research_task"],
    agent=researcher,
    output_file="outputs/research_report.md"
)

writing_task = Task(
    config=tasks_config["writing_task"],
    agent=writer,
    context=[research_task],
    output_file="outputs/article_draft.md"
)

editing_task = Task(
    config=tasks_config["editing_task"],
    agent=editor,
    context=[writing_task],
    output_file="outputs/final_article.md"
)

# Create crew
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential,
    verbose=True,
    memory=True,
    cache=True
)

4. Run Crew

# main.py
from crew.my_crew import crew

result = crew.kickoff(inputs={
    "topic": "AI agent frameworks",
    "audience": "technical decision-makers",
    "tone": "professional yet accessible"
})

print(result)

Advanced Features

Hierarchical Process

from crewai import Process

crew = Crew(
    agents=[researcher, writer, editor, manager],
    tasks=[research_task, writing_task, editing_task],
    process=Process.hierarchical,
    manager_llm=ChatOpenAI(model="gpt-4o"),
    verbose=True
)

Custom Tools

from crewai_tools import tool

@tool
def fetch_webpage(url: str) -> str:
    """Fetch and return the content of a webpage."""
    import requests
    response = requests.get(url)
    return response.text

@tool
def calculate_metrics(data: dict) -> dict:
    """Calculate metrics from provided data."""
    # Custom calculation logic
    return {"total": sum(data.values()), "average": sum(data.values()) / len(data)}

# Add tools to agent
researcher = Agent(
    config=agents_config["researcher"],
    tools=[fetch_webpage, calculate_metrics]
)

Structured Output

from pydantic import BaseModel, Field

class ResearchOutput(BaseModel):
    title: str = Field(description="Research report title")
    summary: str = Field(description="Executive summary")
    key_findings: list[str] = Field(description="List of key findings")
    sources: list[str] = Field(description="List of source URLs")
    confidence: float = Field(description="Confidence score 0-1")

# Use with task
research_task = Task(
    config=tasks_config["research_task"],
    agent=researcher,
    output_json=ResearchOutput,
    output_file="outputs/research.json"
)

Agent Memory

from crewai import Agent
from langchain_openai import ChatOpenAI

researcher = Agent(
    config=agents_config["researcher"],
    llm=ChatOpenAI(model="gpt-4o"),
    memory=True,  # Enable memory
    cache=True,   # Enable caching
    verbose=True
)

Error Handling

from crewai import Task
from crewai.task import TaskOutput

def error_callback(error: Exception, task: Task):
    print(f"Task {task.description} failed: {error}")
    # Implement retry logic or fallback

research_task = Task(
    config=tasks_config["research_task"],
    agent=researcher,
    callback=error_callback
)

Parallel Execution

from crewai import Crew, Process

# For independent tasks, use consensual process
crew = Crew(
    agents=[agent1, agent2, agent3],
    tasks=[task1, task2, task3],  # Independent tasks
    process=Process.consensual,
    verbose=True
)

Integration Examples

With LangChain Tools

from langchain_community.tools import DuckDuckGoSearchRun
from crewai_tools import LangChainTool

search_tool = LangChainTool(
    tool=DuckDuckGoSearchRun(),
    name="DuckDuckGo Search",
    description="Search the web for information"
)

researcher = Agent(
    config=agents_config["researcher"],
    tools=[search_tool]
)

With Custom LLM

from langchain_anthropic import ChatAnthropic

crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    llm=ChatAnthropic(model="claude-3-5-sonnet-20241022"),
    verbose=True
)

With External APIs

@tool
def get_weather(location: str) -> str:
    """Get current weather for a location."""
    import requests
    response = requests.get(
        f"https://api.open-meteo.com/v1/forecast",
        params={"latitude": 37.77, "longitude": -122.42}
    )
    return str(response.json())

writer = Agent(
    config=agents_config["writer"],
    tools=[get_weather]
)

Best Practices

Agent Design

  1. Clear roles: Each agent should have a distinct, non-overlapping role
  2. Specific goals: Goals should be measurable and actionable
  3. Appropriate backstories: Backstories should motivate the agent's behavior
  4. Right tools: Only give agents the tools they need
  5. Delegation: Enable delegation for complex tasks

Task Design

  1. Clear descriptions: Tasks should be unambiguous
  2. Expected outputs: Define what success looks like
  3. Appropriate context: Provide relevant context from previous tasks
  4. Output files: Specify where outputs should be saved
  5. Async options: Use async for I/O-bound tasks

Crew Design

  1. Process selection: Choose the right process for your workflow
  2. Memory usage: Enable memory for context retention
  3. Caching: Enable caching to save costs
  4. Verbose logging: Start verbose for debugging, reduce for production
  5. Error handling: Implement callbacks for error recovery

Troubleshooting

Agents Not Delegating

  1. Check allow_delegation is set to True
  2. Ensure manager agent has clear authority
  3. Verify task dependencies are correctly set

Poor Output Quality

  1. Improve agent backstories and goals
  2. Add more specific task descriptions
  3. Use structured output (Pydantic models)
  4. Enable memory for context retention
  5. Try a more capable LLM

Slow Execution

  1. Enable caching (cache=True)
  2. Use async execution for I/O tasks
  3. Reduce verbose logging
  4. Use smaller/cheaper models for simpler tasks
  5. Parallelize independent tasks

Resources


Last updated: May 2026

View on GitHub

Related Templates