stigmergy-langgraph 0.1.1

Stigmergy pressure-field scheduling adapter for LangGraph

stigmergy-langgraph

Drop-in replacement for LangGraph's conditional edge routing. Register agents and tasks with dependencies, and stigmergy figures out execution order using pressure signals. No more router functions that grow into unmanageable if/else trees.

Works with your existing LangGraph node functions. 8 lines to integrate.

Install

pip install stigmergy-langgraph

Usage

from stigmergy_langgraph import StigmergyRouter, PressureState

# Define your agent node functions (standard LangGraph nodes)
def writer_node(state):
    return {"output": "Draft written"}

def reviewer_node(state):
    return {"output": "Review complete"}

# Create the stigmergy router
router = StigmergyRouter(wake_threshold=0.4, decay_half_life=300)
router.register_agent("writer", writer_node)
router.register_agent("reviewer", reviewer_node)

# Add tasks with dependencies
router.add_task("draft", agent="writer", priority=0.8)
router.add_task("review", agent="reviewer", priority=0.6, deps=["draft"])

# Build and run
graph = router.build_graph(PressureState)
result = graph.invoke({
    "messages": [],
    "completed_tasks": [],
    "pending_tasks": ["draft", "review"],
    "outputs": {},
})

How It Works

Instead of hardcoded conditional edges, the StigmergyRouter:

1. Deposits pressure signals for each task (intensity = priority)
2. On each graph step, decays signals and evaluates pressure per agent
3. Dispatches the highest-pressure agent whose task dependencies are met
4. Deposits completion signals that enable downstream tasks

When to Use

Use this instead of manual conditional edges when:

• You have 3+ agents with complex dependency graphs
• Tasks arrive dynamically (can't hardcode the routing)
• You want automatic parallelization of independent tasks

Stick with native LangGraph routing for simple 2-agent pipelines.

License

MIT