Construindo Subagentes para Pesquisa e Delegação Paralelas

Inicie subagentes paralelos para pesquisar subtarefas independentes e reportar de volta a um coordenador.

Resumo

Tarefas de pesquisa são um dos casos mais claros para um sistema multiagente: uma única pergunta frequentemente se decompõe em várias subperguntas independentes, cada uma das quais precisa de sua própria exploração antes que algo possa ser concluído.

Em vez de um agente pesquisar o concorrente A, depois o concorrente B, depois o concorrente C em sequência, você pode iniciar um subagente por concorrente, executá-los todos simultaneamente e pagar apenas o custo de latência do ramo mais lento em vez da soma de todos eles.

A forma do "fan-out" (distribuição) deve vir da forma da entrada: um subagente por entidade nomeada, por fonte de dados ou por seção de um documento, não um número arbitrário escolhido antecipadamente.

Uma vez que os ramos terminem, uma etapa de coordenação precisa transformar vários conjuntos de descobertas independentes, às vezes parciais ou contraditórias, em um relatório coerente, que é um trabalho distinto da pesquisa em si e merece sua própria etapa.

Esta página cobre como decidir a contagem e a forma dos subagentes a partir da entrada, como fazer o "fan-out" e "fan-in" (distribuição e agregação) com asyncio.gather, e como sintetizar descobertas divergentes sem apenas concatená-las.

Receita

Cartão de receita de referência rápida - pronto para copiar e colar.

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"Pesquisa {topic} e reporta suas descobertas.",
                allowed_tools=["web_search"],
            ),
        ],
    )
    result = ""
    async for message in query(
        prompt=f"Use o subagente research-{topic} para investigar {topic}.",
        options=options,
    ):
        if message.get("type") == "subagent_result":
            result = message["result"]
    return result
 
async def main():
    topics = ["tamanho do mercado", "concorrentes", "risco regulatório"]
    findings = await asyncio.gather(*(research(t) for t in topics))
    # Passe as descobertas para uma chamada query() separada do coordenador para sintetizar.
    print(dict(zip(topics, findings)))
 
asyncio.run(main())

Quando usar isso:

• A entrada se decompõe naturalmente em várias perguntas de pesquisa, fontes ou entidades independentes: concorrentes, fontes de dados, seções de documentos.
• Você deseja que a pesquisa em cada peça ocorra simultaneamente em vez de uma após a outra.
• As descobertas precisam ser sintetizadas em um relatório, não apenas concatenadas na ordem em que terminam.
• Alguns ramos de pesquisa podem falhar ou retornar resultados parciais, e a saída final ainda precisa dizer algo coerente.

Exemplo de Trabalho

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def research_one(topic: str, question: str) -> str:
    """Executa um único subagente de pesquisa em um tópico e retorna suas descobertas."""
    options = AgentOptions(
        allowed_tools=["web_search", "web_fetch"],
        subagents=[
            SubagentConfig(
                name=f"research-{topic}",
                description=f"Pesquisa {topic} e responde: {question}",
                allowed_tools=["web_search", "web_fetch"],
            ),
        ],
    )
    findings = []
    async for message in query(
        prompt=f"Use o subagente research-{topic} para responder: {question}",
        options=options,
    ):
        if message.get("type") == "subagent_result":
            findings.append(message["result"])
    return "\n".join(findings) or "(nenhuma descoberta retornada)"
 
 
async def research_and_synthesize(topics: list[str], question: str) -> str:
    # Fan out: uma invocação de subagente por tópico, despachada simultaneamente.
    results = await asyncio.gather(
        *(research_one(topic, question) for topic in topics),
        return_exceptions=True,
    )
 
    # Alguns ramos podem falhar (timeout, sem resultados, uma fonte ruim). Continue
    # com o que retornou em vez de perder todo o lote devido a um erro.
    findings_by_topic = {}
    for topic, result in zip(topics, results):
        if isinstance(result, Exception):
            findings_by_topic[topic] = f"(pesquisa falhou: {result})"
        else:
            findings_by_topic[topic] = result
 
    # Fan in: uma chamada de coordenador sintetiza as descobertas, que podem ser
    # parciais ou conflitantes, em um único relatório em vez de apenas juntá-las
    # na ordem em que foram concluídas.
    coordinator_prompt = "Sintetize estas descobertas de pesquisa em um relatório coerente.\n"
    for topic, findings in findings_by_topic.items():
        coordinator_prompt += f"\n## {topic}\n{findings}\n"
    coordinator_prompt += (
        "\nChame a atenção para qualquer acordo, desacordo ou lacunas entre as fontes. "
        "Se a pesquisa de um tópico falhou, diga isso explicitamente em vez de omiti-la."
    )
 
    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=["Concorrente A", "Concorrente B", "Concorrente C"],
        question="Qual é o modelo de precificação atual e o mercado-alvo deles?",
    )
)

O que isso demonstra:

• Gerar um subagente por tópico de pesquisa, aqui um por concorrente, para que cada ramo explore independentemente e simultaneamente.
• Usar asyncio.gather com return_exceptions=True para que um ramo de pesquisa falho não derrube todo o lote.
• Rotular cada conjunto de descobertas por seu tópico de origem antes da síntese, para que o coordenador (e o leitor) possam atribuir as afirmações a uma fonte.
• Uma chamada query() separada do coordenador, com escopo allowed_tools=[], cujo único trabalho é sintetizar texto já coletado, não fazer mais pesquisas.
• Instruir explicitamente o coordenador a expor desacordos e lacunas em vez de simplesmente escolher uma versão da verdade silenciosamente.

Mergulho Profundo

Como Funciona

• Cada ramo de pesquisa é sua própria invocação de subagente com seu próprio contexto isolado, então o que um ramo lê ou raciocina nunca vaza para os resultados de outro ramo.
• asyncio.gather despacha todas as corrotinas de ramo ao mesmo tempo e retorna seus resultados na mesma ordem da entrada, independentemente de qual deles realmente termina primeiro, que é o que torna zip(topics, results) seguro.
• return_exceptions=True transforma a exceção de um ramo em um valor na lista de resultados em vez de propagá-la e cancelar as outras tarefas ainda em execução; sem isso, uma falha mata todo o gather.
• A etapa do coordenador é uma chamada query() simples, não outro subagente, porque seu trabalho (ler texto já coletado e escrever uma síntese) não precisa de ferramentas ou exploração adicional, apenas um contexto limpo com todas as descobertas à sua frente.
• Como o coordenador recebe apenas resultados de texto finais, não o raciocínio intermediário de cada ramo, ele sintetiza a partir dos mesmos resumos limpos que um revisor humano veria.

Dimensionando Subagentes de Pesquisa a Partir da Entrada

O número de subagentes deve sempre vir da contagem das unidades independentes na entrada, concorrentes, fontes ou seções, não de escolher um número que pareça razoável. Se a entrada não se decompõe em peças independentes, o "fan-out" adiciona sobrecarga sem adicionar paralelismo.

Sintetizando Descobertas Divergentes

• Rotule cada bloco de descobertas com sua fonte antes que ele chegue ao coordenador; uma parede de texto sem rótulo não pode ser atribuída ou verificada.
• Peça explicitamente ao coordenador para chamar a atenção para acordo, desacordo e lacunas, em vez de apenas "resumir", o que o convida a escolher silenciosamente uma versão quando as fontes entram em conflito.
• Passe ramos falhos ou parciais como descobertas explícitas ("pesquisa falhou: timeout") em vez de descartá-los, para que o relatório final reflita o que é realmente conhecido versus o que está faltando.
• Trate a saída do coordenador como o entregável, não os resultados brutos dos ramos; os ramos são notas de trabalho, a síntese é o relatório.

Notas de Python

# return_exceptions=True é o que torna a falha parcial sobrevivente.
# Sem isso, uma exceção cancela todas as outras tarefas em andamento.
results = await asyncio.gather(
    *(research_one(topic, question) for topic in topics),
    return_exceptions=True,
)
 
# zip() emparelha cada tópico com seu resultado na ordem de entrada, que asyncio.gather
# preserva, mesmo que os ramos completem em ordem arbitrária.
for topic, result in zip(topics, results):
    ...

Parâmetros e Valores de Retorno

Armadilhas

• Fazer "fan-out" de um subagente por elemento de uma lista ilimitada. Uma lista de 5 concorrentes é um bom "fan-out"; uma lista de 500 linhas é um incidente de custo. Correção: limite o número de subagentes de pesquisa concorrentes e agrupe ou amostre além desse limite.
• Chamar asyncio.gather sem return_exceptions=True. A exceção de um ramo cancela todos os outros ramos em andamento, então uma única fonte instável derruba um lote que estava bem. Correção: sempre passe return_exceptions=True e trate as exceções explicitamente ao desempacotar os resultados.
• Pular a etapa do coordenador e retornar os resultados brutos dos ramos. Descobertas concatenadas e sem rótulo de vários subagentes leem como ruído, não como um relatório, e escondem desacordos entre as fontes. Correção: sempre execute uma etapa de síntese sobre as descobertas coletadas antes de tratar a saída como final.
• Dar ao coordenador o mesmo acesso a ferramentas que os subagentes de pesquisa. O trabalho do coordenador é ler e escrever texto, não pesquisar ou buscar, então conceder a ele web_search/web_fetch amplia seu raio de explosão sem benefício. Correção: escopo o coordenador ao mínimo necessário, geralmente allowed_tools=[].
• Não rotular as descobertas por fonte antes da síntese. Se o prompt do coordenador for apenas um bloco de texto mesclado, ele não pode atribuir uma afirmação ao tópico ou fonte de onde veio, o que torna o desacordo invisível. Correção: estruture o prompt do coordenador com uma seção claramente rotulada por tópico.
• Assumir que um ramo falho deve ser simplesmente descartado do relatório. Omitir silenciosamente um tópico cuja pesquisa falhou faz com que o relatório final pareça completo quando não está. Correção: passe as falhas como descobertas explícitas ("pesquisa falhou: ...") para que o coordenador e o leitor vejam a lacuna.

Alternativas

FAQs

Relacionado

• Noções Básicas de Orquestração de Agentes - o modelo mental em que este padrão se baseia
• Padrões Orquestrador/Trabalhador com o Claude Agent SDK - o padrão geral de despacho e mesclagem que este especializa
• Guardrails para Sistemas Multiagentes - limitando a profundidade do "fan-out", o acesso a ferramentas e o gasto de tokens
• Delegando Trabalho para Subagentes no Claude Agent SDK - as mecânicas subjacentes de subagentes usadas aqui

Versões da Stack: Escrito contra a linha de modelos Claude atual por volta de Junho de 2026 - Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (o padrão), e Claude Haiku 4.5 - e o Claude Agent SDK (última versão). Nomes de modelos, versões de SDK e preços mudam rapidamente - verifique os detalhes atuais em platform.claude.com/docs antes de confiar neles.