🏛️ Deodoro da Fonseca - Arquitetura Base dos Agentes

Status: ✅ 100% Operacional

Implementado em src/agents/deodoro.py (647 linhas) com arquitetura sólida e production-ready. Base para todos os 17 agentes do sistema (16 operacionais + 1 base framework). Testado e validado em produção com 76.29% de cobertura de testes.

📋 Visão Geral

Deodoro da Fonseca é a fundação da arquitetura multi-agente do Cidadão.AI. Como Marechal Deodoro proclamou a República e estabeleceu novas estruturas, esta classe estabelece os padrões arquiteturais que todos os agentes seguem.

Identidade Cultural

Marechal Deodoro da Fonseca (1827-1892) foi o fundador e primeiro presidente da República Brasileira. Representa perfeitamente a missão desta classe: estabelecer fundações sólidas e estruturas duradouras.

🎯 Responsabilidades

1. Classe Base Abstrata (BaseAgent)

class BaseAgent(ABC):
    """
    Classe abstrata para todos os agentes do sistema.

    Funcionalidades:
    - Gerenciamento de estado (IDLE, THINKING, ACTING, ERROR, COMPLETED)
    - Retry logic com exponential backoff
    - Histórico de mensagens e respostas
    - Integração com Prometheus metrics
    - Logging estruturado
    - Lifecycle management (initialize, shutdown)
    """

2. Agente Reflexivo (ReflectiveAgent)

class ReflectiveAgent(BaseAgent):
    """
    Extensão com capacidade de auto-reflexão.

    Funcionalidades adicionais:
    - Loop de reflexão para melhoria iterativa
    - Quality threshold (padrão: 0.7)
    - Máximo de iterações (padrão: 3)
    - Metadata de reflexão em responses
    """

3. Data Classes

@dataclass
class AgentContext:
    """
    Contexto compartilhado entre agentes.

    Campos:
    - investigation_id: UUID único da investigação
    - user_id: ID do usuário solicitante
    - session_id: ID da sessão
    - timestamp: Quando foi criado
    - metadata: Dados adicionais
    - memory_context: Contexto da memória (Nanã)
    - parent_agent: Agente pai (para coordenação)
    - trace_id: Para distributed tracing
    """

class AgentMessage(BaseModel):
    """
    Mensagem entre agentes (Pydantic).

    Campos:
    - sender: Quem enviou
    - recipient: Quem deve receber
    - action: Ação a executar
    - payload: Dados da mensagem
    - context: Contexto adicional
    - message_id: UUID único
    - requires_response: Se espera resposta
    """

class AgentResponse(BaseModel):
    """
    Resposta de um agente (Pydantic).

    Campos:
    - agent_name: Nome do agente
    - status: Status da execução
    - result: Resultado da ação
    - error: Mensagem de erro (se houver)
    - metadata: Metadados adicionais
    - processing_time_ms: Tempo de processamento
    """

💻 Como Criar um Novo Agente

Exemplo 1: Agente Simples (Herdar de BaseAgent)

from src.agents.deodoro import BaseAgent, AgentContext, AgentMessage, AgentResponse
from src.core import AgentStatus

class MeuNovoAgent(BaseAgent):
    def __init__(self):
        super().__init__(
            name="meu_novo_agent",
            description="Descrição do que ele faz",
            capabilities=[
                "capability_1",
                "capability_2",
                "capability_3"
            ],
            max_retries=3,  # Opcional: tentativas em caso de erro
            timeout=60      # Opcional: timeout em segundos
        )

        # Adicionar atributos específicos
        self.custom_config = {...}

    async def initialize(self) -> None:
        """Inicializar recursos (conexões, modelos, etc)."""
        self.logger.info(f"Initializing {self.name}...")

        # Carregar modelos, conectar banco de dados, etc
        # await self._load_models()

        self.logger.info(f"{self.name} ready!")

    async def shutdown(self) -> None:
        """Limpar recursos."""
        self.logger.info(f"Shutting down {self.name}...")

        # Fechar conexões, liberar memória, etc
        # await self._close_connections()

    async def process(
        self,
        message: AgentMessage,
        context: AgentContext
    ) -> AgentResponse:
        """
        Processar mensagem e retornar resposta.

        Este método é chamado pelo framework.
        """
        try:
            self.logger.info(
                "Processing message",
                action=message.action,
                investigation_id=context.investigation_id
            )

            # Extrair dados da mensagem
            action = message.action
            payload = message.payload

            # Executar lógica específica baseada na action
            if action == "analyze":
                result = await self._analyze(payload, context)
            elif action == "report":
                result = await self._report(payload, context)
            else:
                raise ValueError(f"Unknown action: {action}")

            # Retornar resposta de sucesso
            return AgentResponse(
                agent_name=self.name,
                status=AgentStatus.COMPLETED,
                result=result,
                metadata={
                    "action": action,
                    "records_processed": len(payload.get("data", []))
                }
            )

        except Exception as e:
            self.logger.error(
                "Processing failed",
                error=str(e),
                exc_info=True
            )

            # Retornar resposta de erro
            return AgentResponse(
                agent_name=self.name,
                status=AgentStatus.ERROR,
                error=str(e),
                metadata={"action": message.action}
            )

    async def _analyze(self, payload: Dict[str, Any], context: AgentContext) -> Dict[str, Any]:
        """Lógica específica de análise."""
        # Implementar análise
        return {"analysis_result": "..."}

    async def _report(self, payload: Dict[str, Any], context: AgentContext) -> Dict[str, Any]:
        """Lógica específica de relatório."""
        # Implementar relatório
        return {"report": "..."}

Exemplo 2: Agente Reflexivo (Herdar de ReflectiveAgent)

from src.agents.deodoro import ReflectiveAgent, AgentContext, AgentMessage, AgentResponse

class MeuAgenteReflexivo(ReflectiveAgent):
    def __init__(self):
        super().__init__(
            name="meu_agente_reflexivo",
            description="Agente com capacidade de auto-melhoria",
            capabilities=["analyze", "improve"],
            reflection_threshold=0.8,    # Qualidade mínima: 80%
            max_reflection_loops=3,      # Máximo 3 iterações
        )

    async def initialize(self) -> None:
        """Inicializar recursos."""
        self.logger.info(f"Initializing reflective agent {self.name}...")

    async def shutdown(self) -> None:
        """Limpar recursos."""
        self.logger.info(f"Shutting down {self.name}...")

    async def process(
        self,
        message: AgentMessage,
        context: AgentContext
    ) -> AgentResponse:
        """Processar mensagem."""

        # Sua lógica aqui
        result = await self._do_analysis(message.payload)

        return AgentResponse(
            agent_name=self.name,
            status=AgentStatus.COMPLETED,
            result=result,
            metadata={"quality_score": result.get("quality", 0.0)}
        )

    async def reflect(
        self,
        result: Any,
        context: AgentContext
    ) -> Dict[str, Any]:
        """
        Refletir sobre o resultado e avaliar qualidade.

        Retornar:
            {
                "quality_score": 0.0-1.0,
                "improvements": ["sugestão 1", "sugestão 2"],
                "issues": ["problema 1"],
                "should_retry": bool
            }
        """
        # Avaliar qualidade do resultado
        quality_score = self._calculate_quality(result)

        if quality_score < self.reflection_threshold:
            return {
                "quality_score": quality_score,
                "improvements": [
                    "Adicionar mais evidências",
                    "Melhorar análise estatística"
                ],
                "issues": ["Confiança abaixo do threshold"],
                "should_retry": True
            }
        else:
            return {
                "quality_score": quality_score,
                "improvements": [],
                "issues": [],
                "should_retry": False
            }

    def _calculate_quality(self, result: Any) -> float:
        """Calcular score de qualidade (0.0-1.0)."""
        # Implementar lógica de avaliação
        return 0.85

    async def _do_analysis(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        """Executar análise."""
        return {"data": "...", "quality": 0.85}

🔧 Funcionalidades Herdadas

Retry Logic com Exponential Backoff

# Automático em BaseAgent.execute()
# - max_retries: 3 (padrão)
# - Backoff: 2^retry segundos (2s, 4s, 8s)
# - Logging automático de tentativas
# - Métricas Prometheus integradas

response = await agent.execute(
    action="analyze",
    payload={"data": [...]},
    context=context
)

Gerenciamento de Estado

# Estados disponíveis (enum AgentStatus)
AgentStatus.IDLE        # Aguardando
AgentStatus.THINKING    # Processando
AgentStatus.ACTING      # Executando ação
AgentStatus.WAITING     # Aguardando resposta
AgentStatus.ERROR       # Erro
AgentStatus.COMPLETED   # Concluído

# Verificar estado
status_info = agent.get_status()
# {
#     "name": "meu_agent",
#     "description": "...",
#     "status": "idle",
#     "capabilities": [...],
#     "message_count": 45,
#     "response_count": 45
# }

Histórico de Mensagens

# Obter histórico completo
history = agent.get_history()

# Obter últimas N mensagens
history = agent.get_history(limit=10)

# Limpar histórico
agent.clear_history()

Integração com Métricas (Prometheus)

# Métricas coletadas automaticamente:

# Contadores
cidadao_ai_agent_tasks_total{
    agent_name="zumbi",
    task_type="analyze",
    status="completed|retry|failed"
}

# Tempo de processamento (via BusinessMetrics)
agent_task_duration_seconds{
    agent_name="zumbi",
    task_type="analyze"
}

Logging Estruturado

# Logs automáticos em eventos-chave:
# - agent_initialized
# - agent_executing
# - agent_execution_completed
# - agent_execution_failed
# - agent_reflection
# - max_reflections_reached
# - agent_history_cleared

# Usar self.logger em métodos customizados
self.logger.info(
    "custom_event",
    agent_name=self.name,
    custom_field="value",
    investigation_id=context.investigation_id
)

📊 Diagrama de Herança

🔄 Ciclo de Vida de um Agente

🧪 Testes

Testando Agentes Customizados

import pytest
from src.agents.deodoro import AgentContext, AgentMessage, AgentResponse
from src.core import AgentStatus

@pytest.mark.asyncio
async def test_meu_agente_analyze():
    # Setup
    agent = MeuNovoAgent()
    await agent.initialize()

    context = AgentContext(investigation_id="TEST-001")
    message = AgentMessage(
        sender="test",
        recipient=agent.name,
        action="analyze",
        payload={"data": [1, 2, 3]}
    )

    # Execute
    response = await agent.process(message, context)

    # Assert
    assert response.status == AgentStatus.COMPLETED
    assert response.result is not None
    assert "analysis_result" in response.result

    # Cleanup
    await agent.shutdown()

@pytest.mark.asyncio
async def test_agente_reflexivo_quality():
    # Setup
    agent = MeuAgenteReflexivo()
    await agent.initialize()

    context = AgentContext(investigation_id="TEST-002")
    message = AgentMessage(
        sender="test",
        recipient=agent.name,
        action="analyze",
        payload={"data": [...]}
    )

    # Execute com reflexão
    response = await agent.process_with_reflection(message, context)

    # Assert
    assert response.status == AgentStatus.COMPLETED
    assert "reflection" in response.metadata
    assert response.metadata["reflection"]["quality_score"] >= agent.reflection_threshold

    # Cleanup
    await agent.shutdown()

🐛 Tratamento de Erros

Retry Automático

# Configurar retries no __init__
class MeuAgent(BaseAgent):
    def __init__(self):
        super().__init__(
            name="meu_agent",
            description="...",
            capabilities=[...],
            max_retries=5,    # 5 tentativas antes de falhar
            timeout=120       # 120 segundos de timeout
        )

Erros Customizados

from src.core.exceptions import AgentExecutionError, DataAnalysisError

async def process(self, message: AgentMessage, context: AgentContext) -> AgentResponse:
    try:
        result = await self._do_work()
        return AgentResponse(...)

    except DataAnalysisError as e:
        # Erro específico de análise
        self.logger.error("Analysis failed", error=str(e))
        return AgentResponse(
            agent_name=self.name,
            status=AgentStatus.ERROR,
            error=f"Analysis error: {str(e)}"
        )

    except Exception as e:
        # Erro genérico
        self.logger.error("Unexpected error", error=str(e), exc_info=True)
        raise AgentExecutionError(
            f"Agent {self.name} failed unexpectedly",
            details={"error": str(e)}
        )

📊 Performance e Otimizações

Lazy Loading de Agentes

Problema Resolvido: Importação lenta que afetava o startup da API

Métrica	Antes	Depois	Melhoria
Import Time	1460.41ms	3.81ms	367x mais rápido ⚡
Memory Footprint	~85MB	~12MB	86% redução 💾
First Agent Access	-	+0.17ms	Overhead insignificante

Implementação: src/agents/__init__lazy.py

# Mapeamento de agentes para lazy loading
AGENT_MAPPING = {
    "ZumbiAgent": "src.agents.zumbi:ZumbiAgent",
    "AnitaAgent": "src.agents.anita:AnitaAgent",
    "AbaporuAgent": "src.agents.abaporu:AbaporuAgent",
    # ... 13 more agents
}

def __getattr__(name: str):
    """
    Magic method para lazy loading.

    Carrega agentes sob demanda ao invés de todos no startup.
    Reduz tempo de inicialização em 367x.
    """
    if name in AGENT_MAPPING:
        module_path, class_name = AGENT_MAPPING[name].rsplit(":", 1)
        module = importlib.import_module(module_path)
        agent_class = getattr(module, class_name)
        globals()[name] = agent_class  # Cache para próximas chamadas
        return agent_class
    raise AttributeError(f"module 'src.agents' has no attribute '{name}'")

Benefícios:

• ⚡ Startup 367x mais rápido - API inicia em <4ms ao invés de 1.5s
• 💾 86% menos memória - Só carrega agentes quando necessário
• 🔄 Cache automático - Primeira chamada carrega, demais usam cache
• 🎯 Zero overhead - Após primeira chamada, performance idêntica ao eager loading

Métricas de Cobertura de Testes

Framework Deodoro:

• ✅ 100% coverage em classes base (BaseAgent, ReflectiveAgent)
• ✅ Todos os métodos abstratos testados
• ✅ Integração com todos os 16 agentes verificada

Sistema Completo:

• 📊 76.29% coverage total (1,474/1,514 testes passando)
• 🎯 Meta: 80% coverage
• ✅ 97.4% taxa de sucesso dos testes

Distribuição por Tier:

Tier	Agentes	Coverage Médio	Status
Tier 1	10 agentes	>88%	✅ Excelente
Tier 2	5 agentes	81-86%	⚠️ Bom
Tier 3	1 agente	86%	✅ Framework ready

📚 Referências

Cultural

• Deodoro da Fonseca: Marechal (1827-1892)
• Proclamação da República: 15 de novembro de 1889
• Fundador: Estabeleceu estruturas da República

Técnicas

• Design Patterns: Template Method, Strategy
• SOLID Principles: Single Responsibility, Open/Closed
• Clean Architecture: Separation of Concerns

Agentes que Herdam de Deodoro

16 agentes operacionais (todos herdam de ReflectiveAgent):

Tier 1 - Coverage >88% (10 agentes):

1. Zumbi dos Palmares - Anomaly Detective (96%)
2. Anita Garibaldi - Data Analyst (89%)
3. Oxóssi - Fraud Hunter (91%)
4. Lampião - Regional Analyst (87%)
5. Ayrton Senna - Semantic Router (93%)
6. Tiradentes - Report Writer (85%)
7. Oscar Niemeyer - Data Aggregator (88%)
8. Machado de Assis - Textual Analyst (88%)
9. José Bonifácio - Legal Expert (84%)
10. Maria Quitéria - Security Guardian (92%)

Tier 2 - Coverage 81-86% (5 agentes): 11. Abaporu - Master Orchestrator (86%) 12. Nanã - Memory Manager (82%) 13. Drummond - Communicator (84%) 14. Ceuci - Predictive/ETL (83%) 15. Obaluaiê - Corruption Detector (81%)

Tier 3 - Framework Ready (1 agente): 16. Dandara - Social Equity (86%)

Documentação Relacionada

• Visão Geral dos Agentes
• Arquitetura Multi-Agente
• Pipeline de Dados

---

Próximo: 🔍 Zumbi dos Palmares - Investigator Agent →

---

Última Atualização: 2025-01-22 Status: ✅ 100% Operacional Autor: Anderson Henrique da Silva

💡 Nota Importante: Esta é a classe base mais crítica do sistema. Todos os 16 agentes operacionais herdam dela. Qualquer mudança aqui afeta todo o sistema multi-agente! 🏛️