Docs

Reactive blackboard for agents

Yoltra's headline use case for Python is a reactive blackboard for agentic and plugin systems: a single shared working-memory that tools, monitors, and sub-agents observe at the exact granularity they care about, coordinated by async reactions instead of polling or hand-threaded callbacks.

#The problem

Agent frameworks (LangGraph, CrewAI, AutoGen, LlamaIndex, …) hold coarse state. None let you say "react when plan.steps[2].status changes." Coordination ends up as polling loops or tangled callbacks.

#The pattern

Model working memory as a slice tree — plan, scratchpad, tool_results, sub_agents — and let each participant:

connect to the precise paths it owns (a monitor on plan.steps.*.status),
emit events others fold in (a planner emitting ("plan", "step_done", i)),
react via effects that fire the next action.

The structural diff wakes only the relevant subscribers, and the event log gives you a replayable trace of the run.

#A worked example

A planner advances a plan; a monitor reacts to each step's status without polling; an effect escalates on failure.

python

from yoltra import create_store

# inline effects_mode keeps this script's output deterministic; use the default
# "auto" mode (below) for real async reactions.
store = create_store("Agent", effects_mode="inline")

@store.reducer(
    name="plan",
    state={"steps": [
        {"name": "research", "status": "pending"},
        {"name": "draft", "status": "pending"},
        {"name": "review", "status": "pending"},
    ]},
    when={"keys": [("plan", "start"), ("plan", "complete"), ("plan", "fail")]},
)
def plan(prev, event):
    steps = [dict(s) for s in prev["steps"]]
    idx = event.payload
    if event.type == "start":
        steps[idx]["status"] = "running"
    elif event.type == "complete":
        steps[idx]["status"] = "done"
    elif event.type == "fail":
        steps[idx]["status"] = "failed"
    return {"steps": steps}

# A monitor reacts to any step's status changing — no polling.
def on_status(change):
    print(f"{change.path}: {change.old_value} → {change.new_value}")

store.connect(reducer="plan", property="steps.*.status", handler=on_status)

# An effect escalates when a step fails.
@store.effect(when={"keys": [("plan", "fail")]})
def escalate(event, get_state, emit):
    step = get_state()["plan"]["steps"][event.payload]
    print(f"[escalate] step {step['name']!r} failed — notifying supervisor")

# Drive the plan.
store.emit("plan", "start", 0)
store.emit("plan", "complete", 0)
store.emit("plan", "start", 1)
store.emit("plan", "fail", 1)

Output:

steps.0.status: pending → running
steps.0.status: running → done
steps.1.status: pending → running
steps.1.status: running → failed
[escalate] step 'draft' failed — notifying supervisor

The monitor never polls; it wakes only when a status leaf actually changes, and the escalation effect runs after the failure commits.

#Multiple participants, one store

Because subscriptions are path-scoped, many agents share one store without coupling:

python

# A tool writes results into its own region.
@store.reducer(name="tool_results", state={},
               when={"keys": [("tool", "result")]})
def tool_results(prev, event):
    name, value = event.payload
    return {**prev, name: value}

# A sub-agent only watches the results it depends on.
store.connect(reducer="tool_results", property="search",
              handler=lambda c: print("search result ready:", c.new_value))

store.emit("tool", "result", ("search", ["doc-1", "doc-2"]))
# search result ready: ['doc-1', 'doc-2']

Each participant subscribes to the regions it cares about and emits events others fold in — the blackboard/knowledge-source pattern, made reactive.

#Async reactions

Reactions that do I/O (call a model, hit a tool server) belong in effects, which run after commit and can be async. Async effects need the default "auto" mode (or "asyncio"), not "inline":

python

agent = create_store("Agent")   # default "auto" mode

@agent.effect(when={"keys": [("plan", "complete")]})
async def kick_off_next(event, get_state, emit):
    idx = event.payload
    steps = get_state()["plan"]["steps"]
    if idx + 1 < len(steps):
        emit("plan", "start", idx + 1)   # re-entrant emit; never deadlocks

Re-entrant emits from within an effect are safe — each event's effects are their own task.

#Replayable traces

Because every write is an event, a run is a sequence you can capture and replay through the reducers for deterministic debugging (gated behind allow_replay):

python

store = create_store("Agent", allow_replay=True)
# ... capture the (channel, type, payload) events you emit ...
store._replay_events(snapshot, captured_events)   # re-run through reducers only

#Why this fits

The triple of atomic subscriptions + async effects + replay maps directly onto the blackboard/knowledge-source pattern that agent coordination needs — and Python didn't previously have a framework-agnostic primitive for it.

#See also

Developer guide — the full model.
Quick start — the five-minute version.

Edit this page on GitHub