stigmergy-scheduler 0.1.0

Pheromone-field coordination for LLM multi-agent systems

stigmergy

Pressure-driven scheduling for multi-agent AI systems. Agents wake when there's work to do, not on fixed timers. Dependencies resolve themselves. Failed tasks retry automatically.

Add a scheduler to your agents in 6 lines. No orchestration code, no polling loops, no cron. Works standalone or drops into LangGraph, AutoGen, and CrewAI.

Based on arXiv:2601.08129 -- pheromone-inspired signal fields that outperform hierarchical control by 30x on solve rates.

Quick Start

Scheduler (batteries included)

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

The scheduler handles tick loops, signal decay, dependency resolution, error recovery, and metrics. You just provide the dispatch function.

Low-level field (build your own)

from stigmergy import PheromoneField, InMemoryFieldBackend

field = PheromoneField(InMemoryFieldBackend())

# Agent deposits a signal
await field.deposit("agent-1", "work_claim", 0.8, {"task": "market research"})

# Another agent reads aggregate pressure
pressure = await field.read_pressure("agent-2")

# Periodic decay (evaporation)
evaporated = await field.decay()

Core Concepts

Signals

The atomic unit of communication. A signal has:

• agent_id — who deposited it
• signal_type — category (e.g. work_claim, need_help, insight_deposit)
• intensity — strength from -1.0 (repulsion) to 1.0 (attraction)
• metadata — arbitrary attached data
• target_agent_id — optional targeting for directed pressure

PheromoneField

The shared state space. Implements the four classical self-organisation mechanisms:

1. Positive feedback — amplify() reinforces valuable signals
2. Negative feedback — decay() evaporates stale signals
3. Randomness — StochasticDecay adds noise to prevent local optima
4. Multiple interactions — many agents reading/writing creates emergence

Decay Strategies

• ExponentialDecay — signals lose half intensity every N seconds (default)
• LinearDecay — fixed rate reduction per second
• StepDecay — full intensity until TTL, then instant evaporation
• StochasticDecay — exponential with random noise

Backends

• InMemoryFieldBackend — for testing and single-process deployments
• RedisFieldBackend — for production multi-process setups

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

Self-Scheduling Pattern

For custom scheduling beyond what StigmergyScheduler provides, use the field directly:

# Each tick (e.g., every 30 seconds):
for agent in agents:
    pressure = await field.read_pressure(
        agent.id,
        weights={"event": 0.3, "goal": 0.25, "peer": 0.15, "time": 0.3}
    )
    if pressure > agent.wake_threshold:
        await wake_agent(agent)

# Apply decay each tick
await field.decay()

Agents wake when accumulated pressure (from events, goals, peer requests, time) exceeds their threshold — no fixed intervals needed.

Install

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

Author

Built by Warwick McIntosh at Production Grade.

• GitHub: warwickmcintosh
• X: @warwickmcintosh
• Email: [e-mail]
• Blog: blog.productiongrade.tech

License

MIT