Un agente hace handoff a mitad de una rama del fan-out

Descripción

Las lecciones 03 y 04 resolvieron dos de los tres Track de PLAN. Esta lección agrega el tercero: boardroom_no_show, etiquetado handoff en la lección 02. Por dentro, esta sub-tarea empieza siendo una tarea de booking_agent —cotizar el Boardroom pro 2h— y, a mitad de camino, se topa con una pregunta que vive en policy_agent —qué pasa si alguien no llega—. booking_agent cede el turno directamente, con run_with_handoff (Módulo 5), sin volver a consultar a ningún supervisor.

Lo que esta lección demuestra, ejecutando, es que run_tracks_parallel no necesita saber nada sobre esto. Desde la perspectiva del fan-out que la contiene, job_boardroom_no_show es un callable más, exactamente igual que job_book_focus o job_compare_rooms — que por dentro haga dos llamadas encadenadas en vez de una, o que una de esas llamadas ceda el turno a otro agente, es invisible para el mecanismo que lo despacha.

Conexión con el módulo

Esta lección reusa, sin cambios, HandoffPackage, run_agent_with_handoff, run_specialist_with_handoff y run_with_handoff del Módulo 5, y run_tracks_parallel de la lección 04. No hay ninguna pieza nueva de bajo nivel — la única novedad es que, por primera vez en la guía, los tres mecanismos (pipeline, fan-out, handoff) corren en la misma corrida, al mismo tiempo. La lección 06 le agrega el Blackboard compartido a esta misma corrida de tres tracks.

Analogía: un mensajero que, a mitad de camino, le pasa el mandado a otro

Retoma los dos mensajeros de la lección 04 — pero esta vez son tres, y uno de ellos tiene un mandado con una vuelta inesperada. Sale a cotizar un servicio (su especialidad) y, a mitad de camino, el cliente le hace una pregunta sobre un tema que no es el suyo — no vuelve corriendo a preguntarle a quien organizó los mandados del día; simplemente le pasa esa parte puntual a un colega que sí sabe del tema, con la información justa para que pueda responder, y termina ahí su propia parte. Los otros dos mensajeros, mientras tanto, siguen sus propios mandados sin enterarse de nada de esto — cada uno solo le importa entregar su propio resultado cuando termine.

Ejemplo trabajado: tres tracks a la vez — pipeline, fan-out plano, y un fan-out con handoff adentro

import ast
import concurrent.futures
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})")


POLICY_DOCS = {
    "no-show-policy": (
        "Si un miembro no se presenta a una reserva confirmada y no cancela "
        "con al menos 2 horas de anticipación, Reservo cobra el 50% del "
        "precio cotizado como cargo por no-presentación."
    ),
    "cancellation-policy": (
        "Las reservas se pueden cancelar sin cargo hasta 2 horas antes del "
        "horario reservado. Cancelaciones dentro de esas 2 horas aplican "
        "el cargo de no-presentación."
    ),
}


def search_docs(query):
    q = query.lower()
    if "cancela" in q:
        return f"[cancellation-policy] {POLICY_DOCS['cancellation-policy']}"
    if "no" in q and ("present" in q or "show" in q):
        return f"[no-show-policy] {POLICY_DOCS['no-show-policy']}"
    return "No se encontró una política relevante para esa pregunta."


SPECIALISTS = {
    "booking_agent": {
        "tools": {
            "list_rooms": rt.list_rooms, "get_quote": rt.get_quote,
            "book_room": rt.book_room, "cancel_booking": rt.cancel_booking,
        },
    },
    "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)


@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, "kind": stage.kind, "agent": stage.name,
            "label": stage.label, "task": task, "history": history,
            "output": final["content"][0]["text"],
        })
    return payload, trace


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


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


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": "Focus", "tier": "pro", "hours": 3, "member": "Ana"}
model_script_quote = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "get_quote",
         "input": {"room": "Focus", "tier": "pro", "hours": 3}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Focus pro 3h cuesta 6000 centavos."}]},
]
model_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."
        )}]},
]
model_script_confirm = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "book_room",
         "input": {"room": "Focus", "tier": "pro", "hours": 3, "member": "Ana"}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Reservé Focus pro 3h para Ana (confirmación #1)."}]},
]


def job_book_focus():
    return run_pipeline(
        PIPELINE_STAGES,
        [model_script_quote, model_script_policy, model_script_confirm],
        INITIAL_PAYLOAD,
    )


model_script_pricing = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "get_quote",
         "input": {"room": "Studio", "tier": "pro", "hours": 3}},
        {"type": "tool_use", "id": "toolu_02", "name": "get_quote",
         "input": {"room": "Boardroom", "tier": "pro", "hours": 3}},
    ]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": (
            "Studio pro 3h: 9600 centavos. Boardroom pro 3h: 19200 centavos. "
            "Studio es la opción más barata de las dos."
        )}]},
]


def job_compare_rooms():
    return run_specialist("pricing_agent", "Compara Studio y Boardroom pro 3h.", model_script_pricing)


model_script_boardroom_booking = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "get_quote",
         "input": {"room": "Boardroom", "tier": "pro", "hours": 2}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_02", "name": HANDOFF_TOOL_NAME,
         "input": {
             "receiver": "policy_agent",
             "reason": "la pregunta de no-presentación es política, fuera de mi expertise",
             "task": "¿qué pasa si un miembro no se presenta a una reserva confirmada?",
             "context": {"room": "Boardroom", "tier": "pro", "hours": 2, "price_cents": 12800},
         }}]},
]
model_script_boardroom_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."
        )}]},
]
model_scripts_boardroom = {
    "booking_agent": model_script_boardroom_booking,
    "policy_agent": model_script_boardroom_policy,
}


def job_boardroom_no_show():
    return run_with_handoff(
        "booking_agent", "Cotiza Boardroom pro 2h. ¿Qué pasa si no llego?", model_scripts_boardroom,
    )


print("--- tres tracks a la vez: pipeline, fan-out plano, y un fan-out que ADENTRO hace handoff ---")
jobs = {
    "book_focus": job_book_focus,
    "compare_rooms": job_compare_rooms,
    "boardroom_no_show": job_boardroom_no_show,
}
results = run_tracks_parallel(jobs)
print("claves según llegaron (informativo):", list(results.keys()))

print()
print("=== salida SIEMPRE en el mismo orden (sorted por key del track) ===")
for key in sorted(results):
    print(f"--- {key} ---")
    if key == "book_focus":
        payload, trace = results[key]
        print("  payload final:", payload)
    elif key == "compare_rooms":
        final, history = results[key]
        print("  respuesta:", final["content"][0]["text"])
    elif key == "boardroom_no_show":
        final, trace = results[key]
        print(f"  {trace[0]['agent']} cedió el turno a {trace[0]['package'].receiver} "
              f"(razón: {trace[0]['package'].reason!r})")
        print("  respuesta final:", final["content"][0]["text"])

print()
print("=== lo que run_tracks_parallel VE de cada job (no le importa qué corre adentro) ===")
for key in sorted(jobs):
    kind = {"book_focus": "un run_pipeline de 3 etapas", "compare_rooms": "una sola llamada a run_specialist",
            "boardroom_no_show": "un run_with_handoff de 2 llamadas encadenadas"}[key]
    print(f"  {key:<18} -> por dentro: {kind}")

Qué esperar (corrida real; el orden de llegada de las claves puede variar entre ejecuciones):

--- tres tracks a la vez: pipeline, fan-out plano, y un fan-out que ADENTRO hace handoff ---
claves según llegaron (informativo): ['compare_rooms', 'boardroom_no_show', 'book_focus']

=== salida SIEMPRE en el mismo orden (sorted por key del track) ===
--- boardroom_no_show ---
  booking_agent cedió el turno a policy_agent (razón: 'la pregunta de no-presentación es política, fuera de mi expertise')
  respuesta final: 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.
--- book_focus ---
  payload final: {'room': 'Focus', 'tier': 'pro', 'hours': 3, 'member': 'Ana', 'price_cents': 6000, 'cancellation_policy': '[cancellation-policy] Las reservas se pueden cancelar sin cargo hasta 2 horas antes del horario reservado. Cancelaciones dentro de esas 2 horas aplican el cargo de no-presentación.', 'cleared_to_book': True, 'booking_id': 1, 'confirmed': True}
--- compare_rooms ---
  respuesta: Studio pro 3h: 9600 centavos. Boardroom pro 3h: 19200 centavos. Studio es la opción más barata de las dos.

=== lo que run_tracks_parallel VE de cada job (no le importa qué corre adentro) ===
  boardroom_no_show  -> por dentro: un run_with_handoff de 2 llamadas encadenadas
  book_focus         -> por dentro: un run_pipeline de 3 etapas
  compare_rooms      -> por dentro: una sola llamada a run_specialist

Boardroom pro 2h cuesta 12800 centavos (8000 * 2 * 80 // 100), y el cargo de no-presentación —calculado más adelante, en la lección 07, cuando se compone la respuesta final— sería la mitad de eso. Por ahora, fíjate en algo estructural: job_boardroom_no_show corrió dos llamadas a run_specialist_with_handoff por dentro —una para booking_agent, una para policy_agent— y aun así ocupó un solo hilo del ThreadPoolExecutor. El handoff, desde afuera, se ve exactamente igual que cualquier otro job: entra, corre, devuelve un resultado. Nadie en run_tracks_parallel sabe —ni necesita saber— que dentro de ese hilo hubo una transferencia de control entre dos agentes.

Por qué el handoff no "rompe" la garantía de independencia del fan-out

Podría parecer, a primera vista, que un handoff introduce una dependencia nueva —¿no depende policy_agent de que booking_agent haya cotizado primero, dentro de ese mismo track?— Sí, mucho: policy_agent literalmente no correría si booking_agent no hubiera cedido el turno. Pero esa dependencia vive completamente adentro de job_boardroom_no_show, en el orden secuencial que ya garantiza run_with_handoff (Módulo 5) — el emisor corre, y si cede el turno, el receptor corre después, dentro de la misma llamada a esa función. Lo que el fan-out de esta lección exige no es "sin ninguna dependencia interna" — es "sin dependencia entre tracks": ningún dato que job_book_focus o job_compare_rooms producen entra en job_boardroom_no_show, ni al revés. Esa es la frontera que de verdad importa para que los tres puedan correr a la vez sin pisarse.

Errores comunes

1. Pensar que run_tracks_parallel necesita un caso especial para manejar jobs con handoff. No lo necesita — y si lo tuviera, sería una señal de que el mecanismo de composición está mal diseñado. La generalización de la lección 04 (un callable por sub-tarea) ya cubre este caso sin ningún cambio.

2. Confundir la dependencia INTERNA de un handoff (receptor depende del emisor, dentro del mismo job) con una dependencia ENTRE tracks (que sí rompería el fan-out). Son cosas distintas: la primera es aceptable y ya la maneja run_with_handoff por sí solo; la segunda es exactamente lo que la pregunta 1 del criterio de la lección 02 busca detectar antes de etiquetar algo como fanout.

3. Olvidar que boardroom_no_show usa booking_agent, el mismo agente que book_focus. Que dos tracks usen el mismo nombre de especialista no crea ningún conflicto — cada llamada a run_specialist/run_specialist_with_handoff es independiente, con su propio messages, su propio guion, y ninguna corre "dentro" del estado de la otra.

4. Pensar que el cargo de no-presentación de 12800 // 2 = 6400 centavos ya se calculó en esta lección. Todavía no — este ejemplo trabajado se detiene en el resultado de cada track por separado; la composición que combina el context del handoff con el resto de la corrida es tarea de la lección 07.

Ejercicios

Ejercicio 1: Confirma tu propia ejecución (Fácil)

Ejecuta el ejemplo trabajado de esta lección tú mismo y confirma, línea por línea, que tu salida coincide con el "Qué esperar". Presta especial atención a que boardroom_no_show nunca haya llamado a book_room — solo get_quote y search_docs.

Ejercicio 2: Confirma que boardroom_no_show NO reservó nada (Medio)

Después de correr el ejemplo trabajado, inspecciona reservo_tools.BOOKINGS directamente y confirma que solo contiene la reserva de book_focus —el track de boardroom_no_show solo cotizó, nunca reservó.

print("reservas en BOOKINGS:", rt.BOOKINGS)
print("cantidad total de reservas:", len(rt.BOOKINGS))

reservas en BOOKINGS: {1: {'booking_id': 1, 'room': 'Focus', 'tier': 'pro', 'hours': 3, 'member': 'Ana', 'price_cents': 6000}}
cantidad total de reservas: 1

Ejercicio 3: ¿Qué pasaría si policy_agent también intentara ceder el turno? (Difícil)

Sin ejecutar código todavía, usando lo que ya sabes del Módulo 5, lección 06 ("Guardando contra cadenas de handoffs sin fin"), explica qué pasaría si el guion de policy_agent dentro de job_boardroom_no_show invocara, a su vez, handoff_to_specialist de vuelta a booking_agent —y por qué run_with_handoff, tal como está construida en esta lección, no lo detectaría.

Resumen y siguiente paso

• El Track boardroom_no_show se resolvió con run_with_handoff del Módulo 5, sin ningún cambio de código — booking_agent cotiza, se topa con la pregunta de no-presentación, cede el turno a policy_agent con el contexto mínimo.
• run_tracks_parallel corrió los tres tracks —pipeline, fan-out plano, fan-out-con-handoff-adentro— a la vez, sin ningún cambio respecto a la lección 04: no le importa qué mecanismo corre dentro de cada job.
• La dependencia interna de un handoff (receptor después de emisor) no rompe la independencia entre tracks que exige el fan-out — son dos niveles de dependencia distintos.
• reservo_tools.BOOKINGS confirma, ejecutando, que solo book_focus reservó de verdad — boardroom_no_show cotizó, pero nunca confirmó nada.

Siguiente lección: 06 — Un blackboard, compartido por los tres patrones a la vez. Agregamos un Blackboard a esta misma corrida de tres tracks, y confirmamos, ejecutando, que sus escrituras quedan siempre en el mismo orden, sin importar que los tracks hayan corrido en paralelo.

Recursos adicionales

1. Anthropic — Building effective agents — El principio de un agente que reconoce el límite de su propia expertise y transfiere el control, ahora ejecutado dentro de una rama de un fan-out más grande.
2. Python — concurrent.futures.Future — El objeto que encapsula el resultado de cada job, sin importar cuántas llamadas encadenadas haya hecho por dentro.
3. Anthropic — Multi-agent research system — Un sistema real donde sub-agentes paralelos pueden, cada uno por su cuenta, delegar partes de su propio trabajo sin que el orquestador de nivel superior lo note.
4. Python — dataclasses — El módulo detrás de HandoffPackage, reusado sin cambios desde el Módulo 5.