# El patrón planificador-ejecutor en profundidad: paralelizar con asyncio en Python > Implementa el patrón planificador-ejecutor en Python: fan-out real con asyncio.gather, un fan-in que sintetiza resultados y un ejemplo completo y ejecutable. - Capítulo: 0.3 - Dificultad: básico - Fuente: https://book-ai-engineer.lucenyo.dev/es/lessons/fundamentos/patron-planificador-ejecutor-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. ```mermaid sequenceDiagram participant M as evaluar_idea() participant V as viabilidad participant Mk as mercado participant Ex as experiencia participant Fin as fan_in() M->>M: fan_out(): asyncio.gather(...) par ~3s en paralelo, no en serie M->>V: consultar_especialista(idea) V-->>M: (rol, texto) and M->>Mk: consultar_especialista(idea) Mk-->>M: (rol, texto) and M->>Ex: consultar_especialista(idea) Ex-->>M: (rol, texto) end M->>Fin: fan_in(opiniones) Fin-->>M: veredicto ``` --- ## 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](/es/lessons/preparacion/preparar-el-proyecto/) y luego instala lo que usa este capítulo: ```bash 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. Una tarea es **I/O-bound** (limitada por entrada/salida) cuando la mayor parte del tiempo la pasa esperando una respuesta externa —una llamada de red, una consulta a una base de datos— en vez de usando la CPU. `asyncio.gather` permite lanzar varias tareas I/O-bound a la vez y esperar a que todas terminen, sin bloquear el programa mientras cada una espera su respuesta. --- ## 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`: ```python 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: ```python 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: ```python 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: ```python 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 ```python 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.