Multi-agent systems

In practice, when tasks cross more than a couple of domains, a single agent often degrades quickly. A multi-agent setup can hold quality steady because each specialist stays in a narrower problem space.

A central supervisor routes requests to specialised sub-agents.

"""
Multi-Agent Supervisor Pattern
==============================
A supervisor agent coordinates specialised sub-agents.
"""

from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from enum import Enum


class AgentRole(Enum):
    """Roles for specialised agents."""
    RESEARCHER = "researcher"
    WRITER = "writer"
    CODER = "coder"
    ANALYST = "analyst"


@dataclass
class AgentMessage:
    """Message passed between agents."""
    sender: str
    recipient: str
    content: str
    message_type: str  # "task", "result", "query", "status"
    metadata: Dict[str, Any] = None


class SpecialisedAgent:
    """A specialised agent with a specific focus area."""
    
    def __init__(
        self,
        name: str,
        role: AgentRole,
        system_prompt: str,
        tools: List = None
    ):
        self.name = name
        self.role = role
        self.system_prompt = system_prompt
        self.tools = tools or []
    
    def process(self, message: AgentMessage) -> AgentMessage:
        """Process an incoming message and return a response."""
        prompt = f"""You are {self.name}, a specialist {self.role.value}.

{self.system_prompt}

Task from supervisor:
{message.content}

Provide your expert response:
"""
        
        response = self._call_llm(prompt)
        
        return AgentMessage(
            sender=self.name,
            recipient=message.sender,
            content=response,
            message_type="result"
        )
    
    def _call_llm(self, prompt: str) -> str:
        """Call the underlying LLM."""
        import requests
        response = requests.post(
            "http://localhost:11434/api/generate",
            json={"model": "llama3.2:3b", "prompt": prompt, "stream": False}
        )
        return response.json().get("response", "")


class SupervisorAgent:
    """Supervisor that coordinates multiple specialised agents."""
    
    ROUTING_PROMPT = """You are a supervisor coordinating a team of specialists.

Available specialists:
{agents}

User request:
{query}

Decide which specialist should handle this. Respond with JSON:
{{"agent": "agent_name", "task": "specific task for them"}}
"""
    
    def __init__(self, agents: List[SpecialisedAgent]):
        self.agents = {agent.name: agent for agent in agents}
    
    def run(self, query: str) -> str:
        """Process a user query through the multi-agent system."""
        # Route to appropriate agent
        routing = self._route_request(query)
        
        agent_name = routing.get("agent")
        task = routing.get("task", query)
        
        if agent_name in self.agents:
            agent = self.agents[agent_name]
            message = AgentMessage(
                sender="supervisor",
                recipient=agent_name,
                content=task,
                message_type="task"
            )
            response = agent.process(message)
            return response.content
        
        return "No suitable agent found for this request."
    
    def _route_request(self, query: str) -> Dict:
        """Determine which agent should handle the request."""
        agents_desc = "\n".join(
            f"- {name}: {agent.role.value}"
            for name, agent in self.agents.items()
        )
        
        prompt = self.ROUTING_PROMPT.format(agents=agents_desc, query=query)
        
        import requests
        response = requests.post(
            "http://localhost:11434/api/generate",
            json={"model": "llama3.2:3b", "prompt": prompt, "stream": False}
        )
        
        import json
        try:
            return json.loads(response.json().get("response", "{}"))
        except:
            return {"agent": list(self.agents.keys())[0], "task": query}

In a swarm, agents hand off directly to each other without going through a supervisor.

Choosing a framework matters, but rankings age badly. Compare frameworks against the operating model you need rather than a momentary popularity chart.

Before diving into visual tools like n8n, use this interactive designer to understand how agent workflows are structured. Explore template workflows or design your own multi-step agent processes.