Creación de Subagentes para Investigación y Delegación Paralelas

Inicia subagentes paralelos para investigar subtareas independientes y reportar a un coordinador.

Resumen

Las tareas de investigación son uno de los casos más claros para un sistema multiagente: una sola pregunta a menudo se descompone en varias subpreguntas independientes, cada una de las cuales necesita su propia exploración antes de que se pueda concluir algo.

En lugar de que un agente investigue al competidor A, luego al competidor B, luego al competidor C en secuencia, puedes iniciar un subagente por competidor, ejecutarlos todos concurrentemente y solo pagar el costo de latencia de la rama más lenta en lugar de la suma de todas ellas.

La forma del "fan-out" (despliegue) debe provenir de la forma de la entrada: un subagente por entidad nombrada, por fuente de datos o por sección de un documento, no un número arbitrario elegido de antemano.

Una vez que las ramas terminan, un paso de coordinador tiene que convertir varios hallazgos independientes, a veces parciales o contradictorios, en un informe coherente, que es un trabajo distinto de la investigación en sí y merece su propio paso.

Esta página cubre cómo decidir el número y la forma de los subagentes a partir de la entrada, cómo hacer el "fan-out" y "fan-in" (despliegue y repliegue) con asyncio.gather, y cómo sintetizar hallazgos divergentes sin simplemente concatenarlos.

Receta

Tarjeta de receta de referencia rápida, lista para copiar y pegar.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def research(topic: str) -> str:
    options = AgentOptions(
        subagents=[
            SubagentConfig(
                name=f"research-{topic}",
                description=f"Investiga {topic} y reporta sus hallazgos.",
                allowed_tools=["web_search"],
            ),
        ],
    )
    result = ""
    async for message in query(
        prompt=f"Usa el subagente research-{topic} para investigar {topic}.",
        options=options,
    ):
        if message.get("type") == "subagent_result":
            result = message["result"]
    return result
 
async def main():
    topics = ["tamaño del mercado", "competidores", "riesgo regulatorio"]
    findings = await asyncio.gather(*(research(t) for t in topics))
    # Pasa los hallazgos a una llamada query() separada del coordinador para sintetizar.
    print(dict(zip(topics, findings)))
 
asyncio.run(main())

Cuándo usar esto:

• La entrada se descompone naturalmente en varias preguntas de investigación, fuentes o entidades independientes: competidores, fuentes de datos, secciones de documentos.
• Quieres que la investigación de cada pieza ocurra concurrentemente en lugar de una tras otra.
• Los hallazgos necesitan ser sintetizados en un solo informe, no solo concatenados en el orden en que terminan.
• Algunas ramas de investigación podrían fallar o regresar de forma parcial, y la salida final aún necesita decir algo coherente.

Ejemplo de Trabajo

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def research_one(topic: str, question: str) -> str:
    """Ejecuta un solo subagente de investigación sobre un tema y devuelve sus hallazgos."""
    options = AgentOptions(
        allowed_tools=["web_search", "web_fetch"],
        subagents=[
            SubagentConfig(
                name=f"research-{topic}",
                description=f"Investiga {topic} y responde: {question}",
                allowed_tools=["web_search", "web_fetch"],
            ),
        ],
    )
    findings = []
    async for message in query(
        prompt=f"Usa el subagente research-{topic} para responder: {question}",
        options=options,
    ):
        if message.get("type") == "subagent_result":
            findings.append(message["result"])
    return "\n".join(findings) or "(no se devolvieron hallazgos)"
 
 
async def research_and_synthesize(topics: list[str], question: str) -> str:
    # Fan out: una invocación de subagente por tema, despachada concurrentemente.
    results = await asyncio.gather(
        *(research_one(topic, question) for topic in topics),
        return_exceptions=True,
    )
 
    # Algunas ramas pueden fallar (tiempo de espera, sin resultados, una fuente incorrecta). Continúa
    # con lo que se recibió en lugar de perder todo el lote por un error.
    findings_by_topic = {}
    for topic, result in zip(topics, results):
        if isinstance(result, Exception):
            findings_by_topic[topic] = f"(investigación fallida: {result})"
        else:
            findings_by_topic[topic] = result
 
    # Fan in: una llamada del coordinador sintetiza los hallazgos, que pueden ser
    # parciales o contradictorios, en un solo informe en lugar de simplemente unirlos
    # en el orden en que se completaron.
    coordinator_prompt = "Sintetiza estos hallazgos de investigación en un informe coherente.\n"
    for topic, findings in findings_by_topic.items():
        coordinator_prompt += f"\n## {topic}\n{findings}\n"
    coordinator_prompt += (
        "\nSeñala cualquier acuerdo, desacuerdo o brecha entre las fuentes. "
        "Si la investigación de un tema falló, dilo explícitamente en lugar de omitirlo."
    )
 
    report_chunks = []
    async for message in query(
        prompt=coordinator_prompt,
        options=AgentOptions(allowed_tools=[]),
    ):
        if message.get("type") == "text":
            report_chunks.append(message["text"])
    return "".join(report_chunks)
 
 
asyncio.run(
    research_and_synthesize(
        topics=["Competidor A", "Competidor B", "Competidor C"],
        question="¿Cuál es su modelo de precios actual y mercado objetivo?",
    )
)

Lo que esto demuestra:

• Generación de un subagente por tema de investigación, aquí uno por competidor, para que cada rama explore de forma independiente y concurrente.
• Uso de asyncio.gather con return_exceptions=True para que una rama de investigación fallida no derribe todo el lote.
• Etiquetado de cada conjunto de hallazgos por su tema de origen antes de la síntesis, para que el coordinador (y el lector) puedan atribuir las afirmaciones a una fuente.
• Una llamada query() separada del coordinador, limitada con allowed_tools=[], cuyo único trabajo es sintetizar texto ya recopilado, no realizar más investigaciones.
• Instrucción explícita al coordinador para que muestre desacuerdos y brechas en lugar de simplemente elegir una versión de la verdad de forma silenciosa.

Análisis Profundo

Cómo Funciona

• Cada rama de investigación es su propia invocación de subagente con su propio contexto aislado, por lo que lo que una rama lee o razona nunca se filtra en los resultados de otra rama.
• asyncio.gather despacha todas las corrutinas de las ramas a la vez y devuelve sus resultados en el mismo orden que la entrada, independientemente de cuál termine primero, que es lo que hace que zip(topics, results) sea seguro.
• return_exceptions=True convierte la excepción de una rama en un valor en la lista de resultados en lugar de propagarla y cancelar las otras ramas que aún se están ejecutando; sin él, un fallo mata a todo el gather.
• El paso del coordinador es una llamada query() simple, no otro subagente, porque su trabajo (leer texto ya recopilado y escribir una síntesis) no necesita herramientas ni más exploración, solo un contexto limpio con todos los hallazgos delante.
• Dado que el coordinador solo recibe resultados de texto finales, no el razonamiento intermedio de cada rama, sintetiza a partir de los mismos resúmenes limpios que vería un revisor humano.

Dimensionamiento de Subagentes de Investigación a partir de la Entrada

El número de subagentes siempre debe provenir del conteo de las unidades independientes en la entrada, competidores, fuentes o secciones, no de elegir un número que parezca razonable. Si la entrada no se descompone en piezas independientes, el "fan-out" agrega sobrecarga sin agregar paralelismo.

Síntesis de Hallazgos Divergentes

• Etiqueta cada bloque de hallazgos con su fuente antes de que llegue al coordinador; un muro de texto sin etiquetar no se puede atribuir ni verificar.
• Pide explícitamente al coordinador que señale acuerdos, desacuerdos y brechas, en lugar de simplemente "resumir", lo que lo invita a elegir silenciosamente una versión cuando las fuentes entran en conflicto.
• Pasa las ramas fallidas o parciales como hallazgos explícitos ("investigación fallida: tiempo de espera agotado") en lugar de descartarlas, para que el informe final refleje lo que se sabe y lo que falta.
• Trata la salida del coordinador como el entregable, no los resultados brutos de las ramas; las ramas son notas de trabajo, la síntesis es el informe.

Notas de Python

# return_exceptions=True es lo que hace que el fallo parcial sea tolerable.
# Sin él, una excepción cancela todas las demás tareas en curso.
results = await asyncio.gather(
    *(research_one(topic, question) for topic in topics),
    return_exceptions=True,
)
 
# zip() empareja cada tema con su resultado en el orden de entrada, que asyncio.gather
# preserva a pesar de que las ramas se completan en orden arbitrario.
for topic, result in zip(topics, results):
    ...

Parámetros y Valores de Retorno

Trampas Comunes

• Despliegue de un subagente por elemento de una lista ilimitada. Una lista de 5 competidores es un buen despliegue; una lista de 500 filas es un incidente de costo. Solución: limita el número de subagentes de investigación concurrentes y agrupa o muestrea más allá de ese límite.
• Llamar a asyncio.gather sin return_exceptions=True. La excepción de una rama cancela todas las demás ramas en curso, por lo que una sola fuente inestable derriba un lote que de otro modo estaría bien. Solución: pasa siempre return_exceptions=True y maneja las excepciones explícitamente al desempaquetar los resultados.
• Omitir el paso del coordinador y devolver los resultados brutos de las ramas. Hallazgos concatenados y sin etiquetar de varios subagentes se leen como ruido, no como un informe, y ocultan desacuerdos entre fuentes. Solución: siempre ejecuta un paso de síntesis sobre los hallazgos recopilados antes de tratar la salida como final.
• Otorgar al coordinador el mismo acceso a herramientas que a los subagentes de investigación. El trabajo del coordinador es leer y escribir texto, no buscar o recuperar, por lo que otorgarle web_search/web_fetch amplía su radio de explosión sin ningún beneficio. Solución: limita el coordinador a lo mínimo que necesita, a menudo allowed_tools=[].
• No etiquetar los hallazgos por fuente antes de la síntesis. Si el prompt del coordinador es solo un bloque de texto fusionado, no puede atribuir una afirmación al tema o fuente de donde provino, lo que hace que el desacuerdo sea invisible. Solución: estructura el prompt del coordinador con una sección claramente etiquetada por tema.
• Asumir que una rama fallida debe simplemente descartarse del informe. Omitir silenciosamente un tema cuya investigación falló hace que el informe final parezca completo cuando no lo es. Solución: pasa los fallos como hallazgos explícitos ("investigación fallida: ...") para que tanto el coordinador como el lector vean la brecha.

Alternativas

Preguntas Frecuentes

Relacionado

• Conceptos Básicos de Orquestación de Agentes - el modelo mental sobre el que se construye este patrón
• Patrones de Orquestador/Trabajador con el Claude Agent SDK - el patrón general de despacho y fusión que este especializa
• Barreras de Protección para Sistemas Multiagente - limitando la profundidad del despliegue, el acceso a herramientas y el gasto de tokens
• Delegación de Trabajo a Subagentes en el Claude Agent SDK - la mecánica subyacente de subagentes utilizada aquí

Versiones de Stack: Escrito contra la línea de modelos Claude actual alrededor de junio de 2026 - Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (el predeterminado), y Claude Haiku 4.5 - y el Claude Agent SDK (última versión). Los nombres de modelos, versiones de SDK y precios cambian rápidamente; verifica los detalles actuales en platform.claude.com/docs antes de confiar en ellos.