Midiendo el costo de coordinación de un pipeline

Descripción

La lección 04 ejecutó el pipeline completo de Reservo y dejó una observación suelta: en ningún momento de las tres etapas hizo falta decidir "¿cuál sigue?". Esta lección convierte esa observación en un número. Vas a tomar la MISMA tarea de tres etapas —cotizar, validar la política de cancelación, confirmar— y calcular cuánto costaría resolverla si, en vez de un pipeline con el orden fijo, un supervisor tipo Módulo 2 tuviera que decidir, desde cero, a quién le toca después de cada resultado — exactamente lo que un supervisor haría si no tuviera ningún concepto de "orden fijo" incorporado.

El resultado no es una opinión: es un conteo. Vas a ver que el pipeline de este módulo ahorra exactamente una llamada de ruteo por etapa —tres llamadas al modelo menos, para esta tarea de tres pasos— y cuatro hops menos, sin perder ni un tool call ni cambiar la respuesta final. Esta es la medición que sostiene la frase de la lección 01: "menos llamadas de coordinación, menos puntos de decisión que pueden salir mal".

Conexión con el módulo

Esta lección reusa, sin cambios, el pipeline ejecutado en la lección 04 —mismo PIPELINE_STAGES, mismos guiones, mismo run_pipeline—. Lo único nuevo es el modelo de costo que compara ese resultado real contra un supervisor hipotético resolviendo la misma secuencia. La lección 07 retoma esta misma comparación desde otro ángulo: no cuánto cuesta el orden fijo, sino si ese orden fijo era, de entrada, la elección correcta para cada etapa.

Analogía: pedir indicaciones en cada esquina, contra seguir un mapa ya trazado

Imagina llegar a una ciudad nueva con dos formas de moverte entre tres lugares que sabes que vas a visitar, siempre en el mismo orden. La primera: en cada esquina, preguntarle a alguien "¿por dónde sigo?" — aunque el destino final sea siempre el mismo, cada pregunta cuesta tiempo, y cada respuesta es una oportunidad más de que te indiquen mal. La segunda: seguir un mapa que ya trazaste antes de salir, con la ruta completa marcada — nunca preguntas nada, porque ya sabías, desde el principio, que ibas a pasar por esos tres lugares en ese orden. El pipeline de este módulo es el mapa ya trazado; un supervisor resolviendo la misma secuencia, paso a paso, es preguntar en cada esquina.

Ejemplo trabajado: el mismo pipeline, con el costo contado

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


POLICY_DOCS = {
    "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']}"
    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)


@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":
        return (f"Reserva {payload['room']} {payload['tier']} {payload['hours']}h "
                 f"para {payload['member']} -- la política de cancelación ya se validó.")
    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["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


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

model_scripts = [model_script_quote, model_script_policy, model_script_confirm]
payload, trace = run_pipeline(PIPELINE_STAGES, model_scripts, INITIAL_PAYLOAD)

specialist_calls = sum(1 for s in trace for m in s["history"] if m["role"] == "assistant")
specialist_tools = sum(
    1 for s in trace for m in s["history"] if isinstance(m["content"], list)
    for b in m["content"] if b["type"] == "tool_use"
)
N = len(PIPELINE_STAGES)
COMPOSE_CALLS = 1  # concepto: la síntesis final para el socio (lección 04)

# --- Pipeline (este módulo): el orden está fijo, cero llamadas de ruteo ---
pipeline_route_calls = 0
pipeline_total = pipeline_route_calls + specialist_calls + COMPOSE_CALLS
pipeline_hops = N - 1  # traspaso DIRECTO de una etapa a la siguiente

# --- Hipotético: un supervisor tipo Módulo 2 resolviendo la MISMA secuencia
# de 3 etapas, decidiendo desde cero a quién le toca después de cada
# resultado -- porque, a diferencia del pipeline, no tiene ningún orden
# incorporado de antemano ---
supervisor_route_calls = N       # una decisión de ruteo antes de CADA etapa
supervisor_total = supervisor_route_calls + specialist_calls + COMPOSE_CALLS
supervisor_hops = 2 * N          # ida y vuelta a cada especialista (convención M2 L06)

print(f"{'':36}{'Pipeline (M3)':>16}{'Supervisor repetido':>22}")
print(f"{'llamadas al modelo (ruteo)':36}{pipeline_route_calls:>16}{supervisor_route_calls:>22}")
print(f"{'llamadas al modelo (especialistas)':36}{specialist_calls:>16}{specialist_calls:>22}")
print(f"{'llamadas al modelo (síntesis)':36}{COMPOSE_CALLS:>16}{COMPOSE_CALLS:>22}")
print(f"{'llamadas al modelo TOTAL':36}{pipeline_total:>16}{supervisor_total:>22}")
print(f"{'hops entre agentes':36}{pipeline_hops:>16}{supervisor_hops:>22}")

extra_calls = supervisor_total - pipeline_total
extra_hops = supervisor_hops - pipeline_hops
print()
print(f"diferencia: el supervisor repetido usa {extra_calls} llamadas al modelo MÁS "
      f"({extra_calls}/{N} = exactamente 1 por etapa) y {extra_hops} hops MÁS que el "
      f"pipeline -- para la MISMA tarea de {N} etapas, con las MISMAS {specialist_calls} "
      f"llamadas internas y las MISMAS {specialist_tools} tool calls.")

Qué esperar:

                                       Pipeline (M3)  Supervisor repetido
llamadas al modelo (ruteo)                        0                    3
llamadas al modelo (especialistas)                6                    6
llamadas al modelo (síntesis)                     1                    1
llamadas al modelo TOTAL                          7                   10
hops entre agentes                                2                    6

diferencia: el supervisor repetido usa 3 llamadas al modelo MÁS (3/3 = exactamente 1 por etapa) y 4 hops MÁS que el pipeline -- para la MISMA tarea de 3 etapas, con las MISMAS 6 llamadas internas y las MISMAS 3 tool calls.

El pipeline resuelve la misma tarea con 7 llamadas al modelo; un supervisor que tuviera que decidir en cada una de las tres etapas gastaría 10 — un 30% más, solo en coordinación, sin que ninguna de las dos rutas haga más trabajo real: las mismas 6 llamadas internas de los especialistas, las mismas 3 tool calls, la misma información final disponible.

De dónde sale cada número

Vale la pena desarmar el conteo, porque cada pieza tiene una razón concreta:

Pipeline:
  0 llamadas de ruteo  -- PIPELINE_STAGES ya trae el orden fijo; run_pipeline
                           nunca evalúa "¿cuál sigue?"
  6 llamadas internas  -- 2 por etapa (un tool_use, un texto final) x 3 etapas
  1 síntesis           -- concepto, la respuesta final para el socio (lección 04)
  ------------------------------------------------------------------
  7 TOTAL

Supervisor repetido (hipotético, NO construido en este módulo):
  3 llamadas de ruteo  -- una decisión antes de CADA etapa (N = 3), porque un
                           supervisor sin orden fijo tiene que volver a decidir
                           "¿a quién le toca ahora?" después de cada resultado
  6 llamadas internas  -- IDÉNTICO al pipeline -- el trabajo real no cambia
  1 síntesis           -- IDÉNTICO al pipeline
  ------------------------------------------------------------------
  10 TOTAL

Los hops siguen la misma lógica. En el pipeline, cada etapa le entrega su payload directamente a la siguiente — N - 1 = 2 traspasos para 3 etapas, sin que ninguno vuelva a pasar por un coordinador central. En el supervisor repetido, cada especialista se consulta con un viaje de ida y vuelta —la misma convención que ya usó el Módulo 2, lección 06: 2 hops por especialista consultado—, así que 3 especialistas consultados de forma independiente cuestan 2 * 3 = 6 hops.

Por qué el ahorro es "exactamente 1 por etapa"

No es una coincidencia de esta tarea puntual — es una propiedad general del patrón. Un supervisor que no sabe que el orden es fijo tiene que gastar una decisión (concepto) en cada transición, sin importar cuántas etapas tenga la secuencia: con 3 etapas, 3 decisiones; con 5, serían 5. Un pipeline, en cambio, siempre gasta cero decisiones, sin importar cuán larga sea la secuencia, porque el orden completo ya estaba escrito en PIPELINE_STAGES antes de la primera petición. El ahorro crece exactamente al mismo ritmo que crece el pipeline — una razón concreta, medible, para preferir este patrón cuando la secuencia de pasos de verdad no cambia nunca.

Errores comunes

1. Pensar que este módulo construyó realmente un "supervisor repetido". No — es un modelo de costo, no un sistema ejecutado. El Módulo 2 nunca construyó un supervisor que re-decidiera en cada etapa de una secuencia fija; esta lección solo calcula cuánto costaría si alguien lo hiciera, para poder comparar contra el pipeline real de la lección 04.

2. Sumar mal el costo total. pipeline_route_calls + specialist_calls + COMPOSE_CALLS no es lo mismo que specialist_tools — mezclar "llamadas al modelo" con "tool calls" lleva a comparaciones erróneas, el mismo error ya advertido en el Módulo 2, lección 06.

3. Generalizar el 30% de esta tarea a cualquier pipeline. El porcentaje exacto depende de cuántas llamadas internas tiene cada etapa —acá, 2 por etapa—. Lo que sí generaliza es el patrón: el ahorro en llamadas de ruteo es siempre N (una por etapa), sin importar cuánto trabajo interno tenga cada una.

4. Concluir que un pipeline siempre es "mejor" que un supervisor. Esta comparación mide el costo de coordinar la MISMA secuencia fija de dos formas distintas — no compara pipeline contra supervisor en general. Cuando el orden de los pasos SÍ puede variar según la petición (el caso central del Módulo 2), un pipeline no aplica en absoluto: no hay ningún "orden fijo" que hardcodear.

5. Olvidar que los hops del pipeline son traspasos directos, no viajes de ida y vuelta. La fórmula N - 1 asume que cada etapa entrega su payload directamente a la siguiente, sin volver a ningún coordinador central en el medio — exactamente cómo run_pipeline está construido desde la lección 03.

Ejercicios

Ejercicio 1: Recalcula el ahorro para un pipeline de 5 etapas (Fácil)

Sin ejecutar código todavía, usa el patrón de esta lección para predecir: si un pipeline tuviera 5 etapas, cada una con 2 llamadas internas, ¿cuántas llamadas de ruteo ahorraría frente a un supervisor repetido? ¿Cuántos hops ahorraría?

def pipeline_savings(n_stages, calls_per_stage=2):
    pipeline_calls = 0 + (calls_per_stage * n_stages) + 1
    supervisor_calls = n_stages + (calls_per_stage * n_stages) + 1
    pipeline_hops = n_stages - 1
    supervisor_hops = 2 * n_stages
    return {
        "llamadas ahorradas": supervisor_calls - pipeline_calls,
        "hops ahorrados": supervisor_hops - pipeline_hops,
    }

print(pipeline_savings(5))

{'llamadas ahorradas': 5, 'hops ahorrados': 6}

Ejercicio 2: Mide el costo de un pipeline de una sola etapa (Medio)

Calcula, con la fórmula del Ejercicio 1, el ahorro de un pipeline de una sola etapa. ¿Tiene sentido el resultado? Relaciónalo con lo que ya sabes del Módulo 1, lección 05.

print(pipeline_savings(1))

{'llamadas ahorradas': 1, 'hops ahorrados': 2}

Ejercicio 3: ¿Cuándo el ahorro deja de importar? (Difícil)

Reservo procesa 500 reservas por día, cada una pasando por este pipeline de 3 etapas. Calcula cuántas llamadas al modelo ahorra el pipeline frente al supervisor repetido, por día. Después, en prosa, argumenta: ¿en qué escenario ese ahorro diario dejaría de ser el criterio decisivo para elegir un patrón sobre otro?

DAILY_BOOKINGS = 500
savings_per_booking = pipeline_savings(3)["llamadas ahorradas"]
print(f"ahorro diario: {DAILY_BOOKINGS * savings_per_booking} llamadas al modelo")

ahorro diario: 1500 llamadas al modelo

Resumen y siguiente paso

• Medimos, con números reales, el costo de coordinar el mismo pipeline de 3 etapas de dos formas: con el orden fijo de este módulo (7 llamadas al modelo, 2 hops) y con un supervisor hipotético que decidiera en cada etapa (10 llamadas, 6 hops).
• El ahorro es exactamente 1 llamada de ruteo por etapa —una propiedad general del patrón, no una coincidencia de esta tarea puntual— y 4 hops menos, sin cambiar ni el trabajo interno de los especialistas ni la respuesta final.
• El ahorro escala linealmente con el número de etapas — más etapas en la secuencia, más llamadas ahorradas, siempre y cuando el orden sea genuinamente fijo.
• Este ahorro nunca justifica forzar un pipeline sobre una tarea que en realidad necesita decidir — el mismo principio de "medir antes de generalizar" de toda esta guía.

Siguiente lección: 06 — Cuándo una etapa falla. Con el costo ya medido, vemos el otro lado de tener cero decisiones en el medio: qué pasa cuando una etapa no tiene ningún resultado bueno para entregarle a la siguiente.

Recursos adicionales

1. Anthropic — Multi-agent research system — El reporte de Anthropic sobre el costo real de coordinar, la misma clase de medición que esta lección ejecuta a mano sobre el pipeline de Reservo.
2. Anthropic — Building effective agents — El principio de usar el mecanismo más simple que resuelva la tarea — un pipeline con orden fijo es, exactamente, ese mecanismo más simple cuando la secuencia nunca cambia.
3. Python — Funciones y argumentos por defecto — La base de pipeline_savings, la función de esta lección que generaliza el conteo a cualquier número de etapas.
4. Python 3.14 — What's New — La versión con la que se ejecutó cada línea de esta medición.