La tool de handoff y el loop interrumpido

Descripción

Con el paquete diseñado (lección 03), esta lección construye el mecanismo que lo produce: handoff_to_specialist, una pseudo-tool que el modelo (concepto) puede invocar exactamente igual que cualquier tool de dominio —mismo protocolo tool_use, mismo input_schema—, pero que el runner no despacha como una tool real. En vez de eso, run_agent_with_handoff —la variante de run_agent_parallel que construyes en esta lección— la reconoce, detiene su propio loop ahí mismo, y devuelve un HandoffPackage armado a partir del input de esa tool. El agente no termina su tarea con una respuesta de texto; la interrumpe, a propósito, para cederle el turno a otro.

Conexión con el módulo

Esta lección reusa HandoffPackage de la lección 03 sin cambios, y extiende run_agent_parallel —no lo reemplaza: la variante que construyes acá se comporta exactamente igual que el original en todo lo que no sea la pseudo-tool de handoff—. La lección 05 toma run_agent_with_handoff de aquí, sin modificarlo, y lo usa dos veces seguidas: una para el emisor, una para el receptor, encadenadas por el paquete que produce esta lección.

Analogía: la campanilla que el mesero toca, no un plato que sirve

Cuando el mesero de la lección 01 decide llamar al sommelier, no hace nada parecido a lo que hace el resto del tiempo —tomar un pedido, traer un plato—. Toca una campanilla específica, que significa "necesito que alguien más tome esto desde acá". La cocina no confunde esa campanilla con un pedido de comida: es una señal de un tipo completamente distinto, que interrumpe el flujo normal del servicio para redirigirlo. handoff_to_specialist es exactamente esa campanilla: tiene la misma forma que una tool cualquiera (un nombre, un input_schema), pero el runner la trata de forma distinta a cualquier tool de dominio en cuanto la reconoce.

Ejemplo trabajado: la pseudo-tool y el runner que se detiene

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)
    ]


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."
    ),
}


def search_docs(query):
    q = query.lower()
    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,
        },
        "expertise": "cotizar, reservar y cancelar salas",
    },
    "policy_agent": {
        "tools": {"search_docs": search_docs},
        "expertise": "responder preguntas de política (cancelación, no-presentación)",
    },
    "pricing_agent": {
        "tools": {"get_quote": rt.get_quote},
        "expertise": "comparar el costo de varias salas/tiers en una sola respuesta",
    },
}


HANDOFF_TOOL_NAME = "handoff_to_specialist"

# La forma COMPLETA que tendría esta tool si el modelo la invocara de
# verdad -- igual que FANOUT_TOOL en el Módulo 4, la mostramos aunque la
# decisión en sí sea concepto, para que el contrato quede explícito.
HANDOFF_TOOL = {
    "name": HANDOFF_TOOL_NAME,
    "description": (
        "Transfiere el control de la conversación a otro especialista de Reservo "
        "cuando una parte de la petición actual no vive en tu propia expertise. "
        "Usa esta tool SOLO durante una tarea que ya estás resolviendo -- no es "
        "para el enrutamiento inicial (eso lo hace el supervisor, Módulo 2). Pasa "
        "el contexto MÍNIMO que el receptor necesita para responder, nunca tu "
        "historial completo."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "receiver": {"type": "string",
                         "enum": ["booking_agent", "policy_agent", "pricing_agent"]},
            "reason": {"type": "string"},
            "task": {"type": "string"},
            "context": {"type": "object"},
        },
        "required": ["receiver", "reason", "task"],
    },
}


@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):
    """Variante de run_agent_parallel (agent-fundamentals M4/M5) que
    reconoce UNA tool especial, handoff_to_specialist: si el turno del
    modelo la invoca, el loop se DETIENE ahí mismo -- no la despacha con
    dispatch_parallel, como haría con cualquier tool de dominio -- y
    devuelve un HandoffPackage en vez de seguir iterando. En todo lo
    demás, se comporta exactamente igual que run_agent_parallel."""
    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):
    """Wrapper de tres líneas, igual que run_specialist del Módulo 2 --
    busca el registro de tools del especialista y se lo pasa al runner."""
    tools = SPECIALISTS[name]["tools"]
    return run_agent_with_handoff(task, model_script, tools, self_name=name)


# --- guion (concepto, claude-sonnet-5) de booking_agent ---
COMPOUND_REQUEST = "Cotiza Focus pro 3h. Y otra cosa, ¿qué pasa si no llego a la reserva?"

model_script_booking = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "get_quote",
         "input": {"room": "Focus", "tier": "pro", "hours": 3}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_02", "name": HANDOFF_TOOL_NAME,
         "input": {
             "receiver": "policy_agent",
             "reason": "la segunda pregunta es sobre la política de no-presentación, fuera de mi expertise",
             "task": "¿qué pasa si un miembro no se presenta a una reserva confirmada?",
             "context": {"room": "Focus", "tier": "pro", "hours": 3, "price_cents": 6000},
         }}]},
]

final, history, package = run_specialist_with_handoff("booking_agent", COMPOUND_REQUEST, model_script_booking)

print("--- historial de booking_agent, hasta el handoff ---")
for i, m in enumerate(history):
    role, content = m["role"], m["content"]
    if isinstance(content, str):
        print(f"  [{i}] {role:<9} pregunta: {content!r}")
        continue
    for block in content:
        if block["type"] == "tool_use":
            print(f"  [{i}] {role:<9} tool_use({block['name']}): {block['input']}")
        elif block["type"] == "tool_result":
            print(f"  [{i}] {role:<9} tool_result: {block['content']}")
print()
print("final (None -- el loop se detuvo en el handoff, no llegó a un texto final):", final)
print("package:", package)

Qué esperar:

--- historial de booking_agent, hasta el handoff ---
  [0] user      pregunta: 'Cotiza Focus pro 3h. Y otra cosa, ¿qué pasa si no llego a la reserva?'
  [1] assistant tool_use(get_quote): {'room': 'Focus', 'tier': 'pro', 'hours': 3}
  [2] user      tool_result: {'price_cents': 6000}
  [3] assistant tool_use(handoff_to_specialist): {'receiver': 'policy_agent', 'reason': 'la segunda pregunta es sobre la política de no-presentación, fuera de mi expertise', 'task': '¿qué pasa si un miembro no se presenta a una reserva confirmada?', 'context': {'room': 'Focus', 'tier': 'pro', 'hours': 3, 'price_cents': 6000}}

final (None -- el loop se detuvo en el handoff, no llegó a un texto final): None
package: HandoffPackage(sender='booking_agent', receiver='policy_agent', reason='la segunda pregunta es sobre la política de no-presentación, fuera de mi expertise', task='¿qué pasa si un miembro no se presenta a una reserva confirmada?', context={'room': 'Focus', 'tier': 'pro', 'hours': 3, 'price_cents': 6000})

Fíjate en la forma del historial: booking_agent primero resolvió get_quote con el mecanismo normal —dispatch_parallel, un tool_result real, todo igual que en cualquier lección anterior—. Recién en el turno [3], cuando el modelo (concepto) invoca handoff_to_specialist, run_agent_with_handoff reconoce el nombre y no llama a dispatch_parallel con ese bloque —lo intercepta antes—. El historial nunca contiene un tool_result para el handoff, porque nunca se despachó como una tool: el loop simplemente termina ahí, con un package en la mano en vez de un final con texto.

Por qué handoff_to_specialist no está en ningún SPECIALISTS[...]["tools"]

Nota algo importante en el código: HANDOFF_TOOL_NAME nunca aparece como clave en el diccionario de tools de ningún especialista. Si estuviera ahí, dispatch_parallel la trataría como cualquier otra tool —intentaría llamarla como una función de Python con **block["input"], y fallaría, porque no existe ninguna función handoff_to_specialist() real que reciba receiver, reason, task y context como argumentos y devuelva algo útil—. El diseño correcto es exactamente el que construye esta lección: run_agent_with_handoff reconoce el nombre de la pseudo-tool de forma explícita, antes de intentar despacharla, y la desvía a un camino completamente distinto. La pseudo-tool vive a un nivel distinto que las tools de dominio: no resuelve nada de Reservo, coordina quién va a resolverlo.

Comparando con route_to_agent del Módulo 2 y split_into_subtasks del Módulo 4

Las tres pseudo-tools de la guía comparten la misma forma —tool_use con un input_schema propio—, pero cada una interrumpe el flujo en un momento distinto:

route_to_agent (M2)       -> se invoca ANTES de que exista ningún trabajo:
                              el supervisor decide a quién delegar la petición COMPLETA.

split_into_subtasks (M4)  -> se invoca ANTES de que exista ningún trabajo:
                              reconoce que la petición se separa en sub-tareas
                              INDEPENDIENTES, y las reparte todas de una vez.

handoff_to_specialist (M5) -> se invoca A MITAD de un trabajo YA en curso:
                              booking_agent ya resolvió una parte real (get_quote)
                              antes de reconocer el límite de su propia expertise.

La diferencia no está en la forma de la tool —las tres son tool_use normales, con su propio input_schema—. Está en cuándo aparece en el historial: route_to_agent y split_into_subtasks son siempre el primer turno de la conversación (nadie trabajó todavía); handoff_to_specialist puede aparecer en cualquier turno, después de que el propio agente ya avanzó una parte genuina del trabajo. Esa es, en código, la misma distinción que la lección 01 planteó en prosa.

Errores comunes

1. Agregar handoff_to_specialist al diccionario tools de un especialista. Rompe el mecanismo de raíz: dispatch_parallel intentaría llamarla como una función real y fallaría, porque no hay ninguna función de Python con ese nombre que haga algo útil — el manejo de la pseudo-tool vive dentro de run_agent_with_handoff, antes de llegar a dispatch_parallel, no como una entrada más del registro.

2. Olvidar que final es None cuando hay handoff. Un código que asuma final["content"][0]["text"] sin revisar primero si package is None va a fallar con un TypeError —final literalmente es None en ese caso—. La lección 05 muestra el patrón correcto de revisar package antes de tocar final.

3. Pensar que run_agent_with_handoff reemplaza a run_agent_parallel. No — son dos runners distintos para dos situaciones distintas. Un agente que nunca necesita ceder el turno —como booking_agent resolviendo una tarea que vive completa en su propio dominio— puede seguir usando run_agent_parallel sin ningún cambio, exactamente como en los Módulos 2 y 3.

4. Asumir que el input de handoff_to_specialist siempre tiene context. El input_schema solo exige receiver, reason y task — context no está en required. Por eso run_agent_with_handoff usa inp.get("context", {}) en vez de inp["context"]: un handoff con context vacío (como el del Ejercicio 2 de la lección 03) es completamente válido.

Ejercicios

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

Ejecuta el ejemplo trabajado completo tú mismo y confirma, línea por línea, que tu salida coincide con el "Qué esperar" de arriba. Presta especial atención a que el historial NO contenga ningún tool_result para el turno [3].

Ejercicio 2: Un agente que resuelve todo y nunca cede el turno (Medio)

Ejecuta run_specialist_with_handoff con la tarea "Cotiza Studio pro 2h" (sin ninguna segunda pregunta) y un guion de dos turnos —get_quote, después texto final— sin ningún handoff_to_specialist. Confirma que package es None y que final sí trae el texto completo.

script_no_handoff = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "get_quote",
         "input": {"room": "Studio", "tier": "pro", "hours": 2}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Studio pro 2h cuesta 6400 centavos."}]},
]
final_nh, history_nh, package_nh = run_specialist_with_handoff("booking_agent", "Cotiza Studio pro 2h", script_no_handoff)
print("package:", package_nh)
print("texto final:", final_nh["content"][0]["text"])

package: None
texto final: Studio pro 2h cuesta 6400 centavos.

Ejercicio 3: ¿Qué pasa si el handoff no es el ÚLTIMO bloque del turno? (Difícil)

run_agent_with_handoff lee block = turn["content"][0] — siempre el primer bloque del turno. Construye un turno donde handoff_to_specialist aparece como el segundo elemento de content (con un tool_use de get_quote como primer elemento), y ejecuta ese turno. ¿Qué hace el runner con el bloque de handoff_to_specialist en ese caso? ¿Por qué el diseño de esta lección asume que el handoff, si aparece, es el único bloque del turno?

turn_mixed = {"stop_reason": "tool_use", "content": [
    {"type": "tool_use", "id": "toolu_01", "name": "get_quote",
     "input": {"room": "Focus", "tier": "pro", "hours": 3}},
    {"type": "tool_use", "id": "toolu_02", "name": HANDOFF_TOOL_NAME,
     "input": {"receiver": "policy_agent", "reason": "...", "task": "..."}},
]}
block = turn_mixed["content"][0]
print("bloque leído:", block["name"])

# Y si el runner NO reconoce el handoff en content[0], cae a dispatch_parallel
# con TODOS los bloques del turno -- incluido el de handoff, que no está en
# ningún registro de tools:
try:
    dispatch_parallel(turn_mixed["content"], SPECIALISTS["booking_agent"]["tools"])
except KeyError as e:
    print(f"KeyError capturado en dispatch_parallel: {e!r}")

bloque leído: get_quote
KeyError capturado en dispatch_parallel: KeyError('handoff_to_specialist')

Resumen y siguiente paso

• handoff_to_specialist es una pseudo-tool: misma forma tool_use que cualquier tool real, pero nunca vive en el registro de tools de ningún especialista.
• run_agent_with_handoff es la variante de run_agent_parallel que la reconoce antes de intentar despacharla: si el turno la invoca, el loop se detiene y devuelve un HandoffPackage en vez de seguir iterando.
• Confirmado ejecutando: el historial de booking_agent nunca contiene un tool_result para el turno de handoff — el mecanismo interrumpe el flujo, no lo despacha.
• Comparado con route_to_agent (M2) y split_into_subtasks (M4) — misma forma de tool_use, pero esas dos siempre aparecen ANTES de que exista trabajo; handoff_to_specialist puede aparecer DESPUÉS de que el agente ya avanzó una parte real.

Siguiente lección: 05 — El handoff de Reservo, ejecutado de punta a punta. Encadenamos dos llamadas a run_specialist_with_handoff —emisor y receptor— y componemos la respuesta final, grounded en el contexto que viajó en el paquete.

Recursos adicionales

1. Anthropic — Tool use (function calling) overview — La forma exacta de tool_use/input_schema que handoff_to_specialist respeta, aunque el runner la trate distinto a una tool de dominio.
2. Anthropic — Building effective agents — El patrón de un agente que reconoce el límite de su propio alcance y transfiere el control, en vez de forzar una respuesta fuera de su expertise.
3. Python — dataclasses — El módulo detrás de HandoffPackage, reusado sin cambios desde la lección 03.
4. Python — Diccionarios: dict.get() — El método detrás de inp.get("context", {}), que permite que context sea opcional en el input del handoff.