🎨 Abaporu - Agente Orquestrador

Status: ✅ 100% Operacional (Produção)

Implementado em src/agents/abaporu.py (1.247 linhas) com 86% de cobertura de testes. Herda de ReflectiveAgent com quality threshold 0.8 e max 3 reflection iterations. Produção desde outubro/2025 via Railway - Master Orchestrator do sistema.

📋 Visão Geral

Abaporu é o agente mestre do sistema Cidadão.AI, responsável por orquestrar investigações complexas coordenando múltiplos agentes especializados. Inspirado na obra icônica de Tarsila do Amaral, representa a "antropofagia" de dados - consumindo informações brutas e transformando em insights acionáveis.

🎯 Capacidades Principais

1. Orquestração de Investigações 🎼

Planejamento estratégico de investigações
Decomposição de tarefas complexas
Coordenação multi-agente paralela
Síntese de resultados distribuídos

2. Auto-Reflexão e Melhoria 🪞

REFLECTION_THRESHOLD = 0.8  # Qualidade mínima aceitável

Avaliação contínua de qualidade
Replanejamento adaptativo
Aprendizado com resultados
Otimização de estratégias

3. Gestão de Contexto 🧠

Manutenção de estado da investigação
Rastreamento de progresso
Gestão de prioridades
Histórico de decisões

4. Roteamento Inteligente 🛤️

Seleção otimizada de agentes
Balanceamento de carga
Fallback strategies
Paralelização de tarefas

💻 Implementação Técnica

Estrutura Principal

class MasterAgent(ReflectiveAgent):
    def __init__(self):
        super().__init__(
            agent_id="abaporu",
            name="Abaporu",
            description="Master orchestrator for multi-agent investigations",
            capabilities=[
                "investigation_planning",
                "multi_agent_coordination",
                "result_synthesis",
                "adaptive_strategy",
                "quality_assurance"
            ]
        )
        self.reflection_threshold = 0.8
        self.max_reflection_attempts = 3

Processo de Investigação

from src.agents.abaporu import MasterAgent, InvestigationPlan

# Inicializar o mestre
abaporu = MasterAgent()

# Criar plano de investigação
plan = InvestigationPlan(
    investigation_id="INV-2024-HEALTH-001",
    title="Análise de Contratos do Ministério da Saúde",
    scope={
        "target": "health_ministry_contracts",
        "period": "2024",
        "focus": ["overpricing", "vendor_concentration", "temporal_patterns"]
    },
    priority="high"
)

# Executar investigação
result = await abaporu.coordinate_investigation(plan)

# Estrutura do resultado
{
    "investigation_id": "INV-2024-HEALTH-001",
    "status": "completed",
    "quality_score": 0.92,
    "phases_completed": 4,
    "agents_involved": ["zumbi", "anita", "tiradentes", "nana"],
    "findings": {
        "critical_issues": 3,
        "anomalies_detected": 15,
        "patterns_identified": 7,
        "recommendations": 12
    },
    "synthesis": {
        "executive_summary": "...",
        "key_findings": [...],
        "risk_assessment": "HIGH",
        "next_steps": [...]
    },
    "reflection_iterations": 1,
    "total_time": "127.3s"
}

🔄 Fluxo de Orquestração

Diagrama de Processo

Estratégias de Coordenação

1. Investigação Paralela

async def parallel_investigation(self, tasks):
    """Executa tarefas em paralelo para eficiência"""
    agent_tasks = []
    
    for task in tasks:
        agent = self.select_best_agent(task)
        agent_tasks.append(
            self.delegate_to_agent(agent, task)
        )
    
    results = await asyncio.gather(*agent_tasks)
    return self.merge_results(results)

2. Investigação em Pipeline

async def pipeline_investigation(self, stages):
    """Executa em sequência quando há dependências"""
    result = None
    
    for stage in stages:
        agent = self.get_agent(stage.agent_id)
        result = await agent.process(
            self.prepare_message(stage, result)
        )
        
        if result.quality < self.reflection_threshold:
            result = await self.reflect_and_improve(stage, result)
    
    return result

📊 Métricas e Performance

KPIs Operacionais

Métrica	Valor	Meta	Status
Cobertura de Testes	86%	>80%	✅ Tier 2
Taxa de Sucesso	94%	>90%	✅
Tempo Médio (investigação)	98s	<120s	✅
Reflexões Necessárias	12%	<15%	✅
Qualidade Média	0.91	>0.85	✅
Reflection Threshold	0.8	-	⚙️ Auto-melhoria
Max Iterations	3	-	⚙️ Quality control

Estatísticas de Uso

# Distribuição de agentes por investigação
{
    "zumbi": 98%,      # Quase sempre usado
    "anita": 85%,      # Análise frequente
    "tiradentes": 95%, # Relatórios sempre
    "nana": 78%,       # Contexto histórico
    "machado": 45%,    # Quando há documentos
    "dandara": 32%,    # Questões sociais
    "senna": 89%,      # Roteamento comum
}

🧠 Sistema de Reflexão

Processo de Auto-Avaliação

async def reflect_on_quality(self, result):
    """Sistema de reflexão para melhoria contínua"""
    
    quality_metrics = {
        "completeness": self.assess_completeness(result),
        "accuracy": self.assess_accuracy(result),
        "relevance": self.assess_relevance(result),
        "actionability": self.assess_actionability(result)
    }
    
    overall_quality = sum(quality_metrics.values()) / len(quality_metrics)
    
    if overall_quality < self.reflection_threshold:
        reflection = {
            "current_quality": overall_quality,
            "gaps": self.identify_gaps(quality_metrics),
            "improvement_plan": self.create_improvement_plan(quality_metrics),
            "additional_agents": self.suggest_additional_agents(result)
        }
        
        return await self.execute_improvement_plan(reflection)
    
    return result

Gatilhos de Reflexão

Qualidade < 0.8: Reflexão automática
Anomalias não explicadas: Investigação adicional
Conflitos entre agentes: Reconciliação
Dados insuficientes: Expansão de escopo

🔧 Configuração e Customização

Parâmetros de Configuração

# config/abaporu.yaml
abaporu:
  reflection_threshold: 0.8
  max_reflection_attempts: 3
  parallel_agent_limit: 5
  timeout_per_agent: 60s
  cache_ttl: 3600s
  
  strategies:
    default: "balanced"
    high_priority: "aggressive"
    exploratory: "thorough"
    
  agent_weights:
    zumbi: 1.0
    anita: 0.9
    tiradentes: 1.0
    nana: 0.7

Personalização por Domínio

# Configurações específicas por tipo de investigação
INVESTIGATION_PROFILES = {
    "contracts": {
        "primary_agents": ["zumbi", "anita"],
        "focus": ["pricing", "vendors", "patterns"],
        "depth": "comprehensive"
    },
    "policies": {
        "primary_agents": ["bonifacio", "machado"],
        "focus": ["effectiveness", "compliance"],
        "depth": "analytical"
    },
    "social": {
        "primary_agents": ["dandara", "anita"],
        "focus": ["equity", "inclusion", "impact"],
        "depth": "detailed"
    }
}

🧪 Testes e Validação

Suite de Testes

# Testes unitários
pytest tests/unit/test_agents/test_abaporu.py -v

# Testes de integração
pytest tests/integration/test_multi_agent_coordination.py

# Testes de stress
pytest tests/performance/test_abaporu_load.py

# Cobertura completa
pytest tests/ -k abaporu --cov=src.agents.abaporu

Cenários de Teste

✅ Orquestração simples (2 agentes)
✅ Orquestração complexa (5+ agentes)
✅ Reflexão e melhoria
✅ Tratamento de falhas
✅ Timeout e recuperação
✅ Cache e performance

🚀 Recursos Avançados

1. Aprendizado Adaptativo

# Sistema aprende com investigações passadas
self.learning_system.update(
    investigation_id=result.id,
    strategies_used=plan.strategies,
    outcome_quality=result.quality,
    time_taken=result.duration
)

2. Previsão de Complexidade

# Estima recursos necessários
complexity_estimate = self.estimate_complexity(investigation_scope)
{
    "estimated_time": "120-180s",
    "agents_needed": ["zumbi", "anita", "tiradentes"],
    "reflection_probability": 0.15,
    "confidence": 0.88
}

3. Modo Exploratório

# Para investigações sem escopo definido
exploration_mode = {
    "start_broad": True,
    "progressive_refinement": True,
    "discover_patterns": True,
    "no_assumptions": True
}

🏗️ Herança e Arquitetura

Relação com Deodoro

Abaporu herda de ReflectiveAgent (definido em src/agents/deodoro.py):

class MasterAgent(ReflectiveAgent):
    """
    Master Orchestrator - herda todas as capacidades do Deodoro.

    Funcionalidades herdadas:
    - Quality threshold: 0.8 (80% de confiança mínima)
    - Max reflection loops: 3 (até 3 iterações de melhoria)
    - Retry logic com exponential backoff
    - State management (IDLE → THINKING → ACTING → COMPLETED)
    - Prometheus metrics integration
    - Structured logging
    """

Capacidades Exclusivas do Abaporu:

🎼 Orquestração multi-agente (coordena 16 agentes)
🧠 Planejamento estratégico (decompõe tarefas complexas)
🔄 Síntese de resultados (agrega outputs de múltiplos agentes)
📊 Gestão de prioridades (otimiza ordem de execução)
🎯 Balanceamento de carga (distribui trabalho eficientemente)

Padrão de Coordenação:

📋 Plan: Analisa escopo e cria plano de investigação
🎯 Delegate: Seleciona e distribui tarefas aos agentes especializados
📊 Collect: Agrega resultados de múltiplos agentes
🔄 Reflect: Avalia qualidade e replanja se necessário (< 0.8)
📝 Synthesize: Consolida insights e gera relatório final

Arquitetura de Coordenação

📚 Melhores Práticas

Para Desenvolvedores

Sempre defina timeouts para evitar travamentos
Use cache para investigações repetitivas
Configure thresholds baseados no domínio
Monitore métricas de reflexão

Para Usuários

Seja específico no escopo da investigação
Defina prioridades claras
Revise planos antes de executar
Interprete resultados no contexto

🔗 Integração com Frontend

API Endpoints

# Criar investigação
POST /api/v1/investigations
{
    "title": "Análise de Contratos 2024",
    "scope": {...},
    "priority": "high"
}

# Acompanhar progresso
GET /api/v1/investigations/{id}/progress

# Obter resultados
GET /api/v1/investigations/{id}/results

WebSocket para Tempo Real

// Conectar ao WebSocket
const ws = new WebSocket('ws://localhost:8000/ws/investigations/{id}');

ws.onmessage = (event) => {
    const update = JSON.parse(event.data);
    console.log(`Fase: ${update.phase}, Progresso: ${update.progress}%`);
};

📚 Referências

Documentação Relacionada

Deodoro - Base Framework - Classe base que Abaporu herda
Arquitetura Multi-Agente
Pipeline de Dados
Visão Geral dos Agentes

Cultural

Abaporu (1928) - Obra icônica de Tarsila do Amaral
Antropofagia: Movimento cultural brasileiro de "devorar" influências externas
Legado: Transformação criativa de dados brutos em insights únicos

Técnicas

Master-Slave Pattern: Coordenação centralizada de agentes
MapReduce: Distribuição paralela e agregação de resultados
Reflective Architecture: Auto-avaliação e melhoria contínua
Pipeline Orchestration: Sequenciamento inteligente de tarefas

Anterior: 🏛️ Deodoro - Base Framework ← Próximo: 🔍 Zumbi dos Palmares - Investigator Agent →

Última Atualização: 2025-01-22 Status: ✅ Tier 2 - 86% Coverage Autor: Anderson Henrique da Silva

💡 Destaque: Abaporu é o Master Orchestrator que coordena todos os 16 agentes operacionais! 🎨

📋 Visão Geral​

🎯 Capacidades Principais​

1. Orquestração de Investigações 🎼​

2. Auto-Reflexão e Melhoria 🪞​

3. Gestão de Contexto 🧠​

4. Roteamento Inteligente 🛤️​

💻 Implementação Técnica​

Estrutura Principal​

Processo de Investigação​

🔄 Fluxo de Orquestração​

Diagrama de Processo​

Estratégias de Coordenação​

1. Investigação Paralela​

2. Investigação em Pipeline​

📊 Métricas e Performance​

KPIs Operacionais​

Estatísticas de Uso​

🧠 Sistema de Reflexão​

Processo de Auto-Avaliação​

Gatilhos de Reflexão​

🔧 Configuração e Customização​

Parâmetros de Configuração​

Personalização por Domínio​

🧪 Testes e Validação​

Suite de Testes​

Cenários de Teste​

🚀 Recursos Avançados​

1. Aprendizado Adaptativo​

2. Previsão de Complexidade​

3. Modo Exploratório​

🏗️ Herança e Arquitetura​

Relação com Deodoro​

Arquitetura de Coordenação​

📚 Melhores Práticas​

Para Desenvolvedores​

Para Usuários​

🔗 Integração com Frontend​

API Endpoints​

WebSocket para Tempo Real​

📚 Referências​

Documentação Relacionada​

Cultural​

Técnicas​