Multi-Agent Orchestration Template
AgentTemplate for building multi-agent systems with role-based specialization.
Multi-Agent Orchestration Template
Overview
This template provides a foundation for building multi-agent systems with role-based specialization. It demonstrates how to create agents with distinct roles, enable them to collaborate, and orchestrate complex workflows.
Multi-agent systems excel at tasks that benefit from specialization—different agents can focus on different aspects of a problem, leading to better results than a single generalist agent.
Prerequisites
- Python 3.10+
- CrewAI or LangGraph installed
- LLM API access
Project Structure
multi-agent-system/
├── agents/
│ ├── __init__.py
│ ├── researcher.py # Research agent definition
│ ├── writer.py # Writer agent definition
│ ├── reviewer.py # Reviewer agent definition
│ └── manager.py # Manager/orchestrator agent
├── tasks/
│ ├── __init__.py
│ ├── research_tasks.py
│ ├── writing_tasks.py
│ └── review_tasks.py
├── tools/
│ ├── __init__.py
│ ├── search_tools.py
│ └── file_tools.py
├── config/
│ └── agents.yaml # Agent configurations
│ └── tasks.yaml # Task definitions
├── main.py # Orchestration entry point
├── requirements.txt
└── README.md
Installation
pip install crewai langchain-openai pyyaml
Agent Definitions
Research Agent
agents/researcher.py:
from crewai import Agent
from langchain_openai import ChatOpenAI
class Researcher:
def __init__(self, llm=None):
self.llm = llm or ChatOpenAI(model="gpt-4o")
def create(self):
return Agent(
role="Senior Research Analyst",
goal="Uncover cutting-edge developments in AI and data science",
backstory="""You are a Senior Research Analyst at a leading tech
think tank. You excel at identifying emerging trends, analyzing
complex technical concepts, and synthesizing information from
multiple sources into actionable insights.""",
verbose=True,
allow_delegation=False,
llm=self.llm,
tools=[], # Add search tools here
)
Writer Agent
agents/writer.py:
from crewai import Agent
from langchain_openai import ChatOpenAI
class Writer:
def __init__(self, llm=None):
self.llm = llm or ChatOpenAI(model="gpt-4o")
def create(self):
return Agent(
role="Technical Content Writer",
goal="Create well-structured, engaging technical content",
backstory="""You are a Technical Content Writer specializing in
AI and data science. You excel at translating complex technical
concepts into clear, engaging content for various audiences.
You have a talent for structuring content logically and
making it accessible without losing technical accuracy.""",
verbose=True,
allow_delegation=False,
llm=self.llm,
)
Reviewer Agent
agents/reviewer.py:
from crewai import Agent
from langchain_openai import ChatOpenAI
class Reviewer:
def __init__(self, llm=None):
self.llm = llm or ChatOpenAI(model="gpt-4o")
def create(self):
return Agent(
role="Content Quality Reviewer",
goal="Ensure content accuracy, clarity, and quality",
backstory="""You are a meticulous Content Quality Reviewer with
deep technical expertise in AI and data science. You excel at
identifying factual errors, unclear explanations, and
structural issues. You provide constructive feedback and
ensure content meets high quality standards.""",
verbose=True,
allow_delegation=True, # Can delegate back to writer
llm=self.llm,
)
Manager Agent
agents/manager.py:
from crewai import Agent
from langchain_openai import ChatOpenAI
class Manager:
def __init__(self, llm=None):
self.llm = llm or ChatOpenAI(model="gpt-4o")
def create(self):
return Agent(
role="Project Manager",
goal="Orchestrate the content creation process efficiently",
backstory="""You are an experienced Project Manager who
specializes in coordinating multi-agent workflows. You excel
at breaking down complex projects into manageable tasks,
assigning them to the right agents, and ensuring smooth
collaboration between team members.""",
verbose=True,
allow_delegation=True,
llm=self.llm,
)
Task Definitions
tasks/research_tasks.py:
from crewai import Task
from textwrap import dedent
class ResearchTasks:
def literature_review(self, agent, topic):
return Task(
description=dedent(f"""
Conduct a comprehensive literature review on: {topic}
Your task is to:
1. Search for recent papers and articles (last 2 years)
2. Identify key trends and developments
3. Note important researchers and institutions
4. Find conflicting viewpoints or debates
Present your findings in a structured format.
"""),
agent=agent,
expected_output="A comprehensive literature review report",
)
def trend_analysis(self, agent, topic):
return Task(
description=dedent(f"""
Analyze emerging trends related to: {topic}
Your task is to:
1. Identify 3-5 emerging trends
2. Assess the maturity level of each trend
3. Identify potential applications and use cases
4. Note any challenges or limitations
Provide specific examples for each trend.
"""),
agent=agent,
expected_output="A trend analysis report with examples",
)
tasks/writing_tasks.py:
from crewai import Task
from textwrap import dedent
class WritingTasks:
def create_outline(self, agent, research_findings):
return Task(
description=dedent(f"""
Create a detailed outline for a technical article based on:
Research Findings: {research_findings}
Your task is to:
1. Structure the article logically
2. Define sections and subsections
3. Identify key points for each section
4. Plan examples and illustrations
Output a hierarchical outline with bullet points.
"""),
agent=agent,
expected_output="A detailed article outline",
)
def write_draft(self, agent, outline, research_findings):
return Task(
description=dedent(f"""
Write a complete draft of the article based on:
Outline: {outline}
Research Findings: {research_findings}
Your task is to:
1. Write engaging, clear content for each section
2. Include relevant examples and explanations
3. Maintain consistent tone throughout
4. Add transitions between sections
Target length: 2000-3000 words.
"""),
agent=agent,
expected_output="A complete article draft",
)
Orchestration
main.py:
from crewai import Crew, Process
from agents.researcher import Researcher
from agents.writer import Writer
from agents.reviewer import Reviewer
from agents.manager import Manager
from tasks.research_tasks import ResearchTasks
from tasks.writing_tasks import WritingTasks
from langchain_openai import ChatOpenAI
def create_content_crew(topic: str):
"""Create a crew for content creation."""
llm = ChatOpenAI(model="gpt-4o")
# Create agents
researcher = Researcher(llm=llm).create()
writer = Writer(llm=llm).create()
reviewer = Reviewer(llm=llm).create()
manager = Manager(llm=llm).create()
# Create tasks
research_tasks = ResearchTasks()
writing_tasks = WritingTasks()
lit_review = research_tasks.literature_review(researcher, topic)
trend_analysis = research_tasks.trend_analysis(researcher, topic)
outline = writing_tasks.create_outline(writer, "${researcher.tools_output}")
draft = writing_tasks.write_draft(writer, "${outline.output}", "${researcher.tools_output}")
review = Task(
description=f"""
Review and improve the draft: {draft.output}
Check for:
1. Factual accuracy
2. Clarity and readability
3. Logical flow
4. Grammar and style
Provide specific feedback and suggested improvements.
""",
agent=reviewer,
expected_output="A reviewed draft with improvement suggestions",
)
# Create crew
crew = Crew(
agents=[researcher, writer, reviewer, manager],
tasks=[lit_review, trend_analysis, outline, draft, review],
process=Process.sequential,
verbose=True,
)
return crew
if __name__ == "__main__":
topic = "The Future of AI Agents in Enterprise Software"
crew = create_content_crew(topic)
result = crew.kickoff()
print("\n\n===== FINAL RESULT =====\n")
print(result)
Running the System
python main.py
Advanced: LangGraph Version
For more complex orchestration with state management:
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
research: str
draft: str
reviewed: str
def research_node(state: AgentState):
# Run research
return {"research": research_result}
def write_node(state: AgentState):
# Write draft based on research
return {"draft": draft_result}
def review_node(state: AgentState):
# Review and provide feedback
return {"reviewed": review_result}
# Build graph
workflow = StateGraph(AgentState)
workflow.add_node("research", research_node)
workflow.add_node("write", write_node)
workflow.add_node("review", review_node)
workflow.set_entry_point("research")
workflow.add_edge("research", "write")
workflow.add_edge("write", "review")
workflow.add_edge("review", END)
app = workflow.compile()
result = app.invoke({"messages": []})
Best Practices
- Define clear roles: Each agent should have a distinct, non-overlapping role
- Use delegation carefully: Allow agents to delegate when appropriate
- Structure tasks well: Clear task descriptions lead to better results
- Iterate on agent prompts: Fine-tune backstories and goals
- Monitor the process: Use verbose mode during development