Multi-Agent Systems

Multi-agent systems use multiple LLM-based agents working together to solve complex tasks. Each agent has a specialized role, specific tools, and a focused context, enabling the system to handle tasks that require diverse expertise.

Agent Architectures

Hierarchical Architecture

class HierarchicalMultiAgent:
    def __init__(self, supervisor, workers):
        self.supervisor = supervisor
        self.workers = workers  # name -> agent mapping
    
    def solve(self, task):
        # Supervisor plans the work
        plan = self.supervisor.plan(task)
        
        # Execute each subtask
        results = {}
        for subtask in plan["subtasks"]:
            worker_name = subtask["assigned_to"]
            worker = self.workers[worker_name]
            result = worker.execute(subtask["description"], context=results)
            results[subtask["id"]] = result
        
        # Supervisor synthesizes results
        final_answer = self.supervisor.synthesize(task, results)
        return final_answer

           [Supervisor Agent]
          /        |        \
         v         v         v
    [Research] [Analysis] [Writing]
     Agent      Agent      Agent

Peer-to-Peer Architecture

class PeerToPeerMultiAgent:
    def __init__(self, agents):
        self.agents = agents
    
    def solve(self, task, max_rounds=5):
        # Initialize with task
        context = {"task": task, "messages": []}
        
        for round_num in range(max_rounds):
            # Each agent contributes
            for agent in self.agents:
                contribution = agent.contribute(context)
                context["messages"].append({
                    "agent": agent.name,
                    "content": contribution
                })
            
            # Check for consensus
            if self.check_consensus(context):
                break
        
        return self.final_answer(context)

Debate-Based Architecture

def multi_agent_debate(question, agents, rounds=3):
    """Conduct a structured debate among agents."""
    positions = {}
    
    # Initial positions
    for agent in agents:
        positions[agent.name] = agent.initial_position(question)
    
    # Debate rounds
    for round_num in range(rounds):
        new_positions = {}
        for agent in agents:
            # Agent sees other positions and argues
            other_positions = {k: v for k, v in positions.items() if k != agent.name}
            response = agent.debates(question, other_positions)
            new_positions[agent.name] = response
        positions = new_positions
    
    # Synthesize final answer
    final_answer = synthesize_debate(question, positions)
    return final_answer

Coordination Patterns

Message Passing

class AgentMessage:
    def __init__(self, sender, receiver, content, msg_type="info"):
        self.sender = sender
        self.receiver = receiver
        self.content = content
        self.msg_type = msg_type  # info, request, response, feedback

class MessageBus:
    def __init__(self):
        self.messages = []
        self.handlers = {}
    
    def register(self, agent_name, handler):
        self.handlers[agent_name] = handler
    
    def send(self, message):
        self.messages.append(message)
        if message.receiver in self.handlers:
            self.handlers[message.receiver](message)

Shared Memory

class SharedMemory:
    def __init__(self):
        self.store = {}
        self.lock = threading.Lock()
    
    def read(self, key):
        with self.lock:
            return self.store.get(key)
    
    def write(self, key, value):
        with self.lock:
            self.store[key] = value
    
    def append(self, key, value):
        with self.lock:
            if key not in self.store:
                self.store[key] = []
            self.store[key].append(value)

Specialized Agent Roles

Common Agent Types

Example: Code Review System

code_review_system = {
    "planner": PlannerAgent(
        role="Break down code review into manageable tasks"
    ),
    "security_reviewer": CodeReviewerAgent(
        role="Review code for security vulnerabilities",
        tools=["static_analysis", "vulnerability_scanner"]
    ),
    "performance_reviewer": CodeReviewerAgent(
        role="Review code for performance issues",
        tools=["profiler", "complexity_analyzer"]
    ),
    "style_reviewer": CodeReviewerAgent(
        role="Review code for style and best practices",
        tools=["linter", "style_checker"]
    ),
    "synthesizer": SynthesizerAgent(
        role="Combine all review feedback into a coherent report"
    )
}

Scalability Considerations

Practice Exercises

1. Hierarchical System: Build a hierarchical multi-agent system with a supervisor and 3 specialized workers. Test on a task requiring research, analysis, and writing.

2. Debate System: Implement a 3-agent debate system where agents argue different positions. Compare the debate outcome to individual agent responses.

3. Coordination Patterns: Compare message passing vs shared memory for coordinating 5 agents. Which is more efficient for different task types?

4. Scalability Test: Scale your multi-agent system from 2 to 10 agents. At what point does communication overhead become a bottleneck?

Key Takeaways

What to Learn Next