YoltraYoltra
TSPyDS
Documentación

Pizarra reactiva para agentes

El caso de uso estelar de Yoltra en Python es una pizarra reactiva (reactive blackboard) para sistemas agénticos y de plugins: una única memoria de trabajo compartida que las herramientas, monitores y sub-agentes observan con la granularidad exacta que les importa, coordinada por reacciones asíncronas en lugar de sondeo (polling) o callbacks enhebrados a mano.

#El problema

Los frameworks de agentes (LangGraph, CrewAI, AutoGen, LlamaIndex, …) mantienen estado grueso. Ninguno te permite decir "reacciona cuando cambie plan.steps[2].status". La coordinación termina como bucles de sondeo o callbacks enredados.

#El patrón

Modela la memoria de trabajo como un árbol de slices — plan, scratchpad, tool_results, sub_agents — y deja que cada participante:

  • connect a las rutas precisas que le corresponden (un monitor sobre plan.steps.*.status),
  • emit eventos que otros pliegan (un planificador que emite ("plan", "step_done", i)),
  • reaccione vía efectos que disparan la siguiente acción.

La comparación estructural despierta solo a los suscriptores relevantes, y el log de eventos te da una traza reproducible de la corrida.

#Un ejemplo trabajado

Un planificador avanza un plan; un monitor reacciona al estado de cada paso sin sondear; un efecto escala ante un fallo.

python
from yoltra import create_store

# effects_mode "inline" mantiene determinista la salida de este script; usa el
# modo "auto" por defecto (más abajo) para reacciones asíncronas reales.
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}

# Un monitor reacciona a cualquier cambio de status de un paso — sin sondeo.
def on_status(change):
    print(f"{change.path}: {change.old_value} → {change.new_value}")

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

# Un efecto escala cuando un paso falla.
@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")

# Ejecuta el plan.
store.emit("plan", "start", 0)
store.emit("plan", "complete", 0)
store.emit("plan", "start", 1)
store.emit("plan", "fail", 1)

Salida:

ts
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

El monitor nunca sondea; despierta solo cuando una hoja status realmente cambia, y el efecto de escalada corre después de que el fallo se confirma.

#Múltiples participantes, un solo store

Como las suscripciones tienen alcance por ruta, muchos agentes comparten un store sin acoplarse:

python
# Una herramienta escribe resultados en su propia región.
@store.reducer(name="tool_results", state={},
               when={"keys": [("tool", "result")]})
def tool_results(prev, event):
    name, value = event.payload
    return {**prev, name: value}

# Un sub-agente solo observa los resultados de los que depende.
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']

Cada participante se suscribe a las regiones que le importan y emite eventos que otros pliegan — el patrón de pizarra/fuente-de-conocimiento, vuelto reactivo.

#Reacciones asíncronas

Las reacciones que hacen E/S (llamar a un modelo, pegarle a un servidor de herramientas) van en efectos, que corren después de la confirmación y pueden ser async. Los efectos async requieren el modo "auto" por defecto (o "asyncio"), no "inline":

python
agent = create_store("Agent")   # modo "auto" por defecto

@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)   # emit reentrante; nunca causa interbloqueo

Los emits reentrantes desde dentro de un efecto son seguros — los efectos de cada evento son su propia tarea.

#Trazas reproducibles

Como cada escritura es un evento, una corrida es una secuencia que puedes capturar y reproducir a través de los reducers para una depuración determinista (protegido detrás de allow_replay):

python
store = create_store("Agent", allow_replay=True)
# ... captura los eventos (channel, type, payload) que emites ...
store._replay_events(snapshot, captured_events)   # re-ejecuta solo a través de los reducers

#Por qué encaja

El trío de suscripciones atómicas + efectos asíncronos + replay mapea directamente sobre el patrón de pizarra/fuente-de-conocimiento que necesita la coordinación de agentes — y Python no tenía antes una primitiva agnóstica al framework para ello.

#Ver también

Editar esta página en GitHub