Conectando el fan-out al sistema

Descripción

La Demo A (lección anterior) resolvió una petición con una sola cadena de dependencias — nunca hubo nada que correr a la vez. Esta lección conecta el segundo mecanismo del sistema, run_tracks_parallel (Módulos 4 y 7), sin ningún cambio, y lo prueba con la segunda demo del capstone: Marta, un socio con dos preguntas genuinamente independientes, ninguna reserva de por medio.

Al final de esta lección vas a tener la Demo B completa: pricing_agent comparando dos salas y policy_agent respondiendo una pregunta general de no-presentación, corriendo a la vez, con un ThreadPoolExecutor real.

Conexión con el módulo

run_tracks_parallel es la misma función que M7 generalizó a partir del run_fanout_parallel del Módulo 4 — sigue despachando cualquier callable, sin conocer de antemano qué patrón resuelve por dentro. Esta lección no le agrega nada nuevo: confirma que sigue funcionando sobre el registro de especialistas de la lección 02, con una petición que —a diferencia de la de Luis— sí tiene dos sub-tareas que pueden resolverse al mismo tiempo. La lección 07 mide su costo de coordinación —incluyendo el ahorro de rondas que el fan-out siempre entrega—; la lección 08 la retoma como parte de la corrida completa.

Analogía: dos mozos, dos preguntas, ningún motivo para esperar

A diferencia de la mesa de Luis, la mesa de Marta no tiene un plato con pasos obligatorios en orden — tiene dos preguntas sueltas: "¿cuánto sale una mesa más grande?" y "¿qué pasa si alguien falta?". Ningún mozo necesita esperar la respuesta del otro para empezar a trabajar la suya. Es exactamente la escena de dos mozos, cada uno resolviendo su pregunta a la vez, que M4 usó para introducir el fan-out.

Ejemplo trabajado: Demo B — Marta compara salas y pregunta por no-presentación

import concurrent.futures
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 "no" in q and ("present" in q or "show" in q or "llega" in q):
        return ("[no-show-policy] 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.")
    return "No se encontró una política relevante para esa pregunta."


SPECIALISTS = {
    "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)


# ---- M4/M7: run_tracks_parallel, sin cambios ----
def run_tracks_parallel(jobs):
    """jobs: dict {track_key: callable sin argumentos}. ThreadPoolExecutor
    real -- los callables corren A LA VEZ, en hilos distintos."""
    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


# ---- Demo B: Marta, SOLO fan-out ----
print("--- Marta: compara Studio y Boardroom pro 2h, y aparte pregunta por no-presentación ---")

script_pricing = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "get_quote",
         "input": {"room": "Studio", "tier": "pro", "hours": 2}},
        {"type": "tool_use", "id": "toolu_02", "name": "get_quote",
         "input": {"room": "Boardroom", "tier": "pro", "hours": 2}},
    ]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": (
            "Studio pro 2h: 6400 centavos. Boardroom pro 2h: 12800 centavos. "
            "Studio es la opción más barata de las dos."
        )}]},
]
script_no_show = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "search_docs",
         "input": {"query": "qué pasa si un socio no se presenta a una 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."
        )}]},
]


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


def job_no_show_policy():
    return run_specialist(
        "policy_agent", "¿Qué pasa si un socio no se presenta a una reserva confirmada?", script_no_show,
    )


jobs = {"compare_rooms": job_compare_rooms, "no_show_policy": job_no_show_policy}
results = run_tracks_parallel(jobs)

print("claves según llegaron (informativo, puede variar entre corridas):", list(results.keys()))
print()
for key in sorted(results):
    final, _ = results[key]
    print(f"  [{key}] {final['content'][0]['text']}")

Qué esperar (la primera línea, el orden de llegada, puede variar entre corridas — el resto, no):

--- Marta: compara Studio y Boardroom pro 2h, y aparte pregunta por no-presentación ---
claves según llegaron (informativo, puede variar entre corridas): ['no_show_policy', 'compare_rooms']

  [compare_rooms] Studio pro 2h: 6400 centavos. Boardroom pro 2h: 12800 centavos. Studio es la opción más barata de las dos.
  [no_show_policy] 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.

Verifica los dos números a mano: Studio pro 2h = 4000 * 2 * 80 // 100 = 6400 centavos, Boardroom pro 2h = 8000 * 2 * 80 // 100 = 12800 centavos — los dos salen de get_quote ejecutado dentro de pricing_agent, en la misma llamada (dos bloques tool_use en un solo turno assistant, la variante de "fan-out dentro de un agente" que ya viste en M4 L05). Como establece la regla dura de esta guía, el orden de finalización de los dos tracks (la línea "claves según llegaron") puede cambiar entre corridas — pero la salida que sí importa, impresa con sorted(), nunca depende de ese orden.

Por qué esta petición es "solo fan-out" y no otra cosa

1. ¿"comparar Studio y Boardroom" depende de otra sub-tarea de la petición?
   NO -- no necesita nada de la pregunta de no-presentación, ni viceversa.
   Sigue a la pregunta 2.

2. ¿Hay otra sub-tarea, también independiente, que podría resolverse a la vez?
   SÍ -- la pregunta de no-presentación. -> FAN-OUT.

3. ¿Algún dato hace falta después, para otra parte del sistema?
   NO -- ninguna de las dos sub-tareas reserva nada. Ninguna escribe al
   Blackboard (la lección 06 lo confirma ejecutando).

La diferencia clave con la petición de Luis: acá SÍ hay una segunda sub-tarea independiente, así que la pregunta 2 del criterio dispara fan-out en vez de "un solo run_specialist alcanza". Y a diferencia de lo que vas a ver en la lección 05, ningún agente empieza a trabajar una tarea y descubre a mitad de camino que necesita a otro — las dos preguntas de Marta ya nacen, desde el principio, en el dominio del especialista correcto (pricing_agent y policy_agent, respectivamente).

Errores comunes

1. Buscar una dependencia entre las dos sub-tareas de Marta que no existe. Es tentador pensar que "si se necesita más espacio" (la comparación) podría depender de la política de no-presentación, o viceversa — pero la petición, leída con cuidado, nunca las conecta. Son dos preguntas sueltas que comparten la misma frase, no la misma dependencia.

2. Asumir que fan-out siempre reduce las llamadas al modelo. Como ya midió M4 L07, y como confirma la lección 07 de este módulo, el fan-out ahorra rondas de coordinación, no llamadas ni hops — el trabajo real de cada especialista (pricing_agent, policy_agent) es exactamente el mismo, corra a la vez o uno después del otro.

3. Confundir la línea "claves según llegaron" con el resultado real. Esa línea es puramente informativa —confirma que el ThreadPoolExecutor corrió de verdad los dos jobs a la vez— y nunca se usa para decidir en qué orden imprimir nada. El bucle for key in sorted(results) es lo único que determina el orden real de la salida.

Ejercicios

Ejercicio 1: Agrega un tercer track independiente a la Demo B (Fácil)

Marta también pregunta: "¿qué salas tiene disponibles Reservo en total?" — una tercera sub-tarea, resuelta con list_rooms() sobre booking_agent, también independiente de las otras dos. Agrégala a jobs y ejecuta el fan-out de tres tracks.

SPECIALISTS["booking_agent"] = {"tools": {"list_rooms": rt.list_rooms}}

script_list_rooms = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "list_rooms", "input": {}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Reservo tiene 3 salas: Focus, Studio y Boardroom."}]},
]


def job_list_rooms():
    return run_specialist("booking_agent", "¿Qué salas tiene Reservo?", script_list_rooms)


jobs_3 = {"compare_rooms": job_compare_rooms, "no_show_policy": job_no_show_policy, "list_rooms": job_list_rooms}
results_3 = run_tracks_parallel(jobs_3)
for key in sorted(results_3):
    final, _ = results_3[key]
    print(f"  [{key}] {final['content'][0]['text']}")

  [compare_rooms] Studio pro 2h: 6400 centavos. Boardroom pro 2h: 12800 centavos. Studio es la opción más barata de las dos.
  [list_rooms] Reservo tiene 3 salas: Focus, Studio y Boardroom.
  [no_show_policy] 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.

Ejercicio 2: ¿Por qué compare_rooms es UN solo track, y no dos? (Medio)

compare_rooms compara Studio y Boardroom con dos llamadas a get_quote, pero está representado como un solo job en jobs, no como dos tracks separados. Sin ejecutar código, explica por qué esta lección lo modela así, en vez de darle a cada sala su propio track.

Ejercicio 3: Mide, sin ejecutar coordination_cost todavía, si esta demo ahorra hops respecto a hacerla secuencial (Difícil)

Antes de llegar a la lección 07, razona: si Marta pidiera sus dos preguntas una DESPUÉS de la otra (sin ThreadPoolExecutor, un run_specialist tras otro), ¿cuántos hops entre agentes cambiarían respecto a la versión paralela de esta lección? Justifica con la convención de M4 L07 (HOPS_PER_SPECIALIST = 2, ida y vuelta a cada especialista).

Resumen y siguiente paso

• run_tracks_parallel, sin ningún cambio desde M4/M7, resolvió la segunda demo del capstone: Marta, comparando dos salas y preguntando por la política de no-presentación, con los dos tracks corriendo a la vez.
• A diferencia de la Demo A, esta petición SÍ tiene una segunda sub-tarea independiente —por eso dispara fan-out, no pipeline.
• Studio pro 2h = 6400, Boardroom pro 2h = 12800 — verificados a mano y confirmados en la salida real.

Siguiente lección: 05 — Conectando un handoff al sistema. La pieza que la Demo C va a necesitar: un agente que empieza cotizando y, a mitad de camino, cede el turno a otro especialista.

Recursos adicionales

1. Anthropic — Multi-agent research system — Sub-agentes que exploran preguntas independientes a la vez, la misma forma que run_tracks_parallel implementa a mano en esta lección.
2. Python — concurrent.futures.ThreadPoolExecutor — El mecanismo real detrás del fan-out de esta lección, sin cambios desde M4.
3. Python — concurrent.futures.as_completed — La función que produce el orden de llegada informativo (y no determinista) de esta lección.
4. Anthropic — Building effective agents — El patrón de "parallelization" como una de las formas base de composición de agentes, la idea detrás de esta lección.