multi-agent-systems

name	multi-agent-systems
description	Design and orchestrate multi-agent AI systems with specialist agents, coordination patterns, and shared memory. Outputs agent topology, communication protocol, error handling, and evaluation framework.
argument-hint	["task complexity","required specialisations","parallelism opportunities","safety requirements"]
allowed-tools	Read, Write

Multi-Agent Systems

Multi-agent systems decompose complex tasks across specialist agents that collaborate — one researches, one writes, one reviews. The design challenges are coordination (how agents communicate), memory (what state they share), and reliability (handling agent failures without failing the whole task).

When to Use Multi-Agent

SINGLE AGENT FIRST — add agents only when:
  ✓ Task is too long for one context window
  ✓ Task has genuinely parallel subtasks
  ✓ Different subtasks need different specialisations
  ✓ Quality benefits from review/critique loop

MULTI-AGENT PATTERNS:
  Orchestrator → Workers (most common)
  Pipeline (Agent A → Agent B → Agent C)
  Parallel fan-out with reducer
  Debate (Agent A vs Agent B → Judge)
  Reflection (Writer → Critic → Writer)

Orchestrator Pattern

import anthropic
import asyncio
from dataclasses import dataclass
from typing import 

client = anthropic.Anthropic()


 :
    agent_name: 
    output: 
    success: 
    error:  |  = 

 :
    
    
     ():
        .model = model
    
      () -> :
        
        plan =  ._plan(task)
        
        
        results =  ._execute(plan, task)
        
        
        synthesis =  ._synthesise(task, results)
        
         synthesis
    
      () -> :
        response = client.messages.create(
            model=.model,
            max_tokens=,
            messages=[{: , : }]
        )
         json
         json.loads(response.content[].text)
    
      () -> [AgentResult]:
        completed: [, AgentResult] = {}
        
         subtask  plan[]:
            
            deps_output = {dep_id: completed[dep_id].output 
                           dep_id  subtask.get(, [])
                           dep_id  completed}
            
            result =  ._run_agent(
                agent_type=subtask[],
                description=subtask[],
                context=deps_output,
                original_task=original_task,
            )
            completed[subtask[]] = result
        
         (completed.values())
    
      () -> AgentResult:
        agent_prompts = {
            : ,
            : ,
            : ,
            : ,
            : ,
        }
        
        system = agent_prompts.get(agent_type, )
        
        context_text = .join(
             
             dep_id, output  context.items()
        )  context  
        
        :
            response = client.messages.create(
                model=  agent_type !=   ,
                max_tokens=,
                system=system,
                messages=[{: , : }]
            )
             AgentResult(agent_name=agent_type, output=response.content[].text, success=)
         Exception  e:
             AgentResult(agent_name=agent_type, output=, success=, error=(e))
    
      () -> :
        successful = [r  r  results  r.success]
        work_summary = .join(  r  successful)
        
        response = client.messages.create(
            model=.model,
            max_tokens=,
            messages=[{: , : }]
        )
         response.content[].text