Conectando el pipeline al sistema

Descripción

El registro de especialistas de la lección anterior es la puerta de entrada; esta lección conecta el primer mecanismo de coordinación encima de esa puerta: run_pipeline (Módulo 3), sin ningún cambio. Y lo hace con la primera de las tres demos del capstone: Luis, un socio con una petición que necesita solo pipeline — sin fan-out, sin handoff, sin ninguna sub-tarea independiente. Es la forma más simple posible de una petición de Reservo, y por eso es el punto de partida correcto para el sistema completo.

Al final de esta lección vas a tener la Demo A completa: la petición de Luis, resuelta de punta a punta con las tres etapas del pipeline (cotizar, validar la política, confirmar), sobre el registro de especialistas de la lección 02.

Conexión con el módulo

run_pipeline, PipelineStage, build_stage_task y extract_payload son exactamente las mismas cuatro piezas que construiste en el Módulo 3 — ni una línea distinta. Lo que agrega esta lección es la petición de Luis como el primer caso real del sistema completo, y la confirmación de que el pipeline sigue funcionando sobre SPECIALISTS/run_specialist tal como quedaron en la lección 02. La lección 07 retoma esta misma Demo A para medir su costo de coordinación; la lección 08 la ejecuta de nuevo como parte de la corrida completa del capstone.

Analogía: la primera mesa de la noche, el pedido más simple

De las tres mesas que este módulo va a atender —Luis, Marta y Valentina—, la de Luis es la más simple: un solo plato, del menú fijo, con una verificación de alergias antes de servirlo. No hay nada que comparar, no hay ningún sommelier al que llamar a mitad de camino. Es exactamente el tipo de pedido que la línea de montaje del Módulo 3 resuelve sin ninguna decisión de más: cortar, verificar, servir, en ese orden, siempre.

Ejemplo trabajado: Demo A — Luis reserva el Studio, validando antes de confirmar

import ast
import concurrent.futures
from dataclasses import dataclass
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] 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.")
    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}},
}


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


# ---- M3: pipeline, sin cambios ----
@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


# ---- Demo A: Luis, SOLO pipeline ----
print("--- Luis: reserva el Studio pro 3h, validando la política antes de confirmar ---")

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

payload, trace = run_pipeline(PIPELINE_STAGES, [script_quote, script_policy, script_confirm], INITIAL_PAYLOAD)

for step in trace:
    print(f"  etapa {step['stage']} ({step['label']}) -> {step['agent']}: {step['output']}")

print()
print("payload final:", payload)

Qué esperar (sobre un Reservo desechable, recién iniciado):

--- Luis: reserva el Studio pro 3h, validando la política antes de confirmar ---
  etapa 1 (cotizar) -> booking_agent: Studio pro 3h cuesta 9600 centavos.
  etapa 2 (validar la política de cancelación) -> policy_agent: Puedes cancelar sin cargo hasta 2 horas antes del horario reservado. No hay ningún impedimento para confirmar.
  etapa 3 (confirmar la reserva) -> booking_agent: Reservé Studio pro 3h para Luis (confirmación #1).

payload final: {'room': 'Studio', 'tier': 'pro', 'hours': 3, 'member': 'Luis', 'price_cents': 9600, '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}

Verifica el número a mano: Studio pro 3h = 4000 * 3 * 80 // 100 = 9600 centavos — la tarifa base de Studio (4000), por 3 horas, con el 20% de descuento del tier pro aplicado como entero (* 80 // 100), exactamente la misma fórmula que ancla toda esta guía desde el Módulo 1. booking_id: 1 porque es la primera reserva real de este proceso — ninguna lección anterior de este módulo llamó a book_room.

Fíjate en algo importante para las demos que siguen: esta petición no tiene ninguna sub-pregunta independiente. Todo lo que Luis pidió vive en una sola cadena de tres etapas, en orden fijo. No hay nada que fan-out pudiera paralelizar acá — es exactamente por eso que la Demo A dispara solo pipeline, sin ningún otro patrón mezclado.

Por qué esta petición es "solo pipeline" y no otra cosa

Aplicando el criterio de las tres preguntas de M7 L02 a la petición completa de Luis:

1. ¿Depende del resultado de otra sub-tarea?
   SÍ -- la confirmación (etapa 3) necesita que la política ya esté
   validada (etapa 2), que a su vez necesita el precio (etapa 1).
   Orden fijo, conocido de antemano -> PIPELINE.

2. (no aplica -- ya hay dependencia, no llega a la pregunta 2)

3. ¿Algún dato hace falta después, para otra parte del sistema?
   SÍ -- el booking_id y el precio final son la reserva real de Luis,
   la lección 06 los escribe al Blackboard.

No hay ninguna segunda sub-tarea en la petición de Luis — por eso el PLAN completo de esta demo tiene un solo Track, y la pregunta 2 del criterio ("¿hay otra sub-tarea independiente?") ni siquiera llega a evaluarse. Esto es exactamente lo que distingue a la Demo A de la Demo B (lección 04), que sí tiene dos sub-tareas independientes desde el principio.

Errores comunes

1. Agregar una decisión de ruteo entre etapas que no existe. El pipeline no evalúa "¿cuál sigue?" en ningún punto — PIPELINE_STAGES ya trae el orden fijo de antemano. Si te encuentras escribiendo un if que decide la próxima etapa según el resultado de la anterior, no es un pipeline — es más cercano a un handoff (Módulo 5) o a un supervisor repetido (la comparación de M3 L05).

2. Confirmar antes de validar la política. El orden de PIPELINE_STAGES —quote, validate_policy, confirm— no es arbitrario: invertir las etapas 2 y 3 confirmaría la reserva de Luis antes de saber si había algún impedimento de cancelación, exactamente el error que M3 L02 advirtió desde el principio del módulo.

3. Ejecutar esta lección después de otra que ya reservó algo en el mismo proceso. Si booking_id no sale 1, el intérprete ya corrió otro ejemplo antes — esta lección, como todas las de esta guía, asume un Reservo desechable, recién iniciado.

Ejercicios

Ejercicio 1: Repite la Demo A con otra sala y otro socio (Fácil)

Repite el pipeline completo de esta lección con "Boardroom basic 2h para Carla" (Boardroom basic 2h = 8000 * 2 = 16000 centavos, sin descuento porque basic no aplica el * 80 // 100).

INITIAL_PAYLOAD_CARLA = {"room": "Boardroom", "tier": "basic", "hours": 2, "member": "Carla"}
script_quote_carla = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "get_quote",
         "input": {"room": "Boardroom", "tier": "basic", "hours": 2}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Boardroom basic 2h cuesta 16000 centavos."}]},
]
script_confirm_carla = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "book_room",
         "input": {"room": "Boardroom", "tier": "basic", "hours": 2, "member": "Carla"}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Reservé Boardroom basic 2h para Carla (confirmación #1)."}]},
]
payload_carla, _ = run_pipeline(
    PIPELINE_STAGES, [script_quote_carla, script_policy, script_confirm_carla], INITIAL_PAYLOAD_CARLA,
)
print("payload final:", payload_carla)

payload final: {'room': 'Boardroom', 'tier': 'basic', 'hours': 2, 'member': 'Carla', 'price_cents': 16000, '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}

Ejercicio 2: ¿Qué pasaría si policy_agent encontrara un impedimento? (Medio)

El payload de esta lección siempre tiene cleared_to_book: True porque el guion concepto de policy_agent siempre encuentra la política sin ningún impedimento. Sin ejecutar código todavía, explica qué tendría que cambiar en build_stage_task (etapa confirm) si, en cambio, la política encontrada SÍ impidiera confirmar (por ejemplo, una reserva ya cancelada dos veces por el mismo socio) — ¿el pipeline tal como está escrito podría manejar ese caso?

Ejercicio 3: Cuenta las llamadas internas de esta pipeline, sin usar todavía coordination_cost (Difícil)

Sin ejecutar la función coordination_cost de la lección 07 (todavía no existe en este punto de la guía), cuenta a mano cuántas llamadas al modelo (turnos assistant) tiene el trace completo de la Demo A, sumando los tres history de sus tres etapas. Después, verifica tu conteo con código.

def count_model_calls(history):
    return sum(1 for m in history if m["role"] == "assistant")

total = sum(count_model_calls(step["history"]) for step in trace)
print("llamadas internas totales:", total)

llamadas internas totales: 6

Resumen y siguiente paso

• run_pipeline, sin ningún cambio desde el Módulo 3, resolvió la primera demo del capstone: Luis, reservando el Studio pro 3h, con la política validada antes de confirmar.
• Es la petición más simple de las tres demos: una sola cadena de tres etapas, sin ninguna sub-pregunta independiente, por eso dispara solo pipeline.
• booking_id: 1, price_cents: 9600 — ambos verificados a mano y confirmados con la salida real de la corrida.

Siguiente lección: 04 — Conectando el fan-out al sistema. La segunda demo del capstone: Marta pregunta dos cosas independientes a la vez — una petición que dispara solo fan-out.

Recursos adicionales

1. Anthropic — Building effective agents — El patrón "prompt chaining" (encadenar pasos fijos), la idea detrás de un pipeline, ahora conectado al sistema completo.
2. Anthropic — Messages API reference — La forma exacta de stop_reason/tool_use que cada etapa del pipeline sigue respetando.
3. Python — Dataclasses — La estructura detrás de PipelineStage, sin cambios desde el Módulo 3.
4. Python — ast.literal_eval — La función que last_tool_result usa para convertir el content de texto de un tool_result de vuelta a un diccionario real de Python.