stigmergy-scheduler 0.1.2

Pheromone-field coordination for LLM multi-agent systems

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.

Based on arXiv:2601.08129. 7.9x faster than sequential execution at scale (100 tasks, 20 agents, real LLM calls).

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

Method	Description
StigmergyScheduler(wake_threshold, tick_interval, ...)	Create scheduler with tunable pressure thresholds
scheduler.set_dispatcher(fn)	Set the async function called when an agent wakes
await scheduler.add_task(TaskSpec(...))	Queue a task with dependencies and priority
await scheduler.inject_urgent(TaskSpec(...))	Force-wake an agent immediately
await scheduler.run()	Block until all tasks complete or stop() is called
scheduler.start()	Run in background, returns asyncio.Task
scheduler.stop()	Graceful shutdown after current tick
scheduler.get_metrics()	Snapshot of ticks, completions, failures, pressure map

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: [e-mail]
• Blog: blog.productiongrade.tech

License

MIT