Multi-Agent Orchestration Patterns: Supervisor, Pipeline, and Swarm Architectures Compared

from hermes_agent import Agent as HermesAgent
from openclaw import Agent as OpenClawAgent

# Specialists — each owns a narrow capability
browser      = OpenClawAgent(name="browser",    tools=["goto", "click", "screenshot", "read_dom"])
file_worker  = OpenClawAgent(name="file_worker", tools=["read_file", "write_file", "list_dir"])
analyst      = HermesAgent(name="analyst",     system="You analyze structured data and return JSON findings.")
writer       = HermesAgent(name="writer",      system="You produce polished written output from findings.")

SPECIALISTS = {a.name: a for a in [browser, file_worker, analyst, writer]}

# Supervisor — reasoning model with ONE tool: dispatch
supervisor = HermesAgent(
    name="supervisor",
    system=(
        "You are the supervisor of a multi-agent team. Decompose the user's task, "
        "then call dispatch(specialist, instruction) one step at a time. "
        f"Available specialists: {list(SPECIALISTS)}. "
        "Return a final answer only when all sub-steps are complete."
    ),
)

@supervisor.tool
def dispatch(specialist: str, instruction: str) -> dict:
    if specialist not in SPECIALISTS:
        return {"error": f"unknown specialist: {specialist}"}
    result = SPECIALISTS[specialist].run(instruction)
    return {"specialist": specialist, "result": result}

# Usage
final = supervisor.run(
    "Pull the top 5 HN posts, summarize each, and write a 200-word briefing."
)

from dataclasses import dataclass
from typing import Callable, Any
from hermes_agent import Agent as HermesAgent
from openclaw import Agent as OpenClawAgent

@dataclass
class Stage:
    name: str
    run: Callable[[Any], Any]

# Stage 1 — scrape (OpenClaw browser)
scraper = OpenClawAgent(name="scraper", tools=["goto", "read_dom"])
def scrape(sources: list[str]) -> list[dict]:
    return [scraper.run(f"Go to {url} and return the article text as JSON.") for url in sources]

# Stage 2 — extract facts (Hermes)
extractor = HermesAgent(name="extractor", system="Extract factual claims from articles as structured JSON.")
def extract(articles: list[dict]) -> list[dict]:
    return extractor.run({"articles": articles, "instruction": "Return facts[]"})

# Stage 3 — draft (Hermes)
drafter = HermesAgent(name="drafter", system="Write a 400-word briefing from extracted facts.")
def draft(facts: list[dict]) -> str:
    return drafter.run({"facts": facts})

# Stage 4 — publish (OpenClaw CMS)
publisher = OpenClawAgent(name="publisher", tools=["goto", "click", "type"])
def publish(briefing: str) -> dict:
    return publisher.run(f"Open the CMS, create a new post, paste: {briefing}, save as draft.")

PIPELINE = [
    Stage("scrape",  scrape),
    Stage("extract", extract),
    Stage("draft",   draft),
    Stage("publish", publish),
]

def run_pipeline(seed: Any, stages: list[Stage] = PIPELINE) -> Any:
    state = seed
    for stage in stages:
        print(f"[pipeline] {stage.name}")
        state = stage.run(state)
    return state

# Usage
result = run_pipeline(["https://hn.example.com", "https://news.example.com"])

import asyncio
from collections import deque
from hermes_agent import Agent as HermesAgent
from openclaw import Agent as OpenClawAgent

class Blackboard:
    """Shared message bus — every agent reads and writes here."""
    def __init__(self):
        self.messages: deque[dict] = deque()
        self.subscribers: list[asyncio.Queue] = []

    def post(self, sender: str, kind: str, content):
        msg = {"from": sender, "kind": kind, "content": content}
        self.messages.append(msg)
        for q in self.subscribers:
            q.put_nowait(msg)

    def subscribe(self) -> asyncio.Queue:
        q = asyncio.Queue()
        self.subscribers.append(q)
        return q

bb = Blackboard()

researcher = HermesAgent(name="researcher", system="Propose claims about the topic with sources.")
critic     = HermesAgent(name="critic",     system="Find flaws or contradictions in proposed claims.")
verifier   = OpenClawAgent(name="verifier", tools=["goto", "read_dom"])

async def run_agent(agent, acts_on: list[str], emits: str):
    q = bb.subscribe()
    while True:
        msg = await q.get()
        if msg["from"] == agent.name or msg["kind"] not in acts_on:
            continue
        if msg["kind"] == "halt":
            return
        output = await asyncio.to_thread(agent.run, msg["content"])
        bb.post(agent.name, emits, output)

async def terminator(max_rounds: int = 12):
    # Simple halt: stop after N messages
    while len(bb.messages) < max_rounds:
        await asyncio.sleep(0.5)
    bb.post("system", "halt", None)

async def main(topic: str):
    bb.post("user", "claim_request", topic)
    await asyncio.gather(
        run_agent(researcher, ["claim_request", "critique"], "claim"),
        run_agent(critic,     ["claim"],                     "critique"),
        run_agent(verifier,   ["claim"],                     "verification"),
        terminator(max_rounds=12),
    )
    return list(bb.messages)

asyncio.run(main("Are serverless GPUs economical for 24/7 agent workloads?"))

1. Termination
   - Supervisor: max_steps + explicit "final_answer" tool
   - Pipeline:   stage count is finite by definition — easy
   - Swarm:      REQUIRES an explicit halt condition (quorum, max_msgs, leader vote)

2. Error handling
   - Supervisor: one retry policy in the coordinator
   - Pipeline:   per-stage retries + dead-letter queue for poison inputs
   - Swarm:      per-agent retries; watch for cascading failures across peers

3. Cost controls
   - Supervisor: rate-limit the dispatch tool itself
   - Pipeline:   token budget per stage, hard cap per run
   - Swarm:      token budget per agent AND per blackboard round

4. Observability
   - Supervisor: trace tree rooted at the supervisor — cleanest
   - Pipeline:   one span per stage — linear, easy to read
   - Swarm:      structured blackboard log + per-agent spans; hardest to follow

5. State persistence
   - Supervisor: persist the supervisor's conversation
   - Pipeline:   persist after each stage (this is the whole point)
   - Swarm:      persist the blackboard; replaying requires event sourcing