stigmergy-scheduler

Your multi-agent system doesn't need a manager. It needs a pressure field.

Most agent orchestration boils down to three bad options: polling on timers (wasteful), hardcoded if/else routing (brittle), or burning an LLM call to pick who goes next (slow and expensive). All three put coordination logic outside the system it's coordinating.

Stigmergy replaces them with a single primitive: a pressure field. Events deposit signals. Signals decay over time. When accumulated pressure crosses a threshold, the agent wakes and does the work. Dependencies resolve themselves through completion signals. Failed tasks create retry pressure automatically.

9.5x faster than LangGraph at scale. 120 tasks across 30 agents in 2.1 seconds -- same workload takes LangGraph 20 seconds. The gap widens with agent count: stigmergy dispatches all available agents per tick, graph-based frameworks batch a fixed few per step.

Scheduling overhead is sub-millisecond at 1,000 tasks. Region-indexed storage and bounded signal accumulation mean it stays flat at 10,000+. The ceiling is your LLM API, not the scheduler.

Install

pip install stigmergy-scheduler              # core (in-memory backend)
pip install stigmergy-scheduler[redis]       # with Redis backend

Quick start

from stigmergy import StigmergyScheduler, TaskSpec, DispatchResult

scheduler = StigmergyScheduler(wake_threshold=0.4)

async def dispatch(agent_id, task):
    result = await my_llm_call(task.description)
    return DispatchResult(success=True, output=result)

scheduler.set_dispatcher(dispatch)

await scheduler.add_task(TaskSpec(
    id="draft", description="Write first draft",
    agent_id="writer", priority=0.8,
))
await scheduler.add_task(TaskSpec(
    id="review", description="Review the draft",
    agent_id="reviewer", dependencies=["draft"], priority=0.6,
))

await scheduler.run()  # blocks until all tasks complete

You provide a dispatch function. Stigmergy handles the tick loop, signal decay, dependency resolution, error recovery, and metrics.

Framework integrations

Already using a framework? Drop stigmergy in as a scheduling layer:

• stigmergy-langgraph -- replaces conditional edge routing
• stigmergy-autogen -- replaces LLM-based speaker selection
• stigmergy-crewai -- replaces manual task ordering

Low-level field (build your own scheduler)

The StigmergyScheduler is batteries-included, but you can use the pressure field directly for custom scheduling:

from stigmergy import PheromoneField, InMemoryFieldBackend

field = PheromoneField(InMemoryFieldBackend())

await field.deposit("agent-1", "work_claim", 0.8, {"task": "market research"})
pressure = await field.read_pressure("agent-2")
evaporated = await field.decay()

Scheduler API

Core concepts

Signals are the atomic unit. Each has an agent_id, signal_type, intensity (-1.0 to 1.0), metadata, and optional target_agent_id.

PheromoneField is the shared state space implementing four self-organisation mechanisms: positive feedback (amplify), negative feedback (decay), randomness (stochastic decay), and multiple interactions (emergence from many agents).

Decay strategies: ExponentialDecay (default), LinearDecay, StepDecay, StochasticDecay.

Backends: InMemoryFieldBackend (testing/single-process), RedisFieldBackend (production/multi-process).

Author

Built by Warwick McIntosh at Production Grade.

• GitHub: warwickmcintosh
• X: @warwickmcintosh
• Email: contact@productiongrade.tech
• Blog: blog.productiongrade.tech

License

MIT