2.9 — Colaboração entre Agentes: Orquestração e Delegação de Tarefas (Multi-agent)

Objetivo da Aula

Ao concluir esta aula, você será capaz de:

Implementar o padrão Supervisor com 3 agentes especializados (Analista, Planejador, Executor)

Projetar tools de delegação que roteiam para agentes especializados

Adicionar mecanismo de consenso em consolidar_resultado

Comparar os 4 padrões de orquestração: Supervisor, Hierarchical, Group Chat, Swarm

Identificar quando sistemas multi-agente superam single-agent

Por que isso importa

Sistemas multi-agente são a fronteira atual da engenharia de IA aplicada. A motivação é clara:

Divisão cognitiva de trabalho: assim como em equipes humanas, especialistas são melhores do que generalistas para tarefas específicas. Um “Analista” treinado (via system prompt) para identificar problemas vai fazer isso melhor que um agente geral.

Paralelismo: múltiplos agentes podem trabalhar em paralelo em subtarefas independentes, reduzindo latência total.

Gerenciamento de contexto: cada agente especializado tem um contexto menor e mais focado, o que melhora a qualidade das respostas.

O starter implementa o padrão Supervisor — o mais simples e mais robusto para produção.

Conceitos Fundamentais

Decisão de Arquitetura: Topologias Multi-Agente

Quatro topologias com trade-offs distintos:
• Supervisor: um agente central delega para sub-agentes via tool_use. Controlável e auditável, mas gargalo no supervisor. Use quando há orquestração centralizada necessária.
• Group Chat / Debate: múltiplos agentes respondem ao mesmo prompt e debatem. Melhora qualidade para decisões complexas; custo alto (N respostas por rodada).
• Swarm / Handoff: um agente passa o controle para outro baseado em critérios. Boa para fluxos lineares onde o próximo especialista é óbvio; difícil auditar.
• Hierárquico: combina Supervisor e Swarm em múltiplos níveis. Escala para tarefas complexas; máximo custo e complexidade de debug.
Todas as topologias custam 2-3x mais tokens que um agente único. Especialização = system prompt diferente por agente; delegação = tool_use que chama o sub-agente.

Os 4 Padrões de Orquestração

Supervisor (Hierarchical)

[Supervisor] → delega → [Analista]

             → delega → [Planejador]

             → delega → [Executor]

             ← consolida todos os resultados

Um agente central orquestra. O supervisor decide quando delegar, para quem e consolida os resultados. É o padrão mais controlado e auditável.

Group Chat (Debate)

[Agente A] ↔ [Agente B] ↔ [Agente C]

Agentes debatem até consenso. Útil para decisões criativas, design de sistemas, revisão de código. Mais tokens, maior latência, melhor qualidade para problemas abertos.

Swarm

[Agente A] → handoff → [Agente B] → handoff → [Agente C]

Agentes passam o controle entre si baseado em expertise. Padrão do CrewAI. Bom para pipelines sequenciais com handoffs naturais.

Hierarchical Planner

[Planner] → subtarefas → [Sub-supervisor 1] → [Worker 1]

                       → [Sub-supervisor 2] → [Worker 2]

Para problemas muito complexos. Alto overhead, difícil de debugar.

O Supervisor como Orquestrador via Tool Use

O insight central: o supervisor usa tool_use para delegar. As “tools” do supervisor não fazem operações — elas chamam outros agentes:

TOOLS_SUPERVISOR = [

    {

        'name': 'delegar_analise',

        'description': 'Delega análise da situação ao Agente Analista.',

        'input_schema': {

            'type': 'object',

            'properties': {

                'subtarefa': {'type': 'string'},

                'contexto': {'type': 'string'},

            },

            'required': ['subtarefa'],

        },

    },

    # ... planejamento, execução, consolidar

]



def executar_tool_supervisor(nome: str, args: dict, tarefa_original: str) -> str:

    if nome == 'delegar_analise':

        return agente_analista(args['subtarefa'], args.get('contexto', ''))

    if nome == 'delegar_planejamento':

        return agente_planejador(args['subtarefa'], args.get('analise_previa', ''))

    if nome == 'delegar_execucao':

        return agente_executor(args['subtarefa'], args.get('plano', ''))

    if nome == 'consolidar_resultado':

        # TODO: mecanismo de consenso

        return f'RESULTADO FINAL:\n{args["conclusao_final"]}'

O supervisor decide sozinho quando e em que ordem delegar. Ele pode delegar análise antes do planejamento (que faz sentido), ou pode pular direto para execução se achar que não precisa de análise.

Agentes Especializados via System Prompt

def agente_analista(tarefa: str, contexto: str = '') -> str:

    response = client.messages.create(

        model='claude-haiku-4-5-20251001',

        max_tokens=400,

        system='Você é um analista. Analise a situação e identifique os 3 principais pontos críticos.',

        messages=[{'role': 'user', 'content': f'Tarefa: {tarefa}\nContexto: {contexto}'}],

    )

    return response.content[0].text



def agente_planejador(tarefa: str, analise: str) -> str:

    response = client.messages.create(

        model='claude-haiku-4-5-20251001',

        max_tokens=400,

        system='Você é um planejador. Com base na análise, crie um plano de 3-5 passos claros.',

        messages=[{'role': 'user', 'content': f'Tarefa: {tarefa}\nAnálise: {analise}'}],

    )

    return response.content[0].text

O system prompt define a “personalidade” e “expertise” de cada agente. É a forma mais simples de especialização — sem fine-tuning, sem modelos diferentes.

TODO: Mecanismo de Consenso em consolidar_resultado

def executar_tool_supervisor(nome: str, args: dict, tarefa_original: str) -> str:

    if nome == 'consolidar_resultado':

        # TODO: consenso entre analise e execucao

        analise = args.get('analise', '')

        execucao = args.get('execucao', '')

        conclusao = args['conclusao_final']

        

        # Verificar convergência: analise e execução apontam para mesma direção?

        if analise and execucao:

            response = client.messages.create(

                model='claude-haiku-4-5-20251001',

                max_tokens=100,

                messages=[{

                    'role': 'user',

                    'content': f'''Há contradição entre análise e execução?

ANÁLISE: {analise[:300]}

EXECUÇÃO: {execucao[:300]}

Responda: CONVERGENTE ou DIVERGENTE. Uma palavra apenas.'''

                }],

            )

            veredicto = response.content[0].text.strip().upper()

            

            if 'DIVERGENTE' in veredicto:

                # Pedir revisão ao analista

                revisao = agente_analista(

                    f'Revise: execução contradiz análise inicial. Análise: {analise[:200]}. Execução: {execucao[:200]}',

                    contexto='Revisão de divergência'

                )

                return f'RESULTADO (após revisão de divergência):\n{conclusao}\n\nNota do analista: {revisao[:200]}'

        

        return f'RESULTADO FINAL:\n{conclusao}'

Aprofundamento Técnico

Custo de Tokens em Sistemas Multi-Agente

Para a tarefa do starter com 4 delegações:

Delegação	Chamadas LLM	Tokens estimados
Supervisor (decisão)	3-4	~500 tokens/iter
Analista	1	~800 tokens
Planejador	1	~800 tokens
Executor	1	~1200 tokens
Consolidação	1	~600 tokens
Total	~7	~4700 tokens

Comparando com single-agent: ~2-3 chamadas, ~2000 tokens. Multi-agent usa 2-3x mais tokens. O trade-off: qualidade e especialização.

Comunicação Assíncrona (Padrão Swarm)

Para sistemas que precisam de agentes paralelos:

import asyncio



async def delegar_paralelo(tarefa: str):

    """Executa analista e pesquisador em paralelo."""

    analise_task = asyncio.create_task(

        asyncio.to_thread(agente_analista, tarefa)

    )

    pesquisa_task = asyncio.create_task(

        asyncio.to_thread(agente_pesquisador, tarefa)

    )

    

    analise, pesquisa = await asyncio.gather(analise_task, pesquisa_task)

    return analise, pesquisa



# No supervisor: await delegar_paralelo(tarefa)

Reduz latência total quando subtarefas são independentes.

Supervisores Hierárquicos

Para problemas muito complexos, o supervisor pode delegar para sub-supervisores:

def super_supervisor(tarefa_complexa: str) -> str:

    """Supervisor de supervisores."""

    # Divide em domínios

    dominio_tecnico = supervisor_tecnico(tarefa_complexa)

    dominio_negocio = supervisor_negocio(tarefa_complexa)

    

    # Integra resultados de múltiplos supervisores

    return integrar_dominios(dominio_tecnico, dominio_negocio)

Exemplos Anotados

Exemplo 1: Supervisor Completo com TODO Resolvido

import os

import anthropic



client = anthropic.Anthropic(api_key=os.environ.get('ANTHROPIC_API_KEY'))



def agente_analista(tarefa: str, contexto: str = '') -> str:

    return client.messages.create(

        model='claude-haiku-4-5-20251001', max_tokens=400,

        system='Você é um analista. Identifique os 3 principais pontos críticos.',

        messages=[{'role': 'user', 'content': f'Tarefa: {tarefa}\nContexto: {contexto}'}],

    ).content[0].text



def agente_planejador(tarefa: str, analise: str) -> str:

    return client.messages.create(

        model='claude-haiku-4-5-20251001', max_tokens=400,

        system='Você é um planejador. Crie um plano de 3-5 passos baseado na análise.',

        messages=[{'role': 'user', 'content': f'Tarefa: {tarefa}\nAnálise: {analise}'}],

    ).content[0].text



def agente_executor(tarefa: str, plano: str) -> str:

    return client.messages.create(

        model='claude-haiku-4-5-20251001', max_tokens=600,

        system='Você é um executor. Execute o plano e relate o resultado.',

        messages=[{'role': 'user', 'content': f'Tarefa: {tarefa}\nPlano: {plano}'}],

    ).content[0].text



def executar_tool_supervisor(nome: str, args: dict, tarefa_original: str) -> str:

    if nome == 'delegar_analise':

        print('  → [ANALISTA] executando...')

        return agente_analista(args['subtarefa'], args.get('contexto', ''))

    

    if nome == 'delegar_planejamento':

        print('  → [PLANEJADOR] executando...')

        return agente_planejador(args['subtarefa'], args.get('analise_previa', ''))

    

    if nome == 'delegar_execucao':

        print('  → [EXECUTOR] executando...')

        return agente_executor(args['subtarefa'], args.get('plano', ''))

    

    if nome == 'consolidar_resultado':

        print('  → [CONSOLIDAÇÃO]')

        analise = args.get('analise', '')

        execucao = args.get('execucao', '')

        

        # TODO: mecanismo de consenso

        if analise and execucao:

            veredicto = client.messages.create(

                model='claude-haiku-4-5-20251001', max_tokens=10,

                messages=[{

                    'role': 'user',

                    'content': f'Análise e execução convergem? CONVERGENTE ou DIVERGENTE.\nA:{analise[:200]}\nE:{execucao[:200]}'

                }],

            ).content[0].text.strip().upper()

            

            if 'DIVERGENTE' in veredicto:

                revisao = agente_analista(f'Divergência detectada. Revise: A:{analise[:100]} E:{execucao[:100]}')

                return f'RESULTADO (revisado):\n{args["conclusao_final"]}\nNota: {revisao[:150]}'

        

        return f'RESULTADO FINAL:\n{args["conclusao_final"]}'

    

    return 'tool desconhecida'

Exemplo 2: Trace de Execução do Supervisor

[SUPERVISOR] Tarefa: Nossa startup está perdendo clientes...



[SUPERVISOR iter 1]

  ⇒ delegar_analise

  → [ANALISTA] executando...

  ⇐ "Pontos críticos: 1) Preço 40% acima do mercado 2) UX inferior..."



[SUPERVISOR iter 2]

  ⇒ delegar_planejamento

  → [PLANEJADOR] executando...

  ⇐ "Plano: 1) Auditoria de preços 2) Research com clientes perdidos 3) Roadmap de UX..."



[SUPERVISOR iter 3]

  ⇒ delegar_execucao

  → [EXECUTOR] executando...

  ⇐ "Executando plano: Auditoria revelou que preço deve reduzir 20%..."



[SUPERVISOR iter 4]

  ⇒ consolidar_resultado

  → [CONSOLIDAÇÃO]

  Consenso: CONVERGENTE

  ⇐ "RESULTADO FINAL: Estratégia de retenção..."

Padrões e Armadilhas

Padrões

Padrão 1: System prompts de agentes especializados são concisos e específicos

# VAGO (agente geral):

system='Você é um assistente útil.'



# ESPECÍFICO (agente especializado):

system='Você é um analista estratégico. SEMPRE identifique exatamente 3 pontos críticos.'

Padrão 2: Supervisor passa contexto relevante entre delegações

# O analista viu a tarefa — o planejador precisa da análise

'delegar_planejamento': {

    'subtarefa': tarefa,

    'analise_previa': resultado_do_analista,  # ← passa contexto

}

Padrão 3: Consenso com modelo barato, não com o modelo principal

# Verificação de convergência: haiku (barato, rápido)

veredicto = client.messages.create(model='claude-haiku-4-5-20251001', max_tokens=10, ...)

Armadilhas

⚠️ Armadilha 1: Custo de tokens escala linearmente com agentes 3 agentes = ~3x os tokens de um single-agent. Calcule o ROI antes de adicionar agentes.

⚠️ Armadilha 2: O supervisor pode ficar em loop se não consolidar

# Sem 'consolidar_resultado', o supervisor pode continuar delegando indefinidamente

# max_iter=8 é o safety valve, mas force o consolidar no system prompt:

system='... ao ter TODOS os resultados necessários, OBRIGATORIAMENTE chame consolidar_resultado'

⚠️ Armadilha 3: Contexto perdido entre delegações Cada agente começa do zero — só tem o que o supervisor enviou. Se o executor precisa da análise original, o supervisor deve passá-la explicitamente no campo contexto do schema.

⚗ Laboratório prático — mozak.tech
Se não for realizar o laboratório, pule para o próximo capítulo.

Ponte para o Lab

O TODO do starter está em executar_tool_supervisor, no bloco 'consolidar_resultado':

if nome == 'consolidar_resultado':

    # TODO: implementar mecanismo de consenso

    # Se analise e execucao concordam → retornar consolidado

    # Se divergem → solicitar revisão ao analista

    return f'RESULTADO FINAL:\n{args["conclusao_final"]}'

Implemente o mecanismo de consenso: 1. Verificar se analise e execucao estão nos args 2. Se sim, pedir ao Claude para avaliar convergência 3. Se DIVERGENTE, chamar agente_analista com contexto da divergência e adicionar nota ao resultado 4. Se CONVERGENTE (ou campos ausentes), retornar o resultado diretamente

Rode com a tarefa do main() e observe a sequência de delegações. Se a análise e execução divergirem, o mecanismo de revisão deve ser acionado.

Agora você está pronto para o lab.