The 6 Swarm Patterns

All implement the Pattern protocol: async execute(agents, task, ctx) → SwarmResult

1. Sequential — ordered pipeline
agent1 → agent2 → agent3 → result
  each receives prior output as input
Key logic: 
result = task
for agent in agents:
    r = await agent.run(result, ctx)
    ctx.add_result(r)
    result = r.content   # chain output→input
return SwarmResult(...)
2. Parallel — concurrent fan-out
       ┌→ agent1 →┐
task ──├→ agent2 →┼── merge → result
       └→ agent3 →┘
Key logic: 
results = await asyncio.gather(
    *[agent.run(task, ctx) for agent in agents]
)
merged = merge_results(results)
return SwarmResult(...)
3. Hierarchical — coordinator + workers
coordinator decomposes task →
  worker1(subtask1), worker2(subtask2)
coordinator synthesizes results
Key logic: 
coordinator, *workers = agents
plan = await coordinator.run(task, ctx)
subtasks = parse_subtasks(plan.content)
results = await asyncio.gather(
    *[w.run(s, ctx) for w,s in zip(workers, subtasks)]
)
final = await coordinator.run(
    synthesize_prompt(results), ctx)
return SwarmResult(...)
4. Decentralized — peer consensus
all agents respond independently →
  vote/consensus mechanism →
  majority or weighted result
Key logic: 
results = await asyncio.gather(
    *[agent.run(task, ctx) for agent in agents]
)
# weighted by confidence score
final = weighted_consensus(results)
return SwarmResult(...)
5. Adaptive — dynamic routing
run agent → check confidence →
  high: done
  low:  route to specialist or retry
Key logic: 
agent_map = {a.name: a for a in agents}
current = agents[0]
for _ in range(max_hops):
    result = await current.run(task, ctx)
    if result.confidence >= threshold:
        break
    next_name = result.next_agent
    if not next_name: break
    current = agent_map[next_name]
return SwarmResult(...)
6. Mesh P2P — direct agent comms
agent1 ←→ agent2 ←→ agent3
  ↕                      ↕
agent4 ←────────────→ agent5
Key logic: 
# agents can message each other via ctx
# rounds of communication until stable
for round in range(max_rounds):
    results = await asyncio.gather(
        *[a.run(task, ctx) for a in agents]
    )
    ctx.state["mesh_round"] = results
    if converged(results): break
return SwarmResult(...)

Shared pattern interface

class SequentialPattern:
    async def execute(self, agents, task, ctx) -> SwarmResult: ...

class ParallelPattern:
    merge_strategy: str = "concatenate"  # or "vote", "synthesize"
    async def execute(self, agents, task, ctx) -> SwarmResult: ...

class AdaptivePattern:
    confidence_threshold: float = 0.8
    max_hops: int = 5
    async def execute(self, agents, task, ctx) -> SwarmResult: ...