1# Multi-Agent Patterns: Technical Reference
2 
3This document provides implementation details for multi-agent architectures across different frameworks.
4 
5## Supervisor Pattern
6 
7### LangGraph Supervisor Implementation
8 
9Implement a supervisor that routes to worker nodes:
10 
11```python
12from typing import TypedDict, Union
13from langgraph.graph import StateGraph, END
14 
15class AgentState(TypedDict):
16    task: str
17    current_agent: str
18    task_output: dict
19    messages: list
20 
21def supervisor_node(state: AgentState) -> AgentState:
22    """
23    Supervisor decides which worker to invoke next.
24    
25    Returns routing decision and updates state.
26    """
27    task = state["task"]
28    messages = state.get("messages", [])
29    
30    # Determine next agent based on task and history
31    if "research" in task.lower():
32        next_agent = "researcher"
33    elif "write" in task.lower() or "create" in task.lower():
34        next_agent = "writer"
35    elif "review" in task.lower() or "analyze" in task.lower():
36        next_agent = "reviewer"
37    else:
38        next_agent = "coordinator"
39    
40    return {
41        "task": task,
42        "current_agent": next_agent,
43        "task_output": {},
44        "messages": messages + [{"supervisor": f"Routing to {next_agent}"}]
45    }
46 
47def researcher_node(state: AgentState) -> AgentState:
48    """Research worker that gathers information."""
49    # Perform research task
50    output = perform_research(state["task"])
51    
52    return {
53        "task": state["task"],
54        "current_agent": "researcher",
55        "task_output": output,
56        "messages": state["messages"] + [{"researcher": "Research complete"}]
57    }
58 
59def writer_node(state: AgentState) -> AgentState:
60    """Writer worker that creates content based on research."""
61    output = create_content(state["task"], state["task_output"])
62    
63    return {
64        "task": state["task"],
65        "current_agent": "writer",
66        "task_output": output,
67        "messages": state["messages"] + [{"writer": "Content created"}]
68    }
69 
70def build_supervisor_graph():
71    """Build the supervisor workflow graph."""
72    workflow = StateGraph(AgentState)
73    
74    # Add nodes
75    workflow.add_node("supervisor", supervisor_node)
76    workflow.add_node("researcher", researcher_node)
77    workflow.add_node("writer", writer_node)
78    
79    # Add edges
80    workflow.add_edge("supervisor", "researcher")
81    workflow.add_edge("researcher", "supervisor")
82    workflow.add_edge("supervisor", "writer")
83    workflow.add_edge("writer", "supervisor")
84    
85    # Set entry point
86    workflow.set_entry_point("supervisor")
87    
88    return workflow.compile()
89```
90 
91### AutoGen Supervisor
92 
93Implement supervisor using GroupChat pattern:
94 
95```python
96from autogen import AssistantAgent, UserProxyAgent, GroupChat
97 
98# Define specialized agents
99researcher = AssistantAgent(
100    name="researcher",
101    system_message="""You are a research specialist.
102    Your goal is to gather accurate, comprehensive information
103    on topics assigned by the supervisor. Always cite sources
104    and note confidence levels.""",
105    llm_config=llm_config
106)
107 
108writer = AssistantAgent(
109    name="writer",
110    system_message="""You are a content creation specialist.
111    Your goal is to create well-structured content based on
112    research provided by the supervisor. Follow style guidelines
113    and ensure factual accuracy.""",
114    llm_config=llm_config
115)
116 
117# Define supervisor
118supervisor = AssistantAgent(
119    name="supervisor",
120    system_message="""You are the project supervisor.
121    Your goal is to coordinate researchers and writers to
122    complete tasks efficiently.
123    
124    Process:
125    1. Break down the task into research and writing phases
126    2. Route to appropriate specialists
127    3. Synthesize results into final output
128    4. Ensure quality before completing""",
129    llm_config=llm_config
130)
131 
132# Configure group chat
133group_chat = GroupChat(
134    agents=[supervisor, researcher, writer],
135    messages=[],
136    max_round=20
137)
138 
139manager = GroupChatManager(
140    groupchat=group_chat,
141    llm_config=llm_config
142)
143```
144 
145## Swarm Pattern Implementation
146 
147### LangGraph Swarms
148 
149Implement peer-to-peer handoffs:
150 
151```python
152def create_agent(name, system_prompt, tools):
153    """Create an agent node for the swarm."""
154    
155    def agent_node(state):
156        # Process current state with agent
157        response = invoke_agent(name, system_prompt, state["input"], tools)
158        
159        # Check for handoff
160        if "handoff" in response:
161            return {"next_agent": response["handoff"], "output": response["output"]}
162        else:
163            return {"next_agent": END, "output": response["output"]}
164    
165    return agent_node
166 
167def build_swarm():
168    """Build a peer-to-peer agent swarm."""
169    workflow = StateGraph(State)
170    
171    # Create agents
172    triage = create_agent("triage", TRIAGE_PROMPT, [search, read])
173    research = create_agent("research", RESEARCH_PROMPT, [search, browse, read])
174    analysis = create_agent("analysis", ANALYSIS_PROMPT, [calculate, compare])
175    writing = create_agent("writing", WRITING_PROMPT, [write, edit])
176    
177    # Add to graph
178    workflow.add_node("triage", triage)
179    workflow.add_node("research", research)
180    workflow.add_node("analysis", analysis)
181    workflow.add_node("writing", writing)
182    
183    # Define handoff edges
184    workflow.add_edge("triage", "research")
185    workflow.add_edge("triage", "analysis")
186    workflow.add_edge("research", "writing")
187    workflow.add_edge("analysis", "writing")
188    
189    workflow.set_entry_point("triage")
190    
191    return workflow.compile()
192```
193 
194## Hierarchical Pattern Implementation
195 
196### CrewAI-Style Hierarchy
197 
198```python
199class ManagerAgent:
200    def __init__(self, name, system_prompt, llm):
201        self.name = name
202        self.system_prompt = system_prompt
203        self.llm = llm
204        self.workers = []
205    
206    def add_worker(self, worker):
207        """Add a worker agent to the team."""
208        self.workers.append(worker)
209    
210    def delegate(self, task):
211        """
212        Analyze task and delegate to appropriate worker.
213        
214        Returns work assignment and expected output format.
215        """
216        # Analyze task requirements
217        requirements = analyze_task_requirements(task)
218        
219        # Select best worker
220        best_worker = select_worker(self.workers, requirements)
221        
222        # Create assignment
223        assignment = {
224            "worker": best_worker.name,
225            "task": task,
226            "context": self.get_relevant_context(task),
227            "output_format": requirements.output_format,
228            "deadline": requirements.deadline
229        }
230        
231        return assignment
232    
233    def review_output(self, worker_output, requirements):
234        """
235        Review worker output against requirements.
236        
237        Returns approval or revision request.
238        """
239        quality_score = assess_quality(worker_output, requirements)
240        
241        if quality_score >= requirements.threshold:
242            return {"status": "approved", "output": worker_output}
243        else:
244            return {
245                "status": "revision_requested",
246                "feedback": generate_feedback(worker_output, requirements),
247                "revise_worker": requirements.revise_worker
248            }
249```
250 
251## Context Isolation Patterns
252 
253### Full Context Delegation
254 
255```python
256def delegate_with_full_context(planner_state, subagent):
257    """
258    Pass entire planner context to subagent.
259    
260    Use for complex tasks requiring complete understanding.
261    """
262    return {
263        "context": planner_state,
264        "subagent": subagent,
265        "isolation_mode": "full"
266    }
267```
268 
269### Instruction Passing
270 
271```python
272def delegate_with_instructions(task_spec, subagent):
273    """
274    Pass only instructions to subagent.
275    
276    Use for simple, well-defined subtasks.
277    """
278    return {
279        "instructions": {
280            "objective": task_spec.objective,
281            "constraints": task_spec.constraints,
282            "inputs": task_spec.inputs,
283            "outputs": task_spec.output_schema
284        },
285        "subagent": subagent,
286        "isolation_mode": "minimal"
287    }
288```
289 
290### File System Coordination
291 
292```python
293class FileSystemCoordination:
294    def __init__(self, workspace_path):
295        self.workspace = workspace_path
296    
297    def write_shared_state(self, key, value):
298        """Write state accessible to all agents."""
299        path = f"{self.workspace}/{key}.json"
300        with open(path, 'w') as f:
301            json.dump(value, f)
302        return path
303    
304    def read_shared_state(self, key):
305        """Read state written by any agent."""
306        path = f"{self.workspace}/{key}.json"
307        with open(path, 'r') as f:
308            return json.load(f)
309    
310    def acquire_lock(self, resource, agent_id):
311        """Prevent concurrent access to shared resources."""
312        lock_path = f"{self.workspace}/locks/{resource}.lock"
313        if os.path.exists(lock_path):
314            return False
315        with open(lock_path, 'w') as f:
316            f.write(agent_id)
317        return True
318```
319 
320## Consensus Mechanisms
321 
322### Weighted Voting
323 
324```python
325def weighted_consensus(agent_outputs, weights):
326    """
327    Calculate weighted consensus from agent outputs.
328    
329    Weight = verbalized_confidence * domain_expertise
330    """
331    weighted_sum = sum(
332        output.vote * weights[output.agent_id]
333        for output in agent_outputs
334    )
335    total_weight = sum(weights[output.agent_id] for output in agent_outputs)
336    
337    return weighted_sum / total_weight
338```
339 
340### Debate Protocol
341 
342```python
343class DebateProtocol:
344    def __init__(self, agents, max_rounds=5):
345        self.agents = agents
346        self.max_rounds = max_rounds
347        self.history = []
348    
349    def run_debate(self, topic):
350        """Execute structured debate on topic."""
351        # Initial statements
352        statements = {agent.name: agent.initial_statement(topic) 
353                      for agent in self.agents}
354        
355        for round_num in range(self.max_rounds):
356            # Generate critiques
357            critiques = {}
358            for agent in self.agents:
359                critiques[agent.name] = agent.critique(
360                    topic, 
361                    statements,
362                    exclude=[agent.name]
363                )
364            
365            # Update statements with critique integration
366            for agent in self.agents:
367                statements[agent.name] = agent.integrate_critique(
368                    statements[agent.name],
369                    critiques
370                )
371            
372            # Check for convergence
373            if self.check_convergence(statements):
374                break
375        
376        # Final evaluation
377        return self.evaluate_final(statements)
378```
379 
380## Failure Recovery
381 
382### Circuit Breaker
383 
384```python
385class AgentCircuitBreaker:
386    def __init__(self, failure_threshold=3, timeout_seconds=60):
387        self.failure_count = {}
388        self.failure_threshold = failure_threshold
389        self.timeout_seconds = timeout_seconds
390    
391    def call(self, agent, task):
392        """Execute agent task with circuit breaker protection."""
393        if self.is_open(agent.name):
394            raise CircuitBreakerOpen(f"Agent {agent.name} temporarily unavailable")
395        
396        try:
397            result = agent.execute(task)
398            self.record_success(agent.name)
399            return result
400        except Exception as e:
401            self.record_failure(agent.name)
402            if self.failure_count[agent.name] >= self.failure_threshold:
403                self.open_circuit(agent.name)
404            raise
405```
406 
407### Checkpoint and Resume
408 
409```python
410class CheckpointManager:
411    def __init__(self, checkpoint_dir):
412        self.checkpoint_dir = checkpoint_dir
413        os.makedirs(checkpoint_dir, exist_ok=True)
414    
415    def save_checkpoint(self, workflow_id, step, state):
416        """Save workflow state for potential resume."""
417        checkpoint = {
418            "workflow_id": workflow_id,
419            "step": step,
420            "state": state,
421            "timestamp": time.time()
422        }
423        path = f"{self.checkpoint_dir}/{workflow_id}.json"
424        with open(path, 'w') as f:
425            json.dump(checkpoint, f)
426    
427    def load_checkpoint(self, workflow_id):
428        """Load last saved checkpoint for workflow."""
429        path = f"{self.checkpoint_dir}/{workflow_id}.json"
430        with open(path, 'r') as f:
431            return json.load(f)
432```
433 
434