Metadata-Version: 2.4
Name: stigmergy-scheduler
Version: 0.1.0
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: Riverstone Labs
License-Expression: MIT
Keywords: agents,coordination,multi-agent,pheromone,stigmergy,swarm
Classifier: Development Status :: 3 - Alpha
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

**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](https://pypi.org/project/stigmergy-langgraph/), [AutoGen](https://pypi.org/project/stigmergy-autogen/), and [CrewAI](https://pypi.org/project/stigmergy-crewai/).

Based on [arXiv:2601.08129](https://arxiv.org/abs/2601.08129) -- pheromone-inspired signal fields that outperform hierarchical control by 30x on solve rates.

## Quick Start

### Scheduler (batteries included)

```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
```

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)

```python
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:

```python
# 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

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

## 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
