swarmagentkit

A tick-based / real-time multi-agent swarm runtime with a hierarchical memory model (per-agent memory and emergent collective memory), a hybrid LLM/rule-based orchestrator with automatic transparent fallback, and pluggable domain physics — so the same runtime drives a market simulation, a particle system, a geopolitics model, a fleet of warehouse robots, or a real drone swarm, by swapping one small piece instead of rewriting the engine.

pip install swarmagentkit

No required third-party dependency beyond Pydantic. No API key needed to get started — the offline rule-based mode runs with zero network access or credentials.

Why this exists

Most "multi-agent" libraries fall into one of two buckets:

• Workflow orchestrators (LangGraph, and similar) are excellent at planner → tool → critic pipelines with a handful of agents and explicit control flow, but they don't give you a swarm's emergent social dynamics — a social graph that evolves, a collective mood the population senses and reacts to, hundreds/thousands of agents where only a bounded subset act per step.
• Social-simulation frameworks give you the swarm dynamics, but are simulation-only by design — there's no path from "10,000 agents in a sandbox" to "12 drones in the sky" without rewriting your entire stack.

swarmagentkit is the middle layer: a swarm engine general enough to sit under a workflow orchestrator (drive it from a LangGraph node — see examples/langgraph_bridge/) and general enough to sit on top of real hardware (drive real sensors/actuators — see examples/drone_swarm/ and docs/realtime.md) without changing your agent logic either way.

Quickstart

import random
from swarmagentkit import Agent, Event, SwarmRunner, build_world, spawn_agents

# 1. Population: how many agents, of what role — one call.
agents = spawn_agents("node", 20, allowed_actions=["maintain", "ping"])
world = build_world(agents)

# 2. Policy: decide what one agent does. This can be a plain function...
def policy(observation: dict, rng: random.Random) -> dict:
    return {"action": "ping", "params": {}, "rationale": "demo heartbeat"}

# 3. Physics: decide what an action actually DOES to the world. This is the
#    one domain-specific piece — everything else in swarmagentkit is generic.
class DemoPhysics:
    def apply(self, world, agent_id, action, params, rng):
        world.metrics["pings"] = world.metrics.get("pings", 0) + 1
        return Event(tick=world.tick, type=action, actor_id=agent_id, public=True)

runner = SwarmRunner(world, DemoPhysics(), policy, max_active=20, seed=42)
result = runner.run(ticks=10)

print(result.ticks_run, result.event_count, result.final_metrics)

That's it — no LLM, no API key, fully deterministic given the seed. Bring an LLM in whenever you want one (see below); swap DemoPhysics for your own domain (see examples/).

Bring any LLM — or none at all — with full control over which one does what

from swarmagentkit.brains import make_brain, RoleRouterBrain, OfflineBrain

# One line per provider, hosted or local:
gpt      = make_brain("openai", "gpt-4o-mini")                     # reads OPENAI_API_KEY
claude   = make_brain("anthropic", "claude-sonnet-4-5")            # reads ANTHROPIC_API_KEY
router_m = make_brain("openrouter", "meta-llama/llama-3.1-70b")    # reads OPENROUTER_API_KEY
local    = make_brain("ollama", "llama3.1:8b")                     # fully local, no key needed
custom   = make_brain("custom", "my-model", base_url="https://llm.internal/v1", api_key="...")

# Mix them freely per role / per specific agent / for the orchestrator /
# for the end-of-run analysis — one router object, full control:
brain = RoleRouterBrain(
    default=OfflineBrain(policy),          # cheap fallback for most agents
    by_role={"buyer": local, "seller": gpt},
    by_agent={"vip_007": claude},          # override one specific agent
    orchestrator_brain=router_m,
    analyze_brain=gpt,
)

runner = SwarmRunner(world, DemoPhysics(), policy, brain=brain, use_llm=True, ...)

Every live brain call automatically, transparently falls back to your offline policy function on any failure (bad key, timeout, malformed JSON, rate limit) — a run never stalls or crashes because of one flaky API call, and the report tells you exactly when a fallback happened (result.orchestrator_mode).

Full control over population size and per-tick attention

These are two separate knobs, on purpose:

from swarmagentkit import spawn_agents, build_world
from swarmagentkit.builders import resize_role, add_agents, remove_agents

agents = spawn_agents("buyer", 5000)     # total population: one number
world = build_world(agents)

resize_role(world, "buyer", 200)          # shrink live, mid-run, to 200
add_agents(world, spawn_agents("seller", 50, start_index=0))

runner = SwarmRunner(world, physics, policy, max_active=30)  # only 30 act per tick

max_active bounds cost/latency regardless of how large the population is — a world can hold thousands of agents while only a small, orchestrator-chosen subset actually thinks and acts on any given tick.

Simulation AND real-world actuation, same code

from swarmagentkit.realtime import RealtimeSwarmRunner

# Same policy_fn / brain / orchestrator you validated in simulation.
# Only the sensor/actuator adapters are new — see docs/realtime.md.
runner = RealtimeSwarmRunner(agents, MySensorAdapter(), MyActuatorAdapter(), policy, loop_hz=5.0)
runner.run_forever()   # a real wall-clock loop; leave it running for months if you want to

See docs/realtime.md for drone (MAVSDK/PX4), warehouse robot (ROS2), and IoT (MQTT) adapter sketches.

Use it with LangGraph (fully optional)

pip install "swarmagentkit[langgraph]"

from swarmagentkit.adapters.langgraph_node import make_swarm_node

swarm_node = make_swarm_node(runner, ticks_per_call=5)
graph.add_node("run_swarm", swarm_node)   # then wire it into your StateGraph as usual

swarmagentkit has zero import-time dependency on LangGraph — the bridge module only matters if and when you import it.

What's in the box

Full architecture write-up: docs/concepts.md.

Non-goals

To be upfront about scope:

• Not a workflow orchestrator. If you need explicit planner → tool → critic control flow for a handful of agents, use LangGraph (and optionally drive a swarm from one of its nodes — see above).
• Not a physics/rigid-body engine. Physics.apply() is where you decide what an action means; swarmagentkit doesn't simulate collisions, forces, or continuous dynamics for you.
• Not a flight-control or motor-control stack. For real hardware, swarmagentkit sits above your flight controller / ROS2 stack, deciding swarm-level behavior — it does not replace PX4, ArduPilot, or a robot's low-level control loop.
• Not (yet) a distributed/multi-process runtime. SwarmRunner and RealtimeSwarmRunner run in a single process. Sharding across machines is on the roadmap, not in v0.1.

Comparison

This table is meant descriptively, not as a ranking — these tools solve different problems and compose well together (see the LangGraph bridge above).

Contributing

See CONTRIBUTING.md. Issues and PRs welcome — especially new domain examples under examples/.

License

Apache License 2.0.