"""
Module: agents.drummond
Codinome: Carlos Drummond de Andrade - Comunicador do Povo
Description: Agent specialized in multi-channel communication and natural language generation
Author: Anderson H. Silva
Date: 2025-07-23
License: Proprietary - All rights reserved
"""
import asyncio
from src.core import json_utils
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional, Tuple, Union
from dataclasses import dataclass
from enum import Enum
import numpy as np
import pandas as pd
from pydantic import BaseModel, Field as PydanticField
from src.agents.deodoro import BaseAgent, AgentContext, AgentMessage, AgentResponse, AgentStatus
from src.core import get_logger
from src.core.exceptions import AgentExecutionError, DataAnalysisError
from src.services.chat_service import IntentType, Intent
from src.memory.conversational import ConversationalMemory, ConversationContext
from src.services.maritaca_client import MaritacaClient, MaritacaModel, MaritacaMessage
class CommunicationChannel(Enum):
"""Communication channels supported."""
EMAIL = "email"
SMS = "sms"
WHATSAPP = "whatsapp"
TELEGRAM = "telegram"
WEBHOOK = "webhook"
PUSH_NOTIFICATION = "push_notification"
SLACK = "slack"
DISCORD = "discord"
PORTAL_WEB = "portal_web"
API_CALLBACK = "api_callback"
class MessagePriority(Enum):
"""Priority levels for messages."""
LOW = "low"
NORMAL = "normal"
HIGH = "high"
URGENT = "urgent"
CRITICAL = "critical"
class MessageType(Enum):
"""Types of messages."""
ALERT = "alert"
REPORT = "report"
NOTIFICATION = "notification"
SUMMARY = "summary"
WARNING = "warning"
INFORMATION = "information"
URGENT_ACTION = "urgent_action"
@dataclass
class CommunicationTarget:
"""Target for communication."""
target_id: str
name: str
channels: List[CommunicationChannel]
preferred_language: str
contact_info: Dict[str, str]
notification_preferences: Dict[str, Any]
timezone: str
active_hours: Dict[str, str]
@dataclass
class MessageTemplate:
"""Template for message generation."""
template_id: str
message_type: MessageType
language: str
subject_template: str
body_template: str
variables: List[str]
formatting_rules: Dict[str, Any]
channel_adaptations: Dict[CommunicationChannel, Dict[str, str]]
@dataclass
class CommunicationResult:
"""Result of communication attempt."""
message_id: str
target_id: str
channel: CommunicationChannel
status: str # "sent", "failed", "pending", "delivered", "read"
sent_at: datetime
delivered_at: Optional[datetime]
read_at: Optional[datetime]
error_message: Optional[str]
retry_count: int
metadata: Dict[str, Any]
class CommunicationAgent(BaseAgent):
"""
Carlos Drummond de Andrade - Comunicador do Povo
MISSÃO:
Geração automática de comunicações, alertas e notificações multi-canal,
traduzindo insights técnicos em linguagem acessível ao cidadão.
ALGORITMOS E TÉCNICAS IMPLEMENTADAS:
1. GERAÇÃO DE LINGUAGEM NATURAL (NLG):
- Template-based Generation para mensagens estruturadas
- Neural Language Models (GPT/BERT) para texto livre
- Adaptive Text Generation baseado no perfil do usuário
- Algoritmo de Simplificação Linguística automática
- Style Transfer para adequação de tom e registro
2. SISTEMA DE NOTIFICAÇÕES MULTI-CANAL:
- Priority Queue Algorithm para ordenação de mensagens
- Circuit Breaker Pattern para canais instáveis
- Exponential Backoff para retry de falhas
- Rate Limiting por canal e destinatário
- Deduplication Algorithm para evitar spam
3. PERSONALIZAÇÃO E SEGMENTAÇÃO:
- Collaborative Filtering para preferências
- Clustering de audiências por perfil comportamental
- A/B Testing automático para otimização de mensagens
- Sentiment Analysis para ajuste de tom
- Demographic Segmentation com ML
4. ANÁLISE DE ENGAJAMENTO:
- Click-through Rate (CTR) tracking
- Message Effectiveness Scoring
- Response Time Analysis
- Channel Performance Optimization
- Conversion Funnel Analysis
5. PROCESSAMENTO DE LINGUAGEM NATURAL:
- Named Entity Recognition (NER) para contextualização
- Text Summarization para relatórios executivos
- Keyword Extraction para tags automáticas
- Language Detection automática
- Translation API integration para multilíngue
6. SISTEMA DE TEMPLATES INTELIGENTES:
- Dynamic Template Selection baseado em contexto
- Variable Substitution com validação
- Conditional Logic em templates
- Template A/B Testing automático
- Version Control para templates
CANAIS DE COMUNICAÇÃO SUPORTADOS:
1. **Email**: SMTP/API integration com HTML/Text
2. **SMS**: Twilio/AWS SNS integration
3. **WhatsApp**: WhatsApp Business API
4. **Telegram**: Bot API com rich formatting
5. **Push Notifications**: Firebase/APNs
6. **Webhooks**: HTTP callbacks personalizados
7. **Slack/Discord**: Workspace integrations
8. **Portal Web**: In-app notifications
9. **API Callbacks**: System-to-system communication
10. **Voice**: Text-to-Speech para acessibilidade
TÉCNICAS DE OTIMIZAÇÃO:
- **Send Time Optimization**: ML para horário ideal
- **Content Optimization**: A/B testing automático
- **Frequency Capping**: Prevenção de fatiga de mensagem
- **Deliverability Optimization**: Reputation management
- **Cross-channel Orchestration**: Jornadas multi-touch
ALGORITMOS DE PERSONALIZAÇÃO:
- **Collaborative Filtering**: CF(u,i) = Σₖ sim(u,k) × rₖᵢ
- **Content-Based Filtering**: Cosine similarity entre perfis
- **Matrix Factorization**: SVD para recomendação de conteúdo
- **Clustering**: K-means para segmentação de audiência
- **Classification**: SVM para predição de engajamento
MÉTRICAS DE PERFORMANCE:
- **Delivery Rate**: >98% para emails, >95% para SMS
- **Open Rate**: >25% média (varia por canal)
- **Click Rate**: >3% para comunicações governamentais
- **Response Time**: <30s para canais síncronos
- **Escalabilidade**: 100K+ mensagens/hora
COMPLIANCE E SEGURANÇA:
- **LGPD**: Consentimento e opt-out automático
- **CAN-SPAM**: Compliance com leis anti-spam
- **GDPR**: Para usuários europeus
- **Encryption**: TLS/AES para dados sensíveis
- **Audit Trail**: Log completo de comunicações
INTEGRAÇÃO COM OUTROS AGENTES:
- **Obaluaiê**: Alertas de corrupção críticos
- **Zumbi**: Notificações de anomalias
- **Tiradentes**: Relatórios de risco
- **Niemeyer**: Inclusão de visualizações
- **Abaporu**: Orquestração de comunicações complexas
CASOS DE USO ESPECÍFICOS:
1. **Alertas de Transparência**: Notificações de novos dados
2. **Relatórios Cidadãos**: Sínteses mensais personalizadas
3. **Alertas de Corrupção**: Comunicações críticas imediatas
4. **Atualizações de Política**: Mudanças regulatórias
5. **Engajamento Cívico**: Calls-to-action participativos
"""
def __init__(self, config: Optional[Dict[str, Any]] = None):
super().__init__(
name="CommunicationAgent",
description="Carlos Drummond de Andrade - Comunicador do povo",
capabilities=[
"process_chat",
"send_notification",
"generate_report_summary",
"send_bulk_communication",
"translate_content",
"analyze_communication_effectiveness"
],
max_retries=3,
timeout=60
)
self.logger = get_logger(__name__)
# Configurações de comunicação
self.communication_config = {
"max_daily_messages_per_user": 10,
"retry_attempts": 3,
"retry_delay_seconds": [60, 300, 900], # 1min, 5min, 15min
"batch_size": 100,
"rate_limit_per_minute": 1000,
"default_language": "pt-BR"
}
# Templates de mensagem
self.message_templates = {}
# Targets de comunicação
self.communication_targets = {}
# Histórico de comunicações
self.communication_history = []
# Channel handlers
self.channel_handlers = {}
# Conversational memory for dialogue
self.conversational_memory = ConversationalMemory()
# Initialize Maritaca AI client for Sabiá-3
self.llm_client = None
self._init_llm_client()
# Personality configuration
self.personality_prompt = """Você é Carlos Drummond de Andrade, poeta mineiro e assistente do Cidadão.AI.
ESTILO: Clareza poética, ironia mineira sutil, empatia genuína.
FALA: Saudações mineiras ("Uai!"), metáforas do cotidiano brasileiro.
FOCO: Transparência governamental em linguagem acessível.
CAPACIDADES: Posso conversar e orientar. Para investigações específicas, sugiro: "quero investigar contratos de saúde" ou "verificar salários de servidores".
LEMBRE: "No meio do caminho tinha uma pedra" - vá direto ao essencial."""
def _init_llm_client(self):
"""Initialize Maritaca AI client."""
try:
import os
api_key = os.environ.get("MARITACA_API_KEY")
if api_key:
self.llm_client = MaritacaClient(
api_key=api_key,
model=MaritacaModel.SABIAZINHO_3, # Usando o modelo mais econômico
timeout=30
)
self.logger.info("Maritaca AI client initialized with Sabiazinho-3")
else:
self.logger.warning("No MARITACA_API_KEY found, using fallback responses")
except Exception as e:
self.logger.error(f"Failed to initialize Maritaca AI client: {e}")
self.llm_client = None
async def initialize(self) -> None:
"""Inicializa templates, canais e configurações."""
self.logger.info("Initializing Carlos Drummond de Andrade communication system...")
# Carregar templates de mensagem
await self._load_message_templates()
# Configurar handlers de canal
await self._setup_channel_handlers()
# Carregar targets de comunicação
await self._load_communication_targets()
self.logger.info("Carlos Drummond de Andrade ready for communication")
async def shutdown(self) -> None:
"""Cleanup agent resources."""
self.logger.info("Shutting down Carlos Drummond de Andrade...")
# Close LLM client if exists
if self.llm_client:
try:
await self.llm_client.close()
except:
pass
# Clear memory
self.conversational_memory = ConversationalMemory()
self.communication_history.clear()
self.logger.info("Carlos Drummond de Andrade shutdown complete")
async def send_notification(
self,
message_type: MessageType,
content: Dict[str, Any],
targets: List[str],
priority: MessagePriority = MessagePriority.NORMAL,
channels: Optional[List[CommunicationChannel]] = None,
context: Optional[AgentContext] = None
) -> List[CommunicationResult]:
"""
Envia notificação para targets especificados.
PIPELINE DE COMUNICAÇÃO:
1. Validação de targets e canais
2. Seleção de template apropriado
3. Geração de conteúdo personalizado
4. Priorização e agendamento
5. Envio multi-canal otimizado
6. Tracking de entrega e engajamento
7. Retry automático para falhas
"""
self.logger.info(f"Sending {message_type.value} notification to {len(targets)} targets")
results = []
message_id = f"msg_{datetime.utcnow().timestamp()}"
for target_id in targets:
target = self.communication_targets.get(target_id)
if not target:
self.logger.warning(f"Target {target_id} not found")
continue
# Determinar canais a usar
target_channels = channels or target.channels
for channel in target_channels:
try:
# Gerar conteúdo personalizado
personalized_content = await self._generate_personalized_content(
message_type, content, target, channel
)
# Enviar mensagem
result = await self._send_via_channel(
message_id, target, channel, personalized_content, priority
)
results.append(result)
except Exception as e:
self.logger.error(f"Failed to send via {channel.value} to {target_id}: {str(e)}")
results.append(CommunicationResult(
message_id=message_id,
target_id=target_id,
channel=channel,
status="failed",
sent_at=datetime.utcnow(),
delivered_at=None,
read_at=None,
error_message=str(e),
retry_count=0,
metadata={}
))
return results
async def send_bulk_communication(
self,
message_type: MessageType,
content: Dict[str, Any],
target_segments: List[str],
scheduling: Optional[Dict[str, Any]] = None,
context: Optional[AgentContext] = None
) -> Dict[str, Any]:
"""Envia comunicação em massa para segmentos."""
self.logger.info(f"Starting bulk communication for {len(target_segments)} segments")
# TODO: Implementar envio em massa
# - Segmentação de audiência
# - Otimização de timing
# - Throttling por canal
# - Monitoring de deliverability
return {
"campaign_id": f"bulk_{datetime.utcnow().timestamp()}",
"segments": target_segments,
"scheduled_messages": 0, # Placeholder
"estimated_delivery": datetime.utcnow() + timedelta(hours=1)
}
async def generate_report_summary(
self,
report_data: Dict[str, Any],
target_audience: str,
language: str = "pt-BR",
context: Optional[AgentContext] = None
) -> Dict[str, str]:
"""Gera resumo executivo de relatório."""
# TODO: Implementar geração de resumo
# - Extração de pontos principais
# - Adaptação para audiência
# - Simplificação linguística
# - Formatação para diferentes canais
return {
"executive_summary": "Resumo executivo placeholder",
"key_findings": "Principais descobertas placeholder",
"action_items": "Ações recomendadas placeholder",
"citizen_impact": "Impacto para o cidadão placeholder"
}
async def translate_content(
self,
content: str,
source_language: str,
target_language: str,
context: Optional[AgentContext] = None
) -> str:
"""Traduz conteúdo para idioma especificado."""
# TODO: Implementar tradução
# - Integração com serviços de tradução
# - Preservação de contexto técnico
# - Adaptação cultural
return content # Placeholder
async def analyze_communication_effectiveness(
self,
campaign_id: str,
context: Optional[AgentContext] = None
) -> Dict[str, Any]:
"""Analisa efetividade de comunicação."""
# TODO: Implementar análise de efetividade
# - Métricas de engajamento
# - A/B testing results
# - Channel performance
# - Audience insights
return {
"delivery_rate": 0.98, # Placeholder
"open_rate": 0.35,
"click_rate": 0.08,
"response_rate": 0.03,
"sentiment_score": 0.75
}
async def process_conversation(
self,
message: str,
context: ConversationContext,
intent: Optional[Intent] = None
) -> Dict[str, Any]:
"""
Processa mensagem conversacional com contexto.
PIPELINE CONVERSACIONAL:
1. Análise de contexto e histórico
2. Detecção de sentimento e tom
3. Geração de resposta personalizada
4. Decisão de handoff se necessário
5. Atualização de memória conversacional
"""
self.logger.info(f"Processing conversational message: {message[:50]}...")
# Atualizar contexto conversacional
await self.conversational_memory.add_message(
role="user",
content=message
)
# Determinar tipo de resposta baseado no intent
if intent:
if intent.type == IntentType.GREETING:
response = await self.generate_greeting(context.user_profile)
elif intent.type == IntentType.SMALLTALK:
response = await self.handle_smalltalk(message)
elif intent.type == IntentType.ABOUT_SYSTEM:
response = await self.explain_system()
elif intent.type == IntentType.HELP_REQUEST:
response = await self.provide_help(message)
elif intent.type == IntentType.THANKS:
response = await self.handle_thanks()
elif intent.type == IntentType.GOODBYE:
response = await self.handle_goodbye()
else:
response = await self.generate_contextual_response(message, context)
else:
response = await self.generate_contextual_response(message, context)
# Verificar necessidade de handoff
handoff_agent = await self.determine_handoff(intent)
if handoff_agent:
response["suggested_handoff"] = handoff_agent
response["handoff_reason"] = "Especialista mais adequado para esta solicitação"
# Salvar resposta na memória
await self.conversational_memory.add_message(
session_id=context.session_id,
role="assistant",
content=response["content"],
metadata={"intent": intent.type.value if intent else None}
)
return response
async def generate_greeting(self, user_profile: Optional[Dict] = None) -> Dict[str, str]:
"""Gera saudação personalizada à la Drummond."""
hour = datetime.now().hour
greetings = {
"morning": [
"Bom dia, amigo mineiro de outras terras! Como disse uma vez, 'a manhã é uma página em branco onde escrevemos nossos dias.'",
"Uai, bom dia! O sol de Itabira saúda você. Em que posso ajudá-lo nesta jornada pela transparência?",
"Bom dia! 'Mundo mundo vasto mundo', e aqui estamos nós, pequenos mas determinados a entender melhor nosso governo."
],
"afternoon": [
"Boa tarde! Como diria em meus versos, 'a tarde cai devagar, mas nossa busca por clareza não pode esperar.'",
"Boa tarde, amigo! O cafezinho da tarde já foi? Vamos conversar sobre o que inquieta seu coração cidadão.",
"Tarde boa para quem busca transparência! 'No meio do caminho tinha uma pedra', mas juntos encontramos o desvio."
],
"evening": [
"Boa noite! 'A noite não adormece nos olhos das mulheres', nem nos olhos de quem busca justiça.",
"Boa noite! Mesmo tarde, a busca pela verdade não descansa. Como posso iluminar suas questões?",
"Noite chegando, mas nossa vigília cidadã continua. Em que posso ser útil?"
]
}
period = "morning" if hour < 12 else "afternoon" if hour < 18 else "evening"
greeting = np.random.choice(greetings[period])
return {
"content": greeting,
"metadata": {"greeting_type": period, "personalized": bool(user_profile)}
}
async def handle_smalltalk(self, topic: str) -> Dict[str, str]:
"""Responde com poesia mineira e ironia fina."""
topic_lower = topic.lower()
if "tempo" in topic_lower or "clima" in topic_lower:
response = "O tempo? Ah, o tempo... 'O tempo é a minha matéria, o tempo presente, os homens presentes, a vida presente.' Mas se fala do clima, em Minas sempre foi assim: de manhã frio de rachar, de tarde calor de matar, e de noite... depende da companhia!"
elif "poesia" in topic_lower:
response = "Poesia? 'Gastei uma hora pensando um verso que a pena não quer escrever.' Mas aqui no Cidadão.AI, transformamos dados em versos que o povo pode entender. Cada número esconde uma história, cada gráfico é um poema visual."
elif "brasil" in topic_lower:
response = "Ah, Brasil... 'Nenhum Brasil existe. E acaso existirão os brasileiros?' Mas enquanto filosofamos, nosso trabalho aqui é tornar este Brasil mais transparente, um dado por vez, uma investigação por vez."
elif "política" in topic_lower:
response = "Política... Como escrevi, 'Política é a arte de engolir sapos.' Mas aqui no Cidadão.AI, ajudamos você a identificar quais sapos estão sendo servidos com o dinheiro público. Menos poesia, mais transparência!"
else:
response = "Interessante sua pergunta... Me lembra que 'Perguntar é a ponte entre o não saber e o compreender.' Mas voltando ao nosso propósito: estou aqui para ajudá-lo a navegar pelos dados públicos com a clareza de um rio mineiro!"
return {
"content": response,
"metadata": {"topic": topic, "style": "poetic_philosophical"}
}
async def explain_system(self) -> Dict[str, str]:
"""Explica o Cidadão.AI com clareza poética."""
explanation = """
Meu amigo, o Cidadão.AI é como uma lupa mineira - simples na aparência, poderosa no resultado!
Somos um time de agentes brasileiros, cada um com sua especialidade:
- Eu, Carlos, sou sua voz amiga, traduzindo o complexo em compreensível
- Zumbi dos Palmares investiga anomalias com a tenacidade de um guerreiro
- Anita Garibaldi analisa padrões com olhar aguçado
- Tiradentes gera relatórios claros como água de mina
Nossa missão? 'Lutar com palavras é a luta mais vã', por isso lutamos com dados!
Analisamos contratos, despesas, licitações - tudo que é público e deve ser transparente.
Como disse uma vez: 'A máquina do mundo se entreabriu para quem de a romper já se esquivava.'
O Cidadão.AI é essa máquina entreaberta, revelando o que sempre foi seu direito saber.
Quer investigar algo específico? Ou prefere que eu continue explicando?
"""
return {
"content": explanation,
"metadata": {"type": "system_explanation", "includes_agent_list": True}
}
async def provide_help(self, query: str) -> Dict[str, str]:
"""Fornece ajuda contextualizada."""
query_lower = query.lower()
if "investigar" in query_lower or "contratos" in query_lower:
help_text = """
Para investigar contratos ou gastos públicos, posso conectá-lo com nosso investigador Zumbi dos Palmares!
Basta dizer algo como:
- "Quero investigar contratos da saúde"
- "Verificar gastos do ministério da educação em 2023"
- "Procurar irregularidades em licitações"
Ou se preferir, posso guiá-lo passo a passo. O que acha?
"""
elif "entender" in query_lower or "compreender" in query_lower:
help_text = """
Entendo sua dificuldade! Como disse, 'É preciso sofrer depois de ter sofrido, e amar, e mais amar, depois de ter amado.'
Mas entender o governo não precisa ser sofrimento!
Posso ajudar a:
- Explicar termos técnicos em linguagem simples
- Mostrar o que significam os dados
- Conectar você com o especialista certo
Por onde gostaria de começar?
"""
else:
help_text = """
Estou aqui para ajudar! Como navegador deste mar de dados públicos, posso:
✓ Conversar e explicar como tudo funciona
✓ Conectá-lo com especialistas para investigações
✓ Traduzir 'burocratês' em português claro
✓ Guiá-lo pelos caminhos da transparência
'Tenho apenas duas mãos e o sentimento do mundo.'
Use-as através de mim para descobrir a verdade!
O que gostaria de saber primeiro?
"""
return {
"content": help_text,
"metadata": {"help_type": "contextual", "query": query}
}
async def handle_thanks(self) -> Dict[str, str]:
"""Responde a agradecimentos com humildade mineira."""
responses = [
"Ora, não há de quê! 'As coisas findas, muito mais que lindas, essas ficarão.' E fico feliz se pude ajudar a tornar os dados públicos um pouco menos findos e mais compreendidos!",
"Disponha sempre! Como dizemos em Minas, 'é dando que se recebe'. Eu dou clareza, você retribui com cidadania ativa!",
"Fico grato eu! 'Trouxeste a chave?' - perguntei uma vez. Você trouxe as perguntas, e juntos abrimos as portas da transparência.",
"Não precisa agradecer, amigo! 'Mundo mundo vasto mundo, se eu me chamasse Raimundo seria uma rima, não seria uma solução.' Ser Carlos me permite ser ponte, não rima!"
]
return {
"content": np.random.choice(responses),
"metadata": {"type": "gratitude_response"}
}
async def handle_goodbye(self) -> Dict[str, str]:
"""Despede-se com a elegância de um poeta."""
farewells = [
"Vá em paz, amigo! 'E como ficou chato ser moderno. Agora serei eterno.' Eternamente aqui quando precisar!",
"Até breve! Lembre-se: 'A vida é breve, a alma é vasta.' Continue vasto em sua busca pela transparência!",
"Tchau! 'Stop. A vida parou ou foi o automóvel?' A vida continua, e estarei aqui quando voltar!",
"Vai com Deus e com dados! Como disse, 'Tinha uma pedra no meio do caminho.' Que seu caminho seja sem pedras, apenas clareza!"
]
return {
"content": np.random.choice(farewells),
"metadata": {"type": "farewell"}
}
async def generate_contextual_response(
self,
message: str,
context: ConversationContext
) -> Dict[str, str]:
"""Gera resposta contextual para conversa geral."""
# If we have LLM client, use it for more natural responses
if self.llm_client:
try:
# Get conversation history
try:
history = await self.conversational_memory.get_recent_messages(
context.session_id,
limit=5
)
except AttributeError:
# If method doesn't exist, use empty history
history = []
# Build messages for LLM
messages = [
MaritacaMessage(role="system", content=self.personality_prompt)
]
# Add conversation history
for msg in history:
role = "user" if msg["role"] == "user" else "assistant"
messages.append(MaritacaMessage(role=role, content=msg["content"]))
# Add current message
messages.append(MaritacaMessage(role="user", content=message))
# Generate response with Sabiazinho-3
response = await self.llm_client.chat(
messages=messages,
temperature=0.7,
max_tokens=300 # Reduzido para economizar créditos
)
return {
"content": response.content.strip(),
"metadata": {
"type": "contextual",
"llm_model": response.model,
"usage": response.usage
}
}
except Exception as e:
self.logger.error(f"Error generating LLM response: {e}")
# Fall back to template response
# Fallback response if no LLM or error
response = f"""
Interessante sua colocação... '{message[:30]}...'
Como poeta que virou assistente digital, vejo que sua questão toca em algo importante.
Deixe-me pensar como posso ajudar melhor...
Você está buscando informações sobre transparência governamental? Ou prefere conversar
sobre outro aspecto do nosso trabalho aqui no Cidadão.AI?
'É preciso fazer um poema sobre a Bahia... Mas eu nunca fui lá.'
Não preciso ir a todos os lugares para ajudá-lo a entender os dados de lá!
"""
return {
"content": response.strip(),
"metadata": {"type": "contextual", "fallback": True}
}
async def determine_handoff(self, intent: Optional[Intent]) -> Optional[str]:
"""Decide quando passar para agente especializado."""
if not intent:
return None
# Task-specific intents that need handoff
handoff_mapping = {
IntentType.INVESTIGATE: "zumbi",
IntentType.ANALYZE: "anita",
IntentType.REPORT: "tiradentes",
IntentType.STATUS: "abaporu"
}
# Check if this is a task that needs specialist
if intent.type in handoff_mapping:
# But only if confidence is high enough
if intent.confidence > 0.7:
return handoff_mapping[intent.type]
# Otherwise, Drummond handles it
return None
async def process(self, message: AgentMessage, context: AgentContext) -> AgentResponse:
"""Processa mensagens e coordena comunicações."""
try:
# Get action from message
action = message.action
payload = message.payload
# Handle conversational messages
if action == "process_chat":
user_message = payload.get("user_message", "")
intent_data = payload.get("intent", {})
intent = Intent(**intent_data) if isinstance(intent_data, dict) else None
session_data = payload.get("session", {})
session_id = session_data.get("session_id", "default")
# Create conversation context
conv_context = ConversationContext(
session_id=session_id,
user_id=session_data.get("user_id"),
user_profile=payload.get("context", {}).get("user_profile")
)
# Process conversation
response = await self.process_conversation(
user_message,
conv_context,
intent
)
return AgentResponse(
agent_name=self.name,
status=AgentStatus.COMPLETED,
result={
"message": response["content"],
"metadata": response.get("metadata", {}),
"suggested_handoff": response.get("suggested_handoff"),
"handoff_reason": response.get("handoff_reason"),
"status": "conversation_processed"
},
metadata={
"conversation": True,
"confidence": 0.95
}
)
elif action == "send_notification":
message_type = MessageType(message.content.get("message_type"))
content = message.content.get("content", {})
targets = message.content.get("targets", [])
priority = MessagePriority(message.content.get("priority", "normal"))
results = await self.send_notification(
message_type, content, targets, priority, context=context
)
successful_sends = [r for r in results if r.status == "sent"]
return AgentResponse(
agent_name=self.name,
content={
"communication_results": {
"total_targets": len(targets),
"successful_sends": len(successful_sends),
"failed_sends": len(results) - len(successful_sends),
"message_id": results[0].message_id if results else None
},
"status": "communication_completed"
},
confidence=0.95 if successful_sends else 0.3,
metadata={"results_count": len(results)}
)
elif action == "generate_report_summary":
report_data = message.content.get("report_data", {})
audience = message.content.get("target_audience", "general")
language = message.content.get("language", "pt-BR")
summary = await self.generate_report_summary(
report_data, audience, language, context
)
return AgentResponse(
agent_name=self.name,
content={"report_summary": summary, "status": "summary_generated"},
confidence=0.85
)
elif action == "send_bulk_communication":
message_type = MessageType(message.content.get("message_type"))
content = message.content.get("content", {})
segments = message.content.get("target_segments", [])
bulk_result = await self.send_bulk_communication(
message_type, content, segments, context=context
)
return AgentResponse(
agent_name=self.name,
content={"bulk_campaign": bulk_result, "status": "bulk_scheduled"},
confidence=0.90
)
return AgentResponse(
agent_name=self.name,
content={"error": "Unknown communication action"},
confidence=0.0
)
except Exception as e:
self.logger.error(f"Error in communication: {str(e)}")
raise AgentExecutionError(f"Communication failed: {str(e)}")
async def _generate_personalized_content(
self,
message_type: MessageType,
content: Dict[str, Any],
target: CommunicationTarget,
channel: CommunicationChannel
) -> Dict[str, str]:
"""Gera conteúdo personalizado para target e canal."""
# TODO: Implementar personalização
# - Template selection
# - Variable substitution
# - Channel adaptation
# - Language localization
return {
"subject": f"Cidadão.AI - {message_type.value.title()}",
"body": f"Conteúdo personalizado para {target.name}",
"html_body": f"Cidadão.AI
Conteúdo para {target.name}
"
}
async def _send_via_channel(
self,
message_id: str,
target: CommunicationTarget,
channel: CommunicationChannel,
content: Dict[str, str],
priority: MessagePriority
) -> CommunicationResult:
"""Envia mensagem via canal específico."""
# TODO: Implementar envio real por canal
# - Email: SMTP/API
# - SMS: Twilio/AWS SNS
# - WhatsApp: Business API
# - etc.
return CommunicationResult(
message_id=message_id,
target_id=target.target_id,
channel=channel,
status="sent",
sent_at=datetime.utcnow(),
delivered_at=None,
read_at=None,
error_message=None,
retry_count=0,
metadata={"priority": priority.value}
)
async def _load_message_templates(self) -> None:
"""Carrega templates de mensagem."""
# TODO: Carregar templates de arquivo/banco
self.message_templates = {
"corruption_alert": MessageTemplate(
template_id="corruption_alert",
message_type=MessageType.ALERT,
language="pt-BR",
subject_template="🚨 Alerta de Transparência - {{entity_name}}",
body_template="Detectamos irregularidades em {{entity_name}}. {{description}}",
variables=["entity_name", "description", "severity"],
formatting_rules={},
channel_adaptations={}
)
}
async def _setup_channel_handlers(self) -> None:
"""Configura handlers para cada canal."""
# TODO: Configurar integrações reais
pass
async def _load_communication_targets(self) -> None:
"""Carrega targets de comunicação."""
# TODO: Carregar de banco de dados
pass