Un blackboard para todo el sistema

Descripción

Las tres piezas de las lecciones 03, 04 y 05 —pipeline, fan-out, handoff— resuelven, cada una, una sub-tarea completa. Esta lección conecta la última pieza estructural del sistema, Blackboard (Módulo 6), sin ningún cambio, y confirma algo que ya estableció el Módulo 6 pero que vale la pena ver, una vez más, sobre las tres demos nuevas de este capstone: no todos los patrones escriben al estado compartido por igual.

Vas a correr, en una sola corrida, la pieza de Luis (pipeline, SÍ escribe), la pieza de Marta (fan-out, NO escribe), y la pieza de Valentina (handoff, tampoco escribe) — y vas a ver, con el WRITE_SEQ del Blackboard como testigo, que el número de secuencia no avanza ni una sola vez durante las piezas de Marta y Valentina.

Conexión con el módulo

Blackboard, WriteLogEntry y WRITE_SEQ son exactamente la misma clase del Módulo 6 — mismos campos (member, room, tier, hours, price_cents, booking_id, log), mismo método write. Esta lección no le agrega nada — reusa las tres piezas ya conectadas de las lecciones 03, 04 y 05 para confirmar, con las tres corriendo en el mismo proceso, la regla de "quién escribe" que M6 estableció con la petición de Ana. La lección 08 retoma exactamente este mismo Blackboard para la Demo C completa.

Analogía: la pizarra de la cocina, un turno con tres mesas distintas

La pizarra de la cocina de M6 no se llena con todo lo que pasa en el restaurante — solo con lo que el resto de la cocina podría necesitar después: qué mesa pidió qué, si ya se confirmó. Esta lección es un turno completo con tres mesas de forma distinta: la mesa de Luis, que termina en una reserva real (eso SÍ va a la pizarra); la mesa de Marta, que solo pidió comparar precios y preguntar por una política general (nada de eso es una reserva, nada va a la pizarra); y la mesa de Valentina, que solo cotizó sin confirmar (tampoco va a la pizarra, todavía).

Ejemplo trabajado: tres socios, un mismo Blackboard

import ast
import concurrent.futures
import itertools
from dataclasses import dataclass, field
import reservo_tools as rt


def dispatch_parallel(tool_use_blocks, tools):
    with concurrent.futures.ThreadPoolExecutor(max_workers=len(tool_use_blocks)) as pool:
        futures = [pool.submit(tools[b["name"]], **b["input"]) for b in tool_use_blocks]
        results = [f.result() for f in futures]
    return [
        {"type": "tool_result", "tool_use_id": b["id"], "content": str(r)}
        for b, r in zip(tool_use_blocks, results)
    ]


def run_agent_parallel(question, model_script, tools, max_iterations=10):
    messages = [{"role": "user", "content": question}]
    for step in range(max_iterations):
        turn = model_script[step]
        messages.append({"role": "assistant", "content": turn["content"]})
        if turn["stop_reason"] != "tool_use":
            return turn, messages
        tool_result_blocks = dispatch_parallel(turn["content"], tools)
        messages.append({"role": "user", "content": tool_result_blocks})
    raise RuntimeError(f"max_iterations alcanzado ({max_iterations})")


def search_docs(query):
    q = query.lower()
    if "cancela" in q:
        return "[cancellation-policy] Puedes cancelar sin cargo hasta 2 horas antes del horario reservado."
    if "no" in q and ("present" in q or "show" in q or "llega" in q):
        return ("[no-show-policy] Si no te presentas a una reserva confirmada y no cancelas "
                 "con al menos 2 horas de anticipación, Reservo cobra el 50% del precio "
                 "cotizado como cargo por no-presentación.")
    return "No se encontró una política relevante para esa pregunta."


SPECIALISTS = {
    "booking_agent": {"tools": {"get_quote": rt.get_quote, "book_room": rt.book_room}},
    "policy_agent": {"tools": {"search_docs": search_docs}},
    "pricing_agent": {"tools": {"get_quote": rt.get_quote}},
}


def run_specialist(name, task, model_script):
    tools = SPECIALISTS[name]["tools"]
    return run_agent_parallel(task, model_script, tools)


# ---- M3: pipeline, sin cambios (lección 03) ----
@dataclass
class PipelineStage:
    kind: str
    name: str
    label: str


def last_tool_result(history):
    for m in reversed(history):
        if isinstance(m["content"], list):
            for b in m["content"]:
                if b["type"] == "tool_result":
                    try:
                        return ast.literal_eval(b["content"])
                    except (ValueError, SyntaxError):
                        return b["content"]
    return None


def build_stage_task(stage, payload):
    if stage.kind == "quote":
        return f"Cotiza {payload['room']} {payload['tier']} {payload['hours']}h."
    if stage.kind == "validate_policy":
        return (f"¿Cuál es la política de cancelación para una reserva de "
                 f"{payload['room']} de {payload['hours']}h, antes de confirmarla?")
    if stage.kind == "confirm":
        if payload.get("cleared_to_book"):
            return (f"Reserva {payload['room']} {payload['tier']} {payload['hours']}h "
                     f"para {payload['member']} -- la política de cancelación ya se validó.")
        return (f"Reserva {payload['room']} {payload['tier']} {payload['hours']}h "
                 f"para {payload['member']}.")
    raise ValueError(f"no sé armar la tarea de la etapa {stage.kind!r}")


def extract_payload(stage, history, payload):
    new_payload = dict(payload)
    result = last_tool_result(history)
    if stage.kind == "quote":
        new_payload["price_cents"] = result["price_cents"]
    elif stage.kind == "validate_policy":
        new_payload["cancellation_policy"] = result
        new_payload["cleared_to_book"] = True
    elif stage.kind == "confirm":
        new_payload["booking_id"] = result["booking_id"]
        new_payload["confirmed"] = result["confirmed"]
    return new_payload


def run_pipeline(stages, model_scripts, initial_payload):
    payload = dict(initial_payload)
    trace = []
    for i, stage in enumerate(stages):
        task = build_stage_task(stage, payload)
        final, history = run_specialist(stage.name, task, model_scripts[i])
        payload = extract_payload(stage, history, payload)
        trace.append({"stage": i + 1, "label": stage.label, "history": history})
    return payload, trace


# ---- M5: handoff, sin cambios (lección 05) ----
HANDOFF_TOOL_NAME = "handoff_to_specialist"


@dataclass
class HandoffPackage:
    sender: str
    receiver: str
    reason: str
    task: str
    context: dict = field(default_factory=dict)


def run_agent_with_handoff(question, model_script, tools, self_name, max_iterations=10):
    messages = [{"role": "user", "content": question}]
    for step in range(max_iterations):
        turn = model_script[step]
        messages.append({"role": "assistant", "content": turn["content"]})
        if turn["stop_reason"] != "tool_use":
            return turn, messages, None
        block = turn["content"][0]
        if block["name"] == HANDOFF_TOOL_NAME:
            inp = block["input"]
            package = HandoffPackage(
                sender=self_name, receiver=inp["receiver"], reason=inp["reason"],
                task=inp["task"], context=inp.get("context", {}),
            )
            return None, messages, package
        tool_result_blocks = dispatch_parallel(turn["content"], tools)
        messages.append({"role": "user", "content": tool_result_blocks})
    raise RuntimeError(f"max_iterations alcanzado ({max_iterations})")


def run_specialist_with_handoff(name, task, model_script):
    tools = SPECIALISTS[name]["tools"]
    return run_agent_with_handoff(task, model_script, tools, self_name=name)


def run_with_handoff(name, task, model_scripts):
    final, history, package = run_specialist_with_handoff(name, task, model_scripts[name])
    trace = [{"agent": name, "history": history, "package": package}]
    if package is None:
        return final, trace
    receiver_final, receiver_history, receiver_package = run_specialist_with_handoff(
        package.receiver, package.task, model_scripts[package.receiver],
    )
    trace.append({"agent": package.receiver, "history": receiver_history, "package": receiver_package})
    return receiver_final, trace


# ---- M6: Blackboard, sin cambios ----
WRITE_SEQ = itertools.count(1)


@dataclass
class WriteLogEntry:
    seq: int
    writer: str
    field: str
    value: object


@dataclass
class Blackboard:
    member: str | None = None
    room: str | None = None
    tier: str | None = None
    hours: int | None = None
    price_cents: int | None = None
    booking_id: int | None = None
    log: list = field(default_factory=list)

    def write(self, writer, **fields):
        for key, value in fields.items():
            setattr(self, key, value)
            self.log.append(
                WriteLogEntry(seq=next(WRITE_SEQ), writer=writer, field=key, value=value)
            )


bb = Blackboard()

print("=== 1) Luis (pipeline): booking_agent SÍ escribe -- hay una reserva real ===")
PIPELINE_STAGES = [
    PipelineStage(kind="quote", name="booking_agent", label="cotizar"),
    PipelineStage(kind="validate_policy", name="policy_agent", label="validar la política de cancelación"),
    PipelineStage(kind="confirm", name="booking_agent", label="confirmar la reserva"),
]
INITIAL_PAYLOAD = {"room": "Studio", "tier": "pro", "hours": 3, "member": "Luis"}
script_quote = [
    {"stop_reason": "tool_use", "content": [{"type": "tool_use", "id": "toolu_01", "name": "get_quote",
     "input": {"room": "Studio", "tier": "pro", "hours": 3}}]},
    {"stop_reason": "end_turn", "content": [{"type": "text", "text": "Studio pro 3h cuesta 9600 centavos."}]},
]
script_policy = [
    {"stop_reason": "tool_use", "content": [{"type": "tool_use", "id": "toolu_01", "name": "search_docs",
     "input": {"query": "política de cancelación"}}]},
    {"stop_reason": "end_turn", "content": [{"type": "text", "text": (
        "Puedes cancelar sin cargo hasta 2 horas antes del horario reservado. "
        "No hay ningún impedimento para confirmar.")}]},
]
script_confirm = [
    {"stop_reason": "tool_use", "content": [{"type": "tool_use", "id": "toolu_01", "name": "book_room",
     "input": {"room": "Studio", "tier": "pro", "hours": 3, "member": "Luis"}}]},
    {"stop_reason": "end_turn", "content": [{"type": "text", "text": "Reservé Studio pro 3h para Luis (confirmación #1)."}]},
]
bb.write("supervisor", member="Luis")
payload, _ = run_pipeline(PIPELINE_STAGES, [script_quote, script_policy, script_confirm], INITIAL_PAYLOAD)
bb.write("booking_agent", room=payload["room"], tier=payload["tier"],
         hours=payload["hours"], price_cents=payload["price_cents"])
bb.write("booking_agent", booking_id=payload["booking_id"])
print("Blackboard tras Luis:", bb)

print()
print("=== 2) Marta (fan-out): ni pricing_agent ni policy_agent escriben -- son comparaciones, no una reserva ===")
seq_before_marta = bb.log[-1].seq
script_pricing = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "get_quote", "input": {"room": "Studio", "tier": "pro", "hours": 2}},
        {"type": "tool_use", "id": "toolu_02", "name": "get_quote", "input": {"room": "Boardroom", "tier": "pro", "hours": 2}},
    ]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Studio pro 2h: 6400 centavos. Boardroom pro 2h: 12800 centavos. Studio es la opción más barata de las dos."}]},
]
run_specialist("pricing_agent", "Compara Studio y Boardroom pro 2h.", script_pricing)
print(f"seq del último write ANTES de Marta: {seq_before_marta} -- después de su fan-out: {bb.log[-1].seq} (sin cambio)")

print()
print("=== 3) Valentina (handoff): tampoco escribe -- el context queda local a la respuesta, no al Blackboard ===")
script_workshop_booking = [
    {"stop_reason": "tool_use", "content": [{"type": "tool_use", "id": "toolu_01", "name": "get_quote",
     "input": {"room": "Studio", "tier": "pro", "hours": 4}}]},
    {"stop_reason": "tool_use", "content": [{"type": "tool_use", "id": "toolu_02", "name": HANDOFF_TOOL_NAME,
     "input": {"receiver": "policy_agent", "reason": "pregunta de no-presentación, fuera de mi expertise",
               "task": "¿qué pasa si un miembro no se presenta a una reserva confirmada?",
               "context": {"room": "Studio", "tier": "pro", "hours": 4, "price_cents": 12800}}}]},
]
script_workshop_policy = [
    {"stop_reason": "tool_use", "content": [{"type": "tool_use", "id": "toolu_01", "name": "search_docs",
     "input": {"query": "qué pasa si no me presento a mi reserva"}}]},
    {"stop_reason": "end_turn", "content": [{"type": "text", "text": (
        "Si no te presentas a una reserva confirmada y no cancelas con al menos 2 horas de anticipación, "
        "Reservo cobra el 50% del precio cotizado como cargo por no-presentación.")}]},
]
run_with_handoff("booking_agent", "Cotiza Studio pro 4h. ¿Qué pasa si no llego?",
                  {"booking_agent": script_workshop_booking, "policy_agent": script_workshop_policy})
print(f"seq del último write tras Valentina: {bb.log[-1].seq} (sin cambio otra vez)")

print()
print("--- log completo del Blackboard (una sola corrida, 3 socios distintos) ---")
for entry in bb.log:
    print(f"  #{entry.seq} {entry.writer:<14} escribió {entry.field}={entry.value!r}")

Qué esperar:

=== 1) Luis (pipeline): booking_agent SÍ escribe -- hay una reserva real ===
Blackboard tras Luis: Blackboard(member='Luis', room='Studio', tier='pro', hours=3, price_cents=9600, booking_id=1, log=[WriteLogEntry(seq=1, writer='supervisor', field='member', value='Luis'), WriteLogEntry(seq=2, writer='booking_agent', field='room', value='Studio'), WriteLogEntry(seq=3, writer='booking_agent', field='tier', value='pro'), WriteLogEntry(seq=4, writer='booking_agent', field='hours', value=3), WriteLogEntry(seq=5, writer='booking_agent', field='price_cents', value=9600), WriteLogEntry(seq=6, writer='booking_agent', field='booking_id', value=1)])

=== 2) Marta (fan-out): ni pricing_agent ni policy_agent escriben -- son comparaciones, no una reserva ===
seq del último write ANTES de Marta: 6 -- después de su fan-out: 6 (sin cambio)

=== 3) Valentina (handoff): tampoco escribe -- el context queda local a la respuesta, no al Blackboard ===
seq del último write tras Valentina: 6 (sin cambio otra vez)

--- log completo del Blackboard (una sola corrida, 3 socios distintos) ---
  #1 supervisor     escribió member='Luis'
  #2 booking_agent  escribió room='Studio'
  #3 booking_agent  escribió tier='pro'
  #4 booking_agent  escribió hours=3
  #5 booking_agent  escribió price_cents=9600
  #6 booking_agent  escribió booking_id=1

Seis entradas en el log, todas de la mesa de Luis — ni Marta ni Valentina agregan una sola línea, aunque las dos corrieron especialistas reales, con tool calls reales, dentro del mismo proceso. El seq del Blackboard es un testigo honesto de esto: si Marta o Valentina hubieran escrito algo, bb.log[-1].seq habría avanzado más allá de 6 — y no avanza, confirmando en código lo que la prosa de esta lección afirma.

La pregunta 3 del criterio, aplicada tres veces

¿Algún dato que esta sub-tarea produce hace falta MÁS ADELANTE en la misma
corrida, para otra parte del sistema que todavía no sabemos quién va a leerlo?

  Luis (pipeline)   -> SÍ. La reserva confirmada (room, tier, hours, price_cents,
                        booking_id) es un HECHO del sistema: si más adelante otra
                        parte de Reservo necesita saber si Luis tiene una reserva
                        activa, ese dato tiene que estar disponible sin que nadie
                        se lo tenga que volver a preguntar a booking_agent.

  Marta (fan-out)    -> NO. Una comparación de precios y una pregunta general de
                        política son información que Marta consume UNA VEZ, en su
                        respuesta -- no hay ningún "hecho del sistema" que otra
                        parte necesite después. Nadie va a preguntar "¿qué le
                        cotizó pricing_agent a Marta?" en otra corrida.

  Valentina (handoff) -> NO, todavía. Cotizar sin confirmar es información de
                        UNA respuesta, no un hecho persistente -- si Valentina
                        más adelante SÍ confirma esa reserva (como hace en la
                        Demo C completa de la lección 08), ESA parte de su
                        petición sí escribe al Blackboard, con un Track de
                        pipeline distinto.

La regla no es "pipeline siempre escribe, fan-out y handoff nunca" — es más precisa que eso: se escribe lo que produce un hecho confirmado que el resto del sistema podría necesitar después, sin importar qué patrón lo produjo. En esta lección, la única sub-tarea que produce un hecho así es la reserva confirmada de Luis — es una coincidencia de estas tres demos en particular, no una regla fija del patrón. La Demo C de la lección 08 lo confirma: su track de pipeline (la reserva de Focus de Valentina) SÍ escribe, mientras que su track de fan-out y su track de handoff, otra vez, no.

Errores comunes

1. Concluir "fan-out y handoff nunca escriben al Blackboard". No es una regla del patrón — es consecuencia de qué produce cada sub-tarea. Un fan-out que SÍ confirmara una reserva (por ejemplo, dos reservas independientes resueltas a la vez) SÍ debería escribir cada una; esta lección no lo muestra porque ni la pieza de Marta ni la de Valentina confirman nada, no porque el mecanismo se lo impida.

2. Escribir al Blackboard DENTRO de la sección paralela de un fan-out. Aunque esta lección no tiene ningún caso que lo necesite, la regla de M6 L07 sigue firme para cuando sí haga falta: cualquier escritura ocurre DESPUÉS de que el ThreadPoolExecutor cierra, nunca durante, para evitar condiciones de carrera sobre el mismo objeto Blackboard.

3. Pensar que WRITE_SEQ reinicia con cada Blackboard() nuevo. Como ya confirmó M7 L07 Ejercicio 2, WRITE_SEQ es un contador a nivel de proceso, no de instancia — si esta lección creara un segundo Blackboard() después del primero, su primera entrada seguiría desde 7, no desde 1.

Ejercicios

Ejercicio 1: Agrega una cuarta mesa que SÍ escribe, con fan-out (Fácil)

Diseña una cuarta pieza —un socio nuevo, "Pedro"— cuya petición sea un fan-out de DOS reservas independientes (por ejemplo, "Reserva el Focus basic 1h para mí, y aparte reserva el Boardroom basic 1h para mi colega"), y confirma que el Blackboard SÍ recibe escrituras de las dos, aunque corran a la vez.

# run_tracks_parallel: la misma función de M4/M7, reusada sin cambios
# (ya la conectaste en la lección 04 de este módulo).
def run_tracks_parallel(jobs):
    results = {}
    with concurrent.futures.ThreadPoolExecutor(max_workers=len(jobs)) as pool:
        future_to_key = {pool.submit(fn): key for key, fn in jobs.items()}
        for future in concurrent.futures.as_completed(future_to_key):
            key = future_to_key[future]
            results[key] = future.result()
    return results


def job_book_focus_pedro():
    script = [
        {"stop_reason": "tool_use", "content": [
            {"type": "tool_use", "id": "toolu_01", "name": "book_room",
             "input": {"room": "Focus", "tier": "basic", "hours": 1, "member": "Pedro"}}]},
        {"stop_reason": "end_turn", "content": [
            {"type": "text", "text": "Reservé Focus basic 1h para Pedro."}]},
    ]
    return run_specialist("booking_agent", "Reserva Focus basic 1h para Pedro.", script)


def job_book_boardroom_colega():
    script = [
        {"stop_reason": "tool_use", "content": [
            {"type": "tool_use", "id": "toolu_01", "name": "book_room",
             "input": {"room": "Boardroom", "tier": "basic", "hours": 1, "member": "Colega de Pedro"}}]},
        {"stop_reason": "end_turn", "content": [
            {"type": "text", "text": "Reservé Boardroom basic 1h para el colega de Pedro."}]},
    ]
    return run_specialist("booking_agent", "Reserva Boardroom basic 1h para el colega.", script)


results_pedro = run_tracks_parallel({"focus": job_book_focus_pedro, "boardroom": job_book_boardroom_colega})
bb.write("supervisor", member="Pedro")
for key in sorted(results_pedro):
    final, history = results_pedro[key]
    tool_result = next(
        b["content"] for m in history if isinstance(m["content"], list)
        for b in m["content"] if b["type"] == "tool_result"
    )
    print(f"  [{key}] tool_result: {tool_result}")

Ejercicio 2: ¿Por qué supervisor escribe member ANTES de correr ningún especialista? (Medio)

En las tres piezas de esta lección, bb.write("supervisor", member=...) ocurre antes de correr booking_agent, policy_agent o pricing_agent. Sin ejecutar código, explica por qué este orden importa, usando el razonamiento de M6 L07 sobre la petición de Ana.

Ejercicio 3: Diseña una petición donde el track de HANDOFF sí debería escribir (Difícil)

Esta lección mostró que el handoff de Valentina no escribe porque termina en una cotización, no en una confirmación. Diseña, en una frase, una petición de Reservo donde un handoff SÍ terminara produciendo un hecho digno de escribirse al Blackboard — y explica exactamente qué campo(s) escribirías y por qué.

Resumen y siguiente paso

• Blackboard, sin ningún cambio desde el Módulo 6, sigue registrando solo lo que otra parte del sistema podría necesitar después — confirmado sobre tres socios nuevos en una sola corrida.
• La regla no es "el patrón decide si se escribe" — es "el resultado decide": la reserva confirmada de Luis escribe seis entradas; la comparación de Marta y la cotización sin confirmar de Valentina, ninguna.
• WRITE_SEQ se mantuvo en 6 durante toda la pieza de Marta y de Valentina — el testigo en código de que ninguna de las dos escribió nada.

Siguiente lección: 07 — Midiendo el costo de coordinación del sistema completo. La única pieza genuinamente nueva de este módulo: una función que cuenta llamadas, hops y rondas para cualquier combinación de tracks — probada contra las Demos A y B.

Recursos adicionales

1. Anthropic — Multi-agent research system — Un sistema real donde solo los hechos que otro componente necesita después se persisten en un estado compartido, no cada paso intermedio.
2. Python — Dataclasses — La estructura detrás de Blackboard y WriteLogEntry, sin cambios desde el Módulo 6.
3. Python — itertools.count — El contador determinista detrás de WRITE_SEQ y de _booking_ids, la razón por la que este módulo nunca necesita random ni uuid4.
4. Anthropic — Building effective agents — El principio de mantener el estado compartido mínimo y explícito, la misma disciplina que esta lección confirma sobre tres casos nuevos.