Metadata-Version: 2.4
Name: stigmergy-scheduler
Version: 0.1.2
Summary: Pheromone-field coordination for LLM multi-agent systems
Project-URL: Homepage, https://github.com/Production-Grade/stigmergy
Project-URL: Documentation, https://github.com/Production-Grade/stigmergy/blob/main/docs/integrations.md
Project-URL: Repository, https://github.com/Production-Grade/stigmergy
Project-URL: Issues, https://github.com/Production-Grade/stigmergy/issues
Author: Production Grade
License-Expression: MIT
Keywords: agents,coordination,multi-agent,pheromone,stigmergy,swarm
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Requires-Python: >=3.11
Provides-Extra: dev
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: redis>=5.0; extra == 'dev'
Provides-Extra: redis
Requires-Dist: redis>=5.0; extra == 'redis'
Description-Content-Type: text/markdown

# 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](https://arxiv.org/abs/2601.08129). 7.9x faster than sequential execution at scale (100 tasks, 20 agents, real LLM calls).

## Install

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

## Quick start

```python
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`](https://pypi.org/project/stigmergy-langgraph/) -- replaces conditional edge routing
- [`stigmergy-autogen`](https://pypi.org/project/stigmergy-autogen/) -- replaces LLM-based speaker selection
- [`stigmergy-crewai`](https://pypi.org/project/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:

```python
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](https://github.com/warwickmcintosh) at [Production Grade](https://blog.productiongrade.tech).

- GitHub: [warwickmcintosh](https://github.com/warwickmcintosh)
- X: [@warwickmcintosh](https://x.com/warwickmcintosh)
- Email: contact@productiongrade.tech
- Blog: [blog.productiongrade.tech](https://blog.productiongrade.tech)

## License

MIT
