El patrón planificador-ejecutor en profundidad: paralelizar con asyncio en Python

En este capítulo vas a implementar el patrón planificador-ejecutor: vamos a lanzar tres agentes especialistas al mismo tiempo, con concurrencia real, y a combinar sus resultados en un veredicto único. Al terminar este capítulo serás capaz de:

1. Entender la forma de diamante del patrón planificador-ejecutor: fan-out en paralelo y fan-in para sintetizar.
2. Entender por qué asyncio.gather es la forma idiomática de paralelizar en Python llamadas de red como las de un LLM.
3. Escribir un fan-out real: varios especialistas, cada uno con su propio prompt, ejecutándose a la vez.
4. Escribir el fan-in: un agente sintetizador que integra las opiniones de los especialistas en un único resultado.
5. Explicar por qué el tiempo total de un fan-out es el del especialista más lento, no la suma de todos.

El patrón planificador-ejecutor

El planificador-ejecutor separa “decidir el plan” de “ejecutarlo”: un conjunto de agentes especialistas trabaja en paralelo (fan-out) y, cuando todos terminan, un agente sintetizador integra sus resultados en uno solo (fan-in). Lo veremos con un ejemplo concreto: evaluar una idea de producto —una app que avisa cuándo regar cada planta de tu casa— desde tres ángulos a la vez. En este capítulo el plan ya viene fijado de antemano en un diccionario de especialistas; que el propio planificador decida dinámicamente a quién lanzar es la extensión natural que verás apuntada al final.

Preparar el proyecto

Si vienes siguiendo los capítulos anteriores, el proyecto ya tiene lo necesario. Si empiezas directamente aquí, sigue primero Preparar el proyecto: herramientas y entorno y luego instala lo que usa este capítulo:

uv add openai

La diferencia con los capítulos anteriores es que aquí usamos el cliente asíncrono del SDK, AsyncOpenAI, en vez de OpenAI. Es la misma API, con async/await delante de cada llamada.

Por qué asyncio (fan-out real, no secuencial)

Si escribieras las tres llamadas a los especialistas una detrás de otra, tardarías la suma de las tres. Pero una llamada a un LLM es una operación de red: mientras esperas la respuesta, tu programa no está haciendo ningún cálculo, solo está esperando. Eso es exactamente lo que asyncio está diseñado para aprovechar.

Los especialistas: un rol es solo un prompt distinto

Cada especialista del comité es la misma función, invocada con instrucciones distintas. No hace falta ni una línea de código diferente por especialista: solo cambia el texto del instructions:

import os

from openai import AsyncOpenAI

DEFAULT_MODEL = os.environ.get("OPENAI_MODEL", "gpt-5.1")

EQUIPO = {
    "viabilidad": (
        "Eres ingeniero de software senior. Di qué haría falta para "
        "construir esta idea y cuál sería la parte más difícil. Sé breve."
    ),
    "mercado": (
        "Eres analista de mercado. Di quién compraría esto, qué "
        "competencia existe y cómo destacar. Sé breve."
    ),
    "experiencia": (
        "Eres diseñador de experiencia de usuario. Di cómo debería "
        "sentirse usar esto en el día a día. Sé breve."
    ),
}


async def consultar_especialista(
    client: AsyncOpenAI, rol: str, instrucciones: str, idea: str
) -> tuple[str, str]:
    respuesta = await client.responses.create(
        model=DEFAULT_MODEL,
        instructions=instrucciones,
        input=f"Idea de producto: {idea}",
    )
    print(f"   ✔ {rol} ha terminado")
    return rol, respuesta.output_text

La función devuelve una tupla (rol, texto) en vez de solo el texto: así, cuando lancemos varias a la vez, sabremos qué opinión viene de qué especialista sin depender del orden en el que terminen.

Fan-out: lanzar los tres workers en paralelo

asyncio.gather recibe varias coroutines y las lanza todas a la vez, esperando a que terminen todas antes de continuar:

import asyncio


async def fan_out(client: AsyncOpenAI, idea: str) -> dict[str, str]:
    print("Fan-out: los 3 especialistas trabajan en paralelo…")
    pares = await asyncio.gather(
        *[
            consultar_especialista(client, rol, instrucciones, idea)
            for rol, instrucciones in EQUIPO.items()
        ]
    )
    return dict(pares)

El detalle importante está en el tiempo: si cada especialista tarda alrededor de 3 segundos, las tres llamadas lanzadas con asyncio.gather tardan en conjunto unos 3 segundos, no 9. El tiempo total de un fan-out lo marca el especialista más lento, no la suma de todos: eso es lo que compras al paralelizar de verdad en vez de llamar a cada uno por turnos.

Fan-in: el sintetizador integra las opiniones

Cuando los tres especialistas han terminado, un último agente —el sintetizador— recibe las tres opiniones y las integra en un único veredicto:

async def fan_in(client: AsyncOpenAI, opiniones: dict[str, str]) -> str:
    print("Fan-in: el orquestador junta las tres opiniones…")
    contexto = "\n\n".join(
        f"[{rol.upper()}]\n{texto}" for rol, texto in opiniones.items()
    )
    respuesta = await client.responses.create(
        model=DEFAULT_MODEL,
        instructions=(
            "Eres el orquestador del comité. Integra las tres opiniones "
            "en un veredicto claro: ¿merece la pena intentarlo y por qué? "
            "Menciona los desacuerdos si los hay."
        ),
        input=contexto,
    )
    return respuesta.output_text

El comité completo

Con el fan-out y el fan-in ya escritos, el patrón completo es solo encadenarlos:

async def evaluar_idea(idea: str, client: AsyncOpenAI | None = None) -> dict:
    client = client or AsyncOpenAI()
    opiniones = await fan_out(client, idea)
    veredicto = await fan_in(client, opiniones)
    return {"opiniones": opiniones, "veredicto": veredicto}

Ejecutarlo

async def main() -> None:
    resultado = await evaluar_idea(
        "una app que avisa cuándo regar cada planta de tu casa"
    )
    print("\nVeredicto del comité")
    print(resultado["veredicto"])


if __name__ == "__main__":
    asyncio.run(main())

Ejecútalo con uv run main.py: verás los tres ”✔” de los especialistas aparecer casi a la vez —no uno detrás de otro— y, justo después, el veredicto final que los combina.

Qué te llevas de este ejemplo

• La concurrencia es real, no solo conceptual. asyncio.gather lanza las tres llamadas a la vez; el tiempo total es el del especialista más lento, no la suma de los tres.
• Cada especialista sigue siendo una función aislada. consultar_especialista no sabe nada de los otros especialistas ni del sintetizador: se puede testear pasándole cualquier rol e instrucciones sueltos.
• Añadir un especialista más no cambia la estructura. Basta con añadir una entrada al diccionario EQUIPO; el fan-out itera sobre lo que haya ahí, sin tocar el resto del código.
• El planificador podría decidir dinámicamente qué especialistas lanzar, en vez de lanzar siempre los mismos tres. Esa decisión es, en esencia, un patrón de enrutado por delante del fan-out: los patrones de este bloque se combinan sin dificultad.

Resumen

• El patrón planificador-ejecutor separa “decidir el plan” de “ejecutarlo”: varios agentes especialistas trabajan en paralelo (fan-out) y un sintetizador integra sus resultados (fan-in).
• Las llamadas a un LLM son I/O-bound: asyncio.gather con un cliente asíncrono (AsyncOpenAI) las paraleliza de verdad, sin bloquear el programa mientras cada una espera su respuesta.
• El tiempo total de un fan-out es el del especialista más lento, no la suma de todos.
• Cada especialista sigue siendo una función aislada y testeable; añadir uno nuevo es solo una entrada más en un diccionario de roles.

En el siguiente capítulo cerramos el bloque de patrones con el último: el evaluador-optimizador, un bucle de crítica y reescritura hasta que el resultado cumple una rúbrica.

Autoevaluación

Antes de continuar, comprueba que los conceptos clave han quedado claros.