Swarm Agents¶
Overview¶
One agent is smart. Ten agents coordinating without central control can be smarter.
Swarm systems are how we solve truly hard problems.
Swarm Principles¶
Core Properties¶
class SwarmAgent:
"""Individual agent in swarm"""
def __init__(self, agent_id, swarm):
self.id = agent_id
self.swarm = swarm
self.local_knowledge = {}
def execute_swarm_task(self, task):
"""Participate in swarm problem-solving"""
# Principle 1: Local computation
local_solution = self.solve_locally(task)
# Principle 2: Information sharing
self.swarm.broadcast(self.id, local_solution)
# Principle 3: Collective decision
collective_solution = self.swarm.aggregate()
# Principle 4: Coordination without central control
next_task = self.swarm.determine_next_work()
return collective_solution
def solve_locally(self, task):
"""Each agent solves its own piece"""
# I don't solve the whole problem
# I solve my part well
my_subtask = task.get_my_portion(self.id)
solution = self.solve(my_subtask)
return solution
-
Swarm Coordination Patterns¶
Pattern 1: Voting/Consensus¶
class VotingSwarm:
"""Agents vote on best solution"""
def solve_with_voting(self, task):
"""Multiple solutions, vote on best"""
solutions = []
# Each agent proposes solution
for agent in self.agents:
solution = agent.propose_solution(task)
solutions.append(solution)
# Rank solutions
rankings = []
for voter in self.agents:
ranking = voter.evaluate_solutions(solutions)
rankings.append(ranking)
# Aggregate votes
winner = self.aggregate_votes(rankings)
return solutions[winner]
def aggregate_votes(self, rankings):
"""Use voting theory (Borda, plurality, etc)"""
# Borda count
scores = {}
for ranking in rankings:
for position, solution_id in enumerate(ranking):
if solution_id not in scores:
scores[solution_id] = 0
scores[solution_id] += len(ranking) - position
return max(scores, key=scores.get)
Pattern 2: Pheromone (Information Sharing)¶
class PheromoneSwarm:
"""Ants leave pheromones, others follow"""
def __init__(self):
self.pheromone_map = {} # Path → strength
self.evaporation_rate = 0.1
def execute_swarm_search(self, task):
"""Each agent explores, leaves pheromone"""
for iteration in range(100):
# Each agent searches independently
for agent in self.agents:
path = agent.search(task, self.pheromone_map)
quality = self.evaluate_path(path)
# Leave pheromone (strong if good path)
self.deposit_pheromone(path, quality)
# Evaporate pheromone (forget old info)
self.evaporate_pheromone()
# Check if converged
if self.is_converged():
break
# Best path has strongest pheromone
return self.best_path()
def deposit_pheromone(self, path, quality):
"""Mark good paths"""
for edge in path.edges:
if edge not in self.pheromone_map:
self.pheromone_map[edge] = 0
self.pheromone_map[edge] += quality
def evaporate_pheromone(self):
"""Forget old information"""
for edge in self.pheromone_map:
self.pheromone_map[edge] *= (1 - self.evaporation_rate)
Pattern 3: Particle Swarm Optimization¶
class ParticleSwarmOptimization:
"""Particles move toward optima"""
def __init__(self, num_particles=30):
self.particles = [Particle() for _ in range(num_particles)]
self.global_best = None
def optimize(self, objective_function, iterations=100):
"""Find optimum through swarm"""
for iteration in range(iterations):
for particle in self.particles:
# Evaluate current position
fitness = objective_function(particle.position)
# Update personal best
if fitness > particle.best_fitness:
particle.best = particle.position
particle.best_fitness = fitness
# Update global best
if fitness > self.global_best.fitness:
self.global_best = particle.position
# Update velocities
for particle in self.particles:
# Move toward personal best
cognitive = particle.best - particle.position
# Move toward global best
social = self.global_best - particle.position
# Update velocity and position
particle.velocity = (
0.7 * particle.velocity +
0.1 * cognitive +
0.2 * social
)
particle.position += particle.velocity
return self.global_best
Scaling Swarms¶
Managing Large Swarms¶
class LargeScaleSwarm:
"""Coordinate 100-1000+ agents"""
def __init__(self, num_agents=1000):
self.agents = [Agent(i) for i in range(num_agents)]
def execute_at_scale(self, task):
"""Coordinate massive swarm efficiently"""
# Partition task
subtasks = task.partition_for_swarm()
# Distribute work
assignments = self.load_balance(subtasks)
results = []
for agent, subtask in assignments:
result = agent.execute(subtask)
results.append(result)
# Aggregate efficiently
final_result = self.hierarchical_aggregate(results)
return final_result
def load_balance(self, subtasks):
"""Assign work fairly"""
# Estimate work per subtask
work_estimates = [self.estimate_work(t) for t in subtasks]
# Sort agents by availability
agent_queue = sorted(
self.agents,
key=lambda a: a.current_load
)
# Greedy assignment
assignments = []
for subtask, work in zip(subtasks, work_estimates):
agent = agent_queue.pop(0)
assignments.append((agent, subtask))
agent.current_load += work
agent_queue.sort(key=lambda a: a.current_load)
return assignments
3 Warnings¶
Warning 1: Swarm Overconfidence¶
# WRONG
# Assume 10 agents = 10x capability
swarm = [Agent() for _ in range(10)]
result = swarm.solve(hard_problem)
# But coordination overhead!
# Actual speedup
# RIGHT
# Measure actual speedup
speedup = benchmark_swarm_vs_single()
# 6.5x for 10 agents
# Use realistic expectations
Warning 2: Consensus Breakdown¶
# WRONG
# Assume agents will reach consensus
result = swarm.vote_on_solution()
# But agents diverge on hard problems!
# Voting fails when agents are split
# RIGHT
# Add tie-breaking mechanism
result = swarm.vote_on_solution()
if len(result.clusters) > 1:
use_tiebreaker(result)
else:
return result
Warning 3: Cascading Errors¶
# WRONG
# All agents believe same wrong thing
agent_1_error = mistake()
agent_2 = learn_from(agent_1) # Learn the mistake!
agent_3 = learn_from(agent_2) # Spread the mistake!
# Wrong belief cascades through swarm
# RIGHT
# Independent verification
agent_1_solution = agent_1.solve()
external_check = verify(agent_1_solution)
if external_check.wrong:
flag_as_incorrect()
don't_let_spread()
-
Last Updated: August 9, 2026