Agent Swarm/Network
Overview
The agent swarm pattern enables decentralized, peer-to-peer agent collaboration without a central supervisor. Agents communicate directly, share context, and dynamically decide which peer should work next. This creates flexible, adaptive systems where complex behaviors emerge from simple local interactions.
Architecture
When to Use
Use agent swarms when:
- Decentralized decision-making is preferred: No single point of control
- Tasks benefit from adaptive routing: Agents choose best peers dynamically
- Rich context sharing is needed: All agents see all work
- Emergent behavior is desired: Complex patterns from simple rules
- Network topology matters: Define specific agent connections
- Flexible collaboration: Agents adapt based on peer outputs
Key Components
1. State Schema
The swarm state differs from supervisor state in key ways:
from typing import Annotated, Any
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
import operator
class SwarmState(TypedDict):
messages: Annotated[list, add_messages] # Conversation history
task: str # Overall task
agents_state: dict[str, dict[str, Any]] # Per-agent state
shared_context: Annotated[list[dict], operator.add] # Accumulated outputs
current_agent: str # Current agent (set by peer)
iteration: int # Current iteration
max_iterations: int # Safety limit
final_result: str # Aggregated resultKey differences from supervisor pattern:
agents_state: Per-agent state tracking (not centralized)shared_context: All agents see all outputs (full transparency)current_agent: Set by previous agent, not by supervisor
2. Agent Configuration
Define agents with their connections (network topology):
from pydantic import BaseModel, Field
class SwarmAgent(BaseModel):
name: str = Field(description="Unique name for the agent")
system_prompt: str = Field(description="System prompt defining role")
connections: list[str] = Field(
default_factory=list,
description="List of agent names this agent can hand off to"
)
tools: list | None = Field(default=None, description="Optional tools")Example swarm configuration:
agents = [
SwarmAgent(
name="researcher",
system_prompt="Research and gather information",
connections=["analyst", "fact_checker"], # Can route to either
),
SwarmAgent(
name="analyst",
system_prompt="Analyze findings",
connections=["writer"], # Routes to writer
),
SwarmAgent(
name="fact_checker",
system_prompt="Verify claims",
connections=["analyst", "writer"], # Can route to either
),
SwarmAgent(
name="writer",
system_prompt="Write final report",
connections=[], # Terminal node
),
]3. Swarm Routing Function
Unlike supervisor pattern, routing simply reads the peer's decision:
def route_swarm(state: SwarmState) -> str:
"""Route based on previous agent's decision."""
if state["iteration"] >= state["max_iterations"]:
return "aggregate"
if state["current_agent"] == "DONE" or not state["current_agent"]:
return "aggregate"
return state["current_agent"] # Route to peer's choiceKey difference: No central decision-maker! Each agent sets current_agent for the next peer.
4. Graph Construction
from langgraph.graph import StateGraph, START, END
from langgraph_ollama_local.patterns.swarm import create_swarm_graph
graph = create_swarm_graph(llm, agents, entry_agent="researcher")Usage
Basic Usage
from langgraph_ollama_local import LocalAgentConfig
from langgraph_ollama_local.patterns import (
SwarmAgent,
create_swarm_graph,
run_swarm_task,
)
config = LocalAgentConfig()
llm = config.create_chat_client()
# Define swarm
agents = [
SwarmAgent(
name="researcher",
system_prompt="Research and gather information",
connections=["analyst"],
),
SwarmAgent(
name="analyst",
system_prompt="Analyze findings and draw insights",
connections=["writer"],
),
SwarmAgent(
name="writer",
system_prompt="Write comprehensive report",
connections=[], # Terminal
),
]
# Create and run
graph = create_swarm_graph(llm, agents)
result = run_swarm_task(
graph,
"Research the benefits of microservices architecture",
max_iterations=8
)
print(result["final_result"])With Tools
from langchain_core.tools import tool
@tool
def search_web(query: str) -> str:
"""Search the web."""
# Implementation
return "Search results..."
agents = [
SwarmAgent(
name="researcher",
system_prompt="Research using web search",
connections=["analyst"],
tools=[search_web], # Provide tools
),
# ... other agents
]
graph = create_swarm_graph(llm, agents)Custom Entry Agent
# Start with a specific agent
graph = create_swarm_graph(
llm,
agents,
entry_agent="fact_checker" # Start here instead of first agent
)Network Topologies
Fully Connected
Every agent can communicate with every other agent:
agents = [
SwarmAgent(name="a", system_prompt="...", connections=["b", "c", "d"]),
SwarmAgent(name="b", system_prompt="...", connections=["a", "c", "d"]),
SwarmAgent(name="c", system_prompt="...", connections=["a", "b", "d"]),
SwarmAgent(name="d", system_prompt="...", connections=["a", "b", "c"]),
]Use when: Maximum flexibility needed
Linear Pipeline
Agents form a sequential pipeline:
agents = [
SwarmAgent(name="a", system_prompt="...", connections=["b"]),
SwarmAgent(name="b", system_prompt="...", connections=["c"]),
SwarmAgent(name="c", system_prompt="...", connections=["d"]),
SwarmAgent(name="d", system_prompt="...", connections=[]),
]Use when: Clear sequential workflow
Star Network
One central agent connects to all others:
agents = [
SwarmAgent(name="hub", system_prompt="...", connections=["a", "b", "c"]),
SwarmAgent(name="a", system_prompt="...", connections=["hub"]),
SwarmAgent(name="b", system_prompt="...", connections=["hub"]),
SwarmAgent(name="c", system_prompt="...", connections=["hub"]),
]Use when: Central coordination with peer specialization
Custom Topology
Define specific communication patterns:
agents = [
SwarmAgent(name="researcher", connections=["analyst", "fact_checker"]),
SwarmAgent(name="analyst", connections=["writer"]),
SwarmAgent(name="fact_checker", connections=["analyst", "writer"]),
SwarmAgent(name="writer", connections=[]),
]Best Practices
- Design clear connections: Define agent connections that match your task structure
- Set reasonable max_iterations: Swarms can iterate more than supervisors (8-12 typical)
- Share context wisely: Use
share_context=Falsefor internal work - Validate topology: Ensure all connections reference valid agents
- Use terminal nodes: Have at least one agent with no connections
- Monitor iterations: Track which agents contribute and how often
- Test routing logic: Verify agents make sensible routing decisions
Comparison: Swarm vs Supervisor
| Aspect | Supervisor Pattern | Swarm Pattern |
|---|---|---|
| Control | Centralized (supervisor decides) | Decentralized (peers decide) |
| Routing | Supervisor routes all agents | Each agent routes to peers |
| Context | Supervisor sees all, agents see subset | All agents see all context |
| Topology | Star (hub-and-spoke) | Any (define connections) |
| Iteration | Supervisor controls flow | Emerges from peer decisions |
| Scalability | Supervisor is bottleneck | Distributed decision-making |
| Predictability | More predictable | Less predictable (emergent) |
| Debugging | Easier (central control) | Harder (distributed) |
| Flexibility | Fixed routing logic | Adaptive routing |
| Best for | Structured, hierarchical tasks | Flexible, collaborative tasks |
When to Use Each
Use Swarm Pattern when:
- Tasks benefit from flexible, adaptive routing
- No clear hierarchical structure exists
- Agents need rich context from all peers
- Emergent behavior is desired
- Decentralized decision-making is preferred
- Network topology is important
Use Supervisor Pattern when:
- Clear task structure and agent roles
- Centralized control and monitoring needed
- Predictable routing is important
- Quality control via central oversight
- Simpler to reason about and debug
- Sequential or parallel agent execution
Hybrid Approach:
- Use hierarchical teams where each team is a swarm
- Combine supervisor coordination with swarm sub-teams
- Use swarms for exploration, supervisors for orchestration
Common Patterns
Research Swarm
agents = [
SwarmAgent(name="researcher", connections=["analyst", "fact_checker"]),
SwarmAgent(name="analyst", connections=["writer"]),
SwarmAgent(name="fact_checker", connections=["analyst"]),
SwarmAgent(name="writer", connections=[]),
]Analysis Pipeline
agents = [
SwarmAgent(name="collector", connections=["preprocessor"]),
SwarmAgent(name="preprocessor", connections=["analyzer"]),
SwarmAgent(name="analyzer", connections=["reporter", "validator"]),
SwarmAgent(name="validator", connections=["reporter"]),
SwarmAgent(name="reporter", connections=[]),
]Collaborative Editing
agents = [
SwarmAgent(name="drafter", connections=["reviewer", "editor"]),
SwarmAgent(name="reviewer", connections=["editor", "drafter"]),
SwarmAgent(name="editor", connections=["reviewer", "finalizer"]),
SwarmAgent(name="finalizer", connections=[]),
]Related Patterns
- Multi-Agent Collaboration - Supervisor pattern
- Hierarchical Teams - Nested team structures
- Subgraph Patterns - Composable graph components
Quiz
Test your understanding of agent swarm/network patterns:
Knowledge Check
What is the key characteristic that distinguishes swarm pattern from supervisor pattern?
Knowledge Check
What does the 'connections' field in SwarmAgent configuration represent?
Knowledge Check
When should you prefer swarm pattern over supervisor pattern?
Knowledge Check
What network topology would you use if every agent needs to be able to route to every other agent?
Knowledge Check
How does context sharing differ between swarm and supervisor patterns?