La anatomía de una etapa

Descripción

El Módulo 2 identificaba a cada especialista por su nombre: SPECIALISTS["booking_agent"], SPECIALISTS["policy_agent"] — un diccionario, porque el supervisor solo necesitaba saber "a quién" delegar cada petición. Este módulo necesita algo distinto, porque un mismo especialista puede aparecer más de una vez en la misma secuencia: booking_agent cotiza en la etapa 1 del pipeline de Reservo y confirma en la etapa 3, con policy_agent en el medio. Si identificaras cada etapa solo por el nombre del especialista, no habría forma de distinguir "la etapa donde booking_agent cotiza" de "la etapa donde booking_agent confirma" — las dos comparten el mismo nombre.

Esta lección construye PipelineStage, la pieza que resuelve esa ambigüedad: cada etapa se identifica por su rol (kind) —qué resuelve, no solo quién la resuelve—, además del especialista que la corre (name) y una etiqueta legible (label). Con las tres etapas del pipeline de Reservo declaradas en PIPELINE_STAGES, vas a ejecutar la primera —cotizar— y armar, a mano, el payload mínimo que la siguiente etapa necesitaría. Ese "a mano" es intencional: expone exactamente el trabajo repetitivo que la lección 03 generaliza.

Conexión con el módulo

Esta lección es la base formal sobre la que se apoya el resto del módulo: PipelineStage y PIPELINE_STAGES se reusan sin cambios en las lecciones 03 a 08. La lección 03 automatiza el paso manual que cierra esta lección —armar el payload— con build_stage_task y extract_payload; la lección 04 corre las tres etapas completas con ese mecanismo ya generalizado.

Analogía: la ficha de cada estación de la línea de ensamblaje

Retoma la receta de la lección 01, pero llevada a una fábrica: una línea de ensamblaje tiene estaciones fijas —cortar, soldar, pintar—, y cada estación tiene una ficha que dice qué hace, con qué máquina, y qué recibe de la estación anterior. Dos estaciones distintas pueden usar la misma máquina —la misma soldadora podría usarse en la estación 2 para unir una pieza y, más adelante en la línea, en la estación 5 para un ajuste final—, pero siguen siendo estaciones distintas, con fichas distintas, porque hacen trabajos distintos en momentos distintos de la línea. PipelineStage es esa ficha: identifica el trabajo (kind), no solo la máquina (name) que lo hace.

Ejemplo trabajado: PipelineStage y el primer paso ejecutado

from dataclasses import dataclass
import concurrent.futures
import reservo_tools as rt


def dispatch_parallel(tool_use_blocks, tools):
    """El de agent-fundamentals M4/M5, sin cambios."""
    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):
    """El de agent-fundamentals M4/M5, sin cambios."""
    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):
    """STUB de policy_agent, idéntico al del Módulo 2."""
    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."


# El registro de especialistas del Módulo 2, sin cambios.
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)",
    },
}


def run_specialist(name, task, model_script):
    """El de la lección 02 del Módulo 2, sin cambios."""
    tools = SPECIALISTS[name]["tools"]
    return run_agent_parallel(task, model_script, tools)


@dataclass
class PipelineStage:
    """Una etapa fija del pipeline: QUÉ resuelve (kind), a QUÉ especialista
    se despacha (name), y una etiqueta corta para trazar (label). `kind`
    identifica el ROL de la etapa -- no el agente -- porque el MISMO
    especialista puede aparecer en más de una etapa."""
    kind: str
    name: str
    label: str


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

print("--- el pipeline de Reservo: orden fijo, sin ninguna decisión de ruteo ---")
for stage in PIPELINE_STAGES:
    print(f"etapa {stage.kind:16} -> {stage.name:15} ({stage.label})")

Qué esperar:

--- el pipeline de Reservo: orden fijo, sin ninguna decisión de ruteo ---
etapa quote            -> booking_agent   (cotizar)
etapa validate_policy  -> policy_agent    (validar la política de cancelación)
etapa confirm          -> booking_agent   (confirmar la reserva)

Compara esta lista con SPECIALISTS del Módulo 2: ahí, SPECIALISTS["booking_agent"] devolvía una entrada, sin importar cuántas veces el supervisor decidiera delegarle una tarea. Acá, booking_agent aparece en la lista dos veces, una vez como kind="quote" y otra como kind="confirm" — dos etapas distintas, mismo especialista. Esta es la diferencia estructural que justifica PipelineStage en vez de reusar SPECIALISTS tal cual: un pipeline necesita distinguir posiciones en una secuencia, no solo identidades de agentes.

Corriendo la primera etapa, sola

print()
print("--- corremos SOLO la etapa 1, a mano, para ver qué produce ---")

task = "Cotiza Focus pro 3h."
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."}]},
]

stage_1 = PIPELINE_STAGES[0]
final, history = run_specialist(stage_1.name, task, model_script_quote)
print("resultado de la etapa 1:", final["content"][0]["text"])

# A mano, todavía sin ninguna función general: armamos el payload que la
# etapa 2 va a necesitar. Esto es exactamente lo que la lección 03
# automatiza -- acá se hace a mano a propósito, para ver el problema antes
# de resolverlo.
payload = {"room": "Focus", "tier": "pro", "hours": 3, "price_cents": 6000}
print("payload armado a mano:", payload)

Qué esperar:

--- corremos SOLO la etapa 1, a mano, para ver qué produce ---
resultado de la etapa 1: Focus pro 3h cuesta 6000 centavos.
payload armado a mano: {'room': 'Focus', 'tier': 'pro', 'hours': 3, 'price_cents': 6000}

run_specialist(stage_1.name, task, model_script_quote) es exactamente el mismo run_specialist del Módulo 2 —sin ninguna modificación—, corriendo sobre booking_agent con el mismo TOOLS de siempre. Lo único nuevo es que la etapa se identifica ahora con un PipelineStage, no con un string suelto. El payload que sigue es, por ahora, texto armado a mano: sé que price_cents es 6000 porque leí el tool_result del historial y lo copié — un proceso manual, propenso a error, que no escala a un pipeline de tres etapas reales.

Por qué kind y no solo name

Vale la pena detenerse en la decisión de diseño: ¿por qué no alcanzaba con SPECIALISTS del Módulo 2, indexado solo por nombre? Tres razones concretas:

1. Un especialista puede repetirse en el pipeline. booking_agent cotiza y confirma —dos trabajos distintos, mismo agente—. Sin kind, no habría forma de decirle a la etapa 3 "confirma" en vez de "cotiza otra vez".
2. Cada etapa arma su tarea de forma distinta. La etapa 1 necesita construir "Cotiza X Y Zh."; la etapa 3 necesita construir "Reserva X Y Zh para W." — funciones distintas, aunque ambas despachen a booking_agent. La lección 03 usa kind exactamente para elegir cuál construir.
3. Cada etapa extrae un dato distinto del resultado. La etapa 1 necesita price_cents; la etapa 3 necesita booking_id. De nuevo, kind es la clave que decide qué extraer — no name, que solo dice quién corrió.

Errores comunes

1. Confundir PIPELINE_STAGES con SPECIALISTS. Son estructuras distintas para preguntas distintas: SPECIALISTS responde "¿qué tools tiene este agente?"; PIPELINE_STAGES responde "¿en qué orden corren las etapas, y qué rol cumple cada una?". Un pipeline sigue necesitando SPECIALISTS por debajo —run_specialist lo usa sin cambios— pero le agrega la secuencia y el rol encima.

2. Pensar que kind tiene que coincidir con name. No hay ninguna regla que los relacione —de hecho, en este módulo, kind="quote" y kind="confirm" comparten el mismo name="booking_agent", y son etapas completamente distintas.

3. Armar el payload a mano en un pipeline real. El ejemplo de esta lección lo hace a propósito, para exponer el problema — pero copiar valores del tool_result a mano no escala ni es confiable una vez que el pipeline tiene varias etapas. La lección 03 resuelve esto con una función que ancla el payload en el resultado real, sin copiar nada a mano.

4. Olvidar que PIPELINE_STAGES es una LISTA, no un diccionario. El orden de la lista es el orden de ejecución — a diferencia de SPECIALISTS, donde el orden de las claves nunca importó porque el supervisor accedía por nombre, no por posición.

5. Ejecutar este ejemplo en un proceso que ya tenía reservas. Si corriste otro ejemplo de la guía en el mismo intérprete, reservo_tools.BOOKINGS no está vacío — aunque esta lección en particular no llama a book_room, las lecciones siguientes del módulo sí, y asumen un proceso nuevo.

Ejercicios

Ejercicio 1: Identifica el rol correcto para una etapa nueva (Fácil)

Reservo agrega, al final del proceso de reserva, un cuarto paso: booking_agent cancela automáticamente cualquier reserva anterior del mismo socio para el mismo horario, antes de crear la nueva. Escribe el PipelineStage para esa etapa, decidiendo un kind que no colisione con "quote" ni "confirm".

cancel_conflicting_stage = PipelineStage(
    kind="cancel_conflicting",
    name="booking_agent",
    label="cancelar reservas previas en conflicto",
)
print(cancel_conflicting_stage)

PipelineStage(kind='cancel_conflicting', name='booking_agent', label='cancelar reservas previas en conflicto')

Ejercicio 2: Corre la etapa 2 sola, con el mismo patrón manual (Medio)

Repite el patrón de esta lección —correr una etapa sola y armar el payload a mano— pero con la etapa 2 (policy_agent, validar la política de cancelación). Usa la tarea "¿Cuál es la política de cancelación para una reserva de Focus de 3h, antes de confirmarla?".

task_2 = "¿Cuál es la política de cancelación para una reserva de Focus de 3h, antes de confirmarla?"
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."
        )}]},
]

stage_2 = PIPELINE_STAGES[1]
final_2, history_2 = run_specialist(stage_2.name, task_2, model_script_policy)
print("resultado de la etapa 2:", final_2["content"][0]["text"])

payload_2 = dict(payload)
payload_2["cleared_to_book"] = True
print("payload armado a mano:", payload_2)

resultado de la etapa 2: Puedes cancelar sin cargo hasta 2 horas antes del horario reservado. No hay ningún impedimento para confirmar.
payload armado a mano: {'room': 'Focus', 'tier': 'pro', 'hours': 3, 'price_cents': 6000, 'cleared_to_book': True}

Ejercicio 3: ¿Qué pasa si PIPELINE_STAGES tiene un name inválido? (Difícil)

Agrega una cuarta entrada a una copia de PIPELINE_STAGES con name="shipping_agent" —un especialista que no existe en SPECIALISTS— e intenta correrla con run_specialist. ¿Qué excepción se produce, y por qué es preferible a que el pipeline siga corriendo con un especialista inventado?

bad_stage = PipelineStage(kind="notify", name="shipping_agent", label="notificar al socio")
try:
    run_specialist(bad_stage.name, "notifica al socio", [])
except KeyError as e:
    print(f"KeyError capturado: {e!r}")

KeyError capturado: KeyError('shipping_agent')

Resumen y siguiente paso

• Un pipeline identifica cada etapa por su rol (kind), no solo por el especialista que la resuelve (name) — porque un mismo especialista puede repetirse en distintas posiciones de la secuencia.
• PipelineStage agrega kind, name y label; PIPELINE_STAGES es una lista, no un diccionario — el orden de la lista es literalmente el orden de ejecución.
• Corrimos la etapa 1 (cotizar) sola, con run_specialist sin ningún cambio del Módulo 2, y armamos el payload que la etapa 2 necesitaría a mano — un proceso que no escala y que la lección 03 automatiza.
• booking_agent aparece dos veces en PIPELINE_STAGES —cotizar y confirmar—, con policy_agent en el medio: la prueba concreta de por qué el rol, no el nombre, es lo que distingue una etapa.

Siguiente lección: 03 — Encadenando el dato entre etapas. Generalizamos el payload armado a mano de esta lección con build_stage_task y extract_payload, ejecutados sobre un pipeline pequeño de dos etapas.

Recursos adicionales

1. Python — dataclasses — El módulo detrás de PipelineStage, la misma herramienta que ya usaste para RoutingDecision en el Módulo 2.
2. Anthropic — Building effective agents — El patrón "prompt chaining", la base conceptual de una secuencia de etapas con orden fijo.
3. Anthropic — Tool use (function calling) overview — El protocolo que run_specialist sigue respetando, sin cambios, dentro de cada etapa.
4. Python — Listas — La estructura detrás de PIPELINE_STAGES, donde el orden de los elementos es, por primera vez en esta guía, parte del significado de la estructura.