Multi-Agent Systems¶
Orchestrating multiple specialized agents: supervisor patterns, parallel execution, and agent communication.
# Install dependencies
# !pip install autogen crewai langchain
Simple Multi-Agent System¶
from typing import List, Dict, Any
from dataclasses import dataclass
@dataclass
class Message:
"""Message between agents"""
sender: str
receiver: str
content: str
message_type: str = "task" # task, result, feedback, question
class Agent:
"""Base agent class"""
def __init__(self, name: str, role: str, expertise: List[str]):
self.name = name
self.role = role
self.expertise = expertise
self.inbox: List[Message] = []
self.outbox: List[Message] = []
def receive(self, message: Message):
"""Receive a message"""
self.inbox.append(message)
print(f"📨 {self.name} received: {message.content[:50]}...")
def send(self, receiver: str, content: str, message_type: str = "task"):
"""Send a message"""
message = Message(self.name, receiver, content, message_type)
self.outbox.append(message)
print(f"📤 {self.name} sent to {receiver}: {content[:50]}...")
return message
def process(self, message: Message) -> str:
"""Process a message (override in subclasses)"""
return f"{self.name} processed: {message.content}"
# Create specialized agents
class ResearcherAgent(Agent):
"""Agent specialized in research"""
def process(self, message: Message) -> str:
query = message.content
# Mock research
result = f"Research findings for '{query}':\n"
result += "- Finding 1: Important insight\n"
result += "- Finding 2: Key data point\n"
result += "- Finding 3: Relevant study"
return result
class CoderAgent(Agent):
"""Agent specialized in coding"""
def process(self, message: Message) -> str:
task = message.content
# Mock code generation
code = f"def solution():\n # Solution for: {task}\n pass\n return result"
return f"Generated code:\n```python\n{code}\n```"
class ReviewerAgent(Agent):
"""Agent specialized in reviewing"""
def process(self, message: Message) -> str:
content = message.content
# Mock review
import random
quality = random.choice(["excellent", "good", "needs improvement"])
feedback = f"Review: {quality}\n"
if quality == "needs improvement":
feedback += "Suggestions:\n- Add error handling\n- Improve documentation"
else:
feedback += "Looks good! No major issues."
return feedback
# Create agents
researcher = ResearcherAgent("Alice", "researcher", ["web search", "data analysis"])
coder = CoderAgent("Bob", "developer", ["python", "javascript"])
reviewer = ReviewerAgent("Charlie", "reviewer", ["code review", "quality assurance"])
print("Agents created:")
for agent in [researcher, coder, reviewer]:
print(f" {agent.name} ({agent.role}): {', '.join(agent.expertise)}")
Supervisor Pattern¶
class SupervisorAgent(Agent):
"""Coordinates other agents"""
def __init__(self, name: str, workers: List[Agent]):
super().__init__(name, "supervisor", ["coordination", "planning"])
self.workers = {agent.name: agent for agent in workers}
def delegate(self, task: str):
"""Break down task and delegate"""
print(f"\n{'='*70}")
print(f"🎯 {self.name} delegating task: {task}")
print(f"{'='*70}\n")
# Step 1: Research
print("Step 1: Research Phase")
research_msg = self.send("Alice", f"Research: {task}")
self.workers["Alice"].receive(research_msg)
research_result = self.workers["Alice"].process(research_msg)
print(f"✅ Research completed\n")
# Step 2: Development
print("Step 2: Development Phase")
code_msg = self.send("Bob", f"Implement: {task}\nBased on: {research_result}")
self.workers["Bob"].receive(code_msg)
code_result = self.workers["Bob"].process(code_msg)
print(f"✅ Code generated\n")
# Step 3: Review
print("Step 3: Review Phase")
review_msg = self.send("Charlie", f"Review: {code_result}")
self.workers["Charlie"].receive(review_msg)
review_result = self.workers["Charlie"].process(review_msg)
print(f"✅ Review completed\n")
# Aggregate results
final_result = {
"research": research_result,
"code": code_result,
"review": review_result
}
print(f"{'='*70}")
print("✨ Task completed!")
print(f"{'='*70}\n")
return final_result
# Create supervisor
supervisor = SupervisorAgent("Manager", [researcher, coder, reviewer])
# Delegate a task
result = supervisor.delegate("Build a web scraper for news articles")
print("\nFinal Results:")
print("\n📊 Research:")
print(result['research'])
print("\n💻 Code:")
print(result['code'])
print("\n✓ Review:")
print(result['review'])
Collaborative Pattern (Debate)¶
class DebateAgent(Agent):
"""Agent that can debate and refine ideas"""
def __init__(self, name: str, position: str):
super().__init__(name, "debater", ["argumentation", "analysis"])
self.position = position
self.arguments = []
def argue(self, topic: str, counter_arg: str = None) -> str:
"""Make an argument"""
if counter_arg:
argument = f"{self.name} ({self.position}): While you raise valid points, consider that..."
else:
argument = f"{self.name} ({self.position}): I believe {topic} because..."
self.arguments.append(argument)
return argument
class Moderator:
"""Moderates debate between agents"""
def __init__(self, agents: List[DebateAgent]):
self.agents = agents
def run_debate(self, topic: str, rounds: int = 3) -> str:
"""Run a debate"""
print(f"\n{'='*70}")
print(f"🎙️ DEBATE: {topic}")
print(f"{'='*70}\n")
last_argument = None
for round_num in range(rounds):
print(f"--- Round {round_num + 1} ---\n")
for agent in self.agents:
argument = agent.argue(topic, last_argument)
print(f"💬 {argument}\n")
last_argument = argument
# Synthesize consensus
consensus = f"After {rounds} rounds of debate, both sides agree that...\n"
consensus += "- Multiple perspectives are valuable\n"
consensus += "- The truth likely lies somewhere in between"
print(f"\n{'='*70}")
print("🤝 CONSENSUS:")
print(consensus)
print(f"{'='*70}\n")
return consensus
# Create debate agents
optimist = DebateAgent("OptimistBot", "PRO")
skeptic = DebateAgent("SkepticBot", "CON")
# Run debate
moderator = Moderator([optimist, skeptic])
consensus = moderator.run_debate("AI will solve climate change", rounds=2)
AutoGen Pattern¶
Microsoft’s AutoGen framework for multi-agent conversations:
# Example with AutoGen (requires installation and API key)
'''
import autogen
config_list = [{
"model": "gpt-4",
"api_key": "your-api-key"
}]
# Create assistant agent
assistant = autogen.AssistantAgent(
name="assistant",
llm_config={"config_list": config_list}
)
# Create user proxy (executes code)
user_proxy = autogen.UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER",
max_consecutive_auto_reply=10,
code_execution_config={"work_dir": "coding"},
)
# Start conversation
user_proxy.initiate_chat(
assistant,
message="Plot a chart of NVIDIA and TESLA stock price changes YTD."
)
'''
print("AutoGen example (commented - requires API key)")
print("AutoGen enables:")
print(" - Automated conversations between agents")
print(" - Code execution and validation")
print(" - Human-in-the-loop when needed")
print(" - Multi-agent collaboration")
CrewAI Pattern¶
Role-based agent orchestration:
# Example with CrewAI (requires installation and API key)
'''
from crewai import Agent, Task, Crew
# Define agents with roles
researcher = Agent(
role="Senior Research Analyst",
goal="Uncover cutting-edge developments in AI",
backstory="Expert in AI with 10 years of experience",
verbose=True
)
writer = Agent(
role="Tech Content Writer",
goal="Write engaging articles about AI",
backstory="Skilled writer with tech background",
verbose=True
)
# Define tasks
research_task = Task(
description="Research latest AI trends in 2024",
agent=researcher
)
write_task = Task(
description="Write a blog post about AI trends",
agent=writer
)
# Create crew
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, write_task],
verbose=2
)
# Execute
result = crew.kickoff()
'''
print("CrewAI example (commented - requires API key)")
print("CrewAI enables:")
print(" - Role-based agent design")
print(" - Sequential task execution")
print(" - Agent delegation")
print(" - Process automation")
Best Practices¶
1. Agent Design¶
Single Responsibility: Each agent has one clear role
Clear Expertise: Define what each agent is good at
Communication Protocol: Standardize message formats
Error Handling: Agents should handle failures gracefully
2. Coordination¶
Supervisor Pattern: For hierarchical control
Sequential Pattern: For pipeline-style workflows
Collaborative Pattern: For peer review and debate
Hybrid Patterns: Combine as needed
3. Communication¶
Use structured messages (sender, receiver, type, content)
Implement message queues for async communication
Log all interactions for debugging
Set timeouts for responses
4. Quality Control¶
Add reviewer agents for validation
Implement feedback loops
Track agent performance
Allow human oversight
When to Use Multi-Agent Systems¶
✅ Good Use Cases:
Complex tasks requiring specialized skills
Workflows with distinct stages (research → code → review)
Tasks benefiting from multiple perspectives
Parallel task execution
❌ Avoid When:
Simple, single-step tasks
Low-latency requirements
Limited API budget
Highly sequential dependencies
Key Takeaways¶
✅ Multi-agent systems enable specialized, collaborative AI
✅ Common patterns: Supervisor, Sequential, Collaborative
✅ AutoGen and CrewAI provide production-ready frameworks
✅ Clear roles and communication protocols are essential
✅ Start simple, add complexity as needed