# type: ignore """ ================================================================================ AKIRA V21 ULTIMATE - REPLY CONTEXT HANDLER MODULE ================================================================================ Sistema dedicado para processar e priorizar contexto de replies. Garante que replies tenham prioridade ligeiramente maior que o contexto geral, especialmente em perguntas curtas. Features: - Extração e processamento de metadados de reply - 3 níveis de prioridade (1=normal, 2=reply, 3=reply-to-bot+pergunta-curta) - Construção de prompt sections otimizadas para replies - Integração com ShortTermMemory - Context hint extraction para melhor compreensão ================================================================================ """ import os import sys import time import json import re import logging from typing import Optional, Dict, Any, List, Tuple from dataclasses import dataclass, field # Imports robustos com fallback - CORRIGIDO para usar modules. try: import modules.config as config from .short_term_memory import ShortTermMemory, MessageWithContext, IMPORTANCIA_REPLY, IMPORTANCIA_REPLY_TO_BOT, IMPORTANCIA_PERGUNTA_CURTA_REPLY REPLY_HANDLER_AVAILABLE = True except ImportError: try: from . import config from .short_term_memory import ShortTermMemory, MessageWithContext REPLY_HANDLER_AVAILABLE = True except ImportError: REPLY_HANDLER_AVAILABLE = False config = None logger = logging.getLogger(__name__) # ============================================================ # NÍVEIS DE PRIORIDADE # ============================================================ PRIORITY_NORMAL = 1 PRIORITY_REPLY = 2 PRIORITY_REPLY_TO_BOT = 3 PRIORITY_REPLY_TO_BOT_SHORT_QUESTION = 4 # Prioridade máxima! # Limite de palavras para "pergunta curta" PERGUNTA_CURTA_LIMITE: int = 5 @dataclass class ProcessedReplyContext: """ Contexto de reply processado e pronto para uso. Attributes: is_reply: Se é um reply reply_to_bot: Se é reply direcionado ao bot priority_level: Nível de prioridade (1-4) quoted_author_name: Nome do autor da mensagem citada quoted_author_numero: Número do autor quoted_text_original: Texto original citado mensagem_citada: Texto da mensagem citada context_hint: Hint de contexto extraído importancia: Peso de importância calculado prompt_section: Section formatada para o prompt should_prioritize_reply: Se deve priorizar no prompt adaptive_multiplier: Multiplicador adaptativo baseado no tamanho """ is_reply: bool = False reply_to_bot: bool = False priority_level: int = PRIORITY_NORMAL quoted_author_name: str = "" quoted_author_numero: str = "" quoted_text_original: str = "" mensagem_citada: str = "" context_hint: str = "" importancia: float = 1.0 prompt_section: str = "" should_prioritize_reply: bool = False adaptive_multiplier: float = 1.0 def to_dict(self) -> Dict[str, Any]: """Converte para dicionário.""" return { "is_reply": self.is_reply, "reply_to_bot": self.reply_to_bot, "priority_level": self.priority_level, "quoted_author_name": self.quoted_author_name, "quoted_author_numero": self.quoted_author_numero, "quoted_text_original": self.quoted_text_original, "mensagem_citada": self.mensagem_citada, "context_hint": self.context_hint, "importancia": self.importancia, "prompt_section": self.prompt_section, "should_prioritize_reply": self.should_prioritize_reply, "adaptive_multiplier": self.adaptive_multiplier } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'ProcessedReplyContext': """Cria instância a partir de dicionário.""" return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__}) # ============================================================ # FUNÇÕES AUXILIARES # ============================================================ def contar_palavras(texto: str) -> int: """Conta palavras em um texto.""" if not texto: return 0 return len(texto.split()) def is_pergunta_curta(texto: str) -> bool: """ Verifica se o texto é uma pergunta curta. Args: texto: Texto a verificar Returns: True se for pergunta com pocas palavras """ if not texto: return False texto_lower = texto.strip().lower() word_count = contar_palavras(texto) # Deve ter marcador de pergunta ou palavras interrogativas has_question_marker = '?' in texto has_interrogative = any(w in texto_lower for w in [ 'qual', 'quais', 'quem', 'como', 'onde', 'quando', 'por que', 'porque', 'para que', 'o que', 'que', 'é o que', 'vc', 'você', 'tu', 'meu', 'minha', 'oq', 'oq', 'n' ]) return word_count <= PERGUNTA_CURTA_LIMITE and (has_question_marker or has_interrogative) def extrair_context_hint(quoted_text: str, mensagem_atual: str) -> str: """ Extrai hint de contexto baseado no texto citado e mensagem atual. Args: quoted_text: Texto original citado mensagem_atual: Mensagem atual do usuário Returns: String de hint de contexto """ hints = [] # Detecta tipo de reply quoted_lower = quoted_text.lower() if quoted_text else "" # Pergunta sobre o bot if any(w in quoted_lower for w in ['akira', 'bot', 'você', 'vc', 'tu']): hints.append("pergunta_sobre_akira") # Pergunta factual if any(w in quoted_lower for w in ['oq', 'o que', 'qual', 'quanto', 'onde', 'quando']): hints.append("pergunta_factual") # Ironia/deboche detectado if any(w in quoted_lower for w in ['kkk', 'haha', '😂', '🤣', 'eita']): hints.append("tom_irreverente") # Expressão de opinião if any(w in quoted_lower for w in ['acho', 'penso', 'creio', 'imagino']): hints.append("expressao_opiniao") return " | ".join(hints) if hints else "contexto_geral" def calcular_prioridade( is_reply: bool, reply_to_bot: bool, mensagem: str, quoted_text: str = "" ) -> Tuple[int, float]: """ Calcula nível de prioridade e importância. Args: is_reply: Se é um reply reply_to_bot: Se é reply para o bot mensagem: Mensagem atual quoted_text: Texto citado Returns: Tupla (priority_level, importancia) """ if not is_reply: return PRIORITY_NORMAL, 1.0 # Reply para o bot if reply_to_bot: # Pergunta curta = prioridade máxima if is_pergunta_curta(mensagem): return PRIORITY_REPLY_TO_BOT_SHORT_QUESTION, IMPORTANCIA_PERGUNTA_CURTA_REPLY # Reply normal ao bot return PRIORITY_REPLY_TO_BOT, IMPORTANCIA_REPLY_TO_BOT # Reply para outro usuário return PRIORITY_REPLY, IMPORTANCIA_REPLY # ============================================================ # CLASSE PRINCIPAL # ============================================================ class ReplyContextHandler: """ Handler dedicado para processar e priorizar contexto de replies. Funcionalidades: - Extração de metadados de reply do payload - Cálculo automático de prioridade - Construção de seções de prompt otimizadas - Integração com ShortTermMemory - Ajuste adaptativo baseado em tamanho da pergunta """ def __init__(self, short_term_memory: Optional[ShortTermMemory] = None): """ Inicializa o handler. Args: short_term_memory: Instância de ShortTermMemory (opcional) """ self.short_term_memory = short_term_memory logger.debug("✅ ReplyContextHandler inicializado") def process_reply( self, mensagem: str, reply_metadata: Dict[str, Any], historico_geral: Optional[List[Dict[str, Any]]] = None ) -> ProcessedReplyContext: """ Processa metadados de reply e gera contexto processado. Args: mensagem: Mensagem atual do usuário reply_metadata: Metadados do reply do payload historico_geral: Histórico geral (opcional) Returns: ProcessedReplyContext pronto para uso """ # Extrai dados do metadata is_reply = reply_metadata.get('is_reply', False) reply_to_bot = reply_metadata.get('reply_to_bot', False) quoted_author_name = reply_metadata.get('quoted_author_name', '') quoted_author_numero = reply_metadata.get('quoted_author_numero', '') quoted_text_original = reply_metadata.get('quoted_text_original', '') mensagem_citada = reply_metadata.get('mensagem_citada', '') or quoted_text_original # 🔧 CORREÇÃO: Se autor é desconhecido, tenta detectar pelo contexto if not quoted_author_name or quoted_author_name.lower() in ['desconhecido', 'unknown', '']: # Detecta pelo conteúdo da mensagem citada quoted_lower = quoted_text_original.lower() if quoted_text_original else "" # Se a mensagem citada contém padrões de resposta do bot bot_patterns = ['akira:', 'eu sou', 'eu sou a akira', 'sou um bot', 'oi!', 'eae!'] if any(p in quoted_lower for p in bot_patterns): quoted_author_name = "Akira (você mesmo)" quoted_author_numero = "BOT" reply_to_bot = True elif mensagem_citada: # Se há histórico, busca última mensagem if historico_geral: # Assumir que é reply para a última mensagem do bot quoted_author_name = "mensagem_anterior" quoted_author_numero = "unknown" # Se ainda não tem autor mas tem mensagem citada e é reply if is_reply and (not quoted_author_name or quoted_author_name == 'desconhecido'): # Se é reply_to_bot=True mas autor desconhecido, assume que é reply para o bot if reply_to_bot: quoted_author_name = "Akira (você mesmo)" quoted_author_numero = "BOT" else: # Tenta extrair do conteúdo quoted_author_name = "participante_desconhecido" # Calcula prioridade e importância priority_level, importancia = calcular_prioridade( is_reply=is_reply, reply_to_bot=reply_to_bot, mensagem=mensagem, quoted_text=quoted_text_original ) # Extrai context hint context_hint = extrair_context_hint(quoted_text_original, mensagem) # Calcula multiplicador adaptativo adaptive_multiplier = self._calculate_adaptive_multiplier( mensagem=mensagem, is_reply=is_reply, priority_level=priority_level ) # Determina se deve priorizar no prompt should_prioritize = is_reply and priority_level >= PRIORITY_REPLY # Constrói section do prompt prompt_section = self._build_reply_prompt_section( mensagem=mensagem, mensagem_citada=mensagem_citada, quoted_author_name=quoted_author_name, reply_to_bot=reply_to_bot, context_hint=context_hint, priority_level=priority_level ) # Cria contexto processado reply_context = ProcessedReplyContext( is_reply=is_reply, reply_to_bot=reply_to_bot, priority_level=priority_level, quoted_author_name=quoted_author_name, quoted_author_numero=quoted_author_numero, quoted_text_original=quoted_text_original, mensagem_citada=mensagem_citada, context_hint=context_hint, importancia=importancia * adaptive_multiplier, prompt_section=prompt_section, should_prioritize_reply=should_prioritize, adaptive_multiplier=adaptive_multiplier ) # Adiciona à memória de curto prazo se disponível if self.short_term_memory and is_reply: self.short_term_memory.add_message( role="user", content=mensagem, importancia=reply_context.importancia, reply_info={ "is_reply": True, "reply_to_bot": reply_to_bot, "quoted_text_original": quoted_text_original, "priority_level": priority_level } ) return reply_context def _calculate_adaptive_multiplier( self, mensagem: str, is_reply: bool, priority_level: int ) -> float: """ Calcula multiplicador adaptativo baseado no tamanho da pergunta. Para perguntas curtas com reply, aumenta a importância do contexto do reply para garantir que o LLM tenha contexto suficiente. Args: mensagem: Mensagem atual is_reply: Se é reply priority_level: Nível de prioridade Returns: Multiplicador entre 1.0 e 2.0 """ if not is_reply: return 1.0 word_count = contar_palavras(mensagem) # Pergunta muito curta (< 3 palavras) = contexto crítico if word_count <= 2: return 1.5 # Pergunta curta (3-5 palavras) = contexto importante if word_count <= PERGUNTA_CURTA_LIMITE: return 1.3 # Pergunta normal = multiplicador padrão baseado em prioridade if priority_level == PRIORITY_REPLY_TO_BOT_SHORT_QUESTION: return 1.2 elif priority_level == PRIORITY_REPLY_TO_BOT: return 1.1 return 1.0 def _build_reply_prompt_section( self, mensagem: str, mensagem_citada: str, quoted_author_name: str, reply_to_bot: bool, context_hint: str, priority_level: int ) -> str: """ Constrói seção formatada do prompt para replies. Args: mensagem: Mensagem atual mensagem_citada: Texto citado quoted_author_name: Nome do autor reply_to_bot: Se é reply para o bot context_hint: Hint de contexto priority_level: Nível de prioridade Returns: String formatada para inserção no prompt """ if not mensagem_citada: return "" sections = [] # Cabeçalho com nível de prioridade if priority_level >= PRIORITY_REPLY_TO_BOT_SHORT_QUESTION: sections.append("[🔴 REPLY CRÍTICO - PERGUNTA CURTA]") elif priority_level == PRIORITY_REPLY_TO_BOT: sections.append("[🟡 REPLY AO BOT]") elif priority_level == PRIORITY_REPLY: sections.append("[🟢 REPLY]") # Contexto do autor if reply_to_bot: sections.append(f"⚠️ VOCÊ ESTÁ SENDO DIRETAMENTE RESPONDIDO!") else: sections.append(f"Respondendo a: {quoted_author_name}") # Texto citado quoted_preview = mensagem_citada[:150] + ("..." if len(mensagem_citada) > 150 else "") sections.append(f"Msg citada: \"{quoted_preview}\"") # Hint de contexto if context_hint and context_hint != "contexto_geral": sections.append(f"Contexto: {context_hint}") # Instrução de resposta if priority_level >= PRIORITY_REPLY_TO_BOT_SHORT_QUESTION: sections.append("💡 RESPONSE: Contextualize sua resposta usando a mensagem citada!") elif reply_to_bot: sections.append("💡 RESPONSE: Você foi diretamente mencionado.") return "\n".join(sections) def prioritize_reply_context( self, prompt: str, reply_context: ProcessedReplyContext, historico_geral: Optional[List[Dict[str, Any]]] = None ) -> str: """ Injeta contexto de reply no prompt com alta prioridade. Args: prompt: Prompt original reply_context: Contexto de reply processado historico_geral: Histórico geral (opcional) Returns: Prompt enriquecido com contexto de reply """ if not reply_context.is_reply or not reply_context.prompt_section: return prompt # Insere contexto de reply no início do prompt reply_block = f""" {'='*60} {reply_context.prompt_section} {'='*60} """ # Determina posição de inserção # Se há seção [SYSTEM], insere após ela if "[SYSTEM]" in prompt: # Encontra final da seção SYSTEM system_end = prompt.find("[/SYSTEM]") if system_end != -1: return prompt[:system_end + 10] + reply_block + prompt[system_end + 10:] # Caso contrário, insere no início return reply_block + "\n" + prompt def get_reply_summary_for_llm(self, reply_context: ProcessedReplyContext) -> str: """ Retorna resumo formatado do reply para contexto do LLM. Args: reply_context: Contexto de reply processado Returns: String resumida para uso no contexto """ if not reply_context.is_reply: return "" parts = [] if reply_context.reply_to_bot: parts.append("REPLY DIRETO AO BOT") else: parts.append(f"REPLY a {reply_context.quoted_author_name}") if reply_context.mensagem_citada: cited = reply_context.mensagem_citada[:100] parts.append(f"Citando: \"{cited}\"") if reply_context.priority_level >= PRIORITY_REPLY_TO_BOT_SHORT_QUESTION: parts.append("PERGUNTA CURTA - Prioridade Alta") return " | ".join(parts) def merge_reply_into_history( self, reply_context: ProcessedReplyContext, history: List[Dict[str, str]] ) -> List[Dict[str, str]]: """ Mescla contexto de reply no histórico para o LLM. Args: reply_context: Contexto de reply processado history: Histórico formatado para LLM Returns: Histórico com reply injetado no início """ if not reply_context.is_reply: return history # Cria entry para o reply reply_entry = { "role": "user", "content": f"[REPLY] {reply_context.get_reply_summary_for_llm(reply_context)}" } # Adiciona texto citado se disponível if reply_context.mensagem_citada: reply_entry["content"] += f"\n\nMensagem citada:\n{reply_context.mensagem_citada}" # Insere no início do histórico return [reply_entry] + history def calculate_token_budget( self, reply_context: ProcessedReplyContext, total_budget: int = 8000 ) -> Tuple[int, int]: """ Calcula alocação de tokens entre reply e contexto geral. Args: reply_context: Contexto de reply total_budget: Total de tokens disponíveis Returns: Tupla (tokens_para_reply, tokens_para_contexto) """ if not reply_context.is_reply: return 0, total_budget # Pergunta curta com reply = mais tokens para reply if reply_context.priority_level >= PRIORITY_REPLY_TO_BOT_SHORT_QUESTION: reply_tokens = min(1500, int(total_budget * 0.25)) elif reply_context.reply_to_bot: reply_tokens = min(1000, int(total_budget * 0.15)) else: reply_tokens = min(800, int(total_budget * 0.10)) return reply_tokens, total_budget - reply_tokens # ============================================================ # HELPERS PARA API # ============================================================ @staticmethod def extract_reply_metadata_from_request(data: Dict[str, Any]) -> Dict[str, Any]: """ Extrai metadados de reply de um request da API. Args: data: Payload do request Returns: Dict com metadados de reply """ reply_metadata = data.get('reply_metadata', {}) # Se não há reply_metadata, tenta extrair de campos individuais if not reply_metadata: mensagem_citada = data.get('mensagem_citada', '') if mensagem_citada: reply_metadata = { 'is_reply': True, 'quoted_text_original': mensagem_citada, 'mensagem_citada': mensagem_citada } else: return {'is_reply': False} # Garante campos obrigatórios return { 'is_reply': reply_metadata.get('is_reply', False), 'reply_to_bot': reply_metadata.get('reply_to_bot', False), 'quoted_author_name': reply_metadata.get('quoted_author_name', ''), 'quoted_author_numero': reply_metadata.get('quoted_author_numero', ''), 'quoted_type': reply_metadata.get('quoted_type', 'texto'), 'quoted_text_original': reply_metadata.get('quoted_text_original', ''), 'context_hint': reply_metadata.get('context_hint', ''), 'mensagem_citada': reply_metadata.get('mensagem_citada', '') } def validate_reply_priority(self, reply_context: ProcessedReplyContext) -> bool: """ Valida se a prioridade calculada está correta. Args: reply_context: Contexto a validar Returns: True se válido """ if not reply_context.is_reply: return reply_context.priority_level == PRIORITY_NORMAL # Reply para bot + pergunta curta deve ter prioridade máxima if reply_context.reply_to_bot and is_pergunta_curta(reply_context.mensagem_citada): return reply_context.priority_level == PRIORITY_REPLY_TO_BOT_SHORT_QUESTION # Reply para bot deve ter alta prioridade if reply_context.reply_to_bot: return reply_context.priority_level >= PRIORITY_REPLY_TO_BOT # Reply normal deve ter prioridade >= 2 return reply_context.priority_level >= PRIORITY_REPLY def __repr__(self) -> str: """Representação textual.""" mem_status = "com STM" if self.short_term_memory else "sem STM" return f"ReplyContextHandler({mem_status})" # ============================================================ # FUNÇÕES DE FÁBRICA # ============================================================ def criar_reply_handler( short_term_memory: Optional[ShortTermMemory] = None ) -> ReplyContextHandler: """ Factory function para criar ReplyContextHandler. Args: short_term_memory: Instância de ShortTermMemory (opcional) Returns: ReplyContextHandler instance """ return ReplyContextHandler(short_term_memory=short_term_memory) def processar_reply_request( mensagem: str, request_data: Dict[str, Any], short_term_memory: Optional[ShortTermMemory] = None ) -> ProcessedReplyContext: """ Função helper para processar reply de request. Args: mensagem: Mensagem atual request_data: Payload do request short_term_memory: Instância de ShortTermMemory (opcional) Returns: ProcessedReplyContext """ handler = criar_reply_handler(short_term_memory) reply_metadata = handler.extract_reply_metadata_from_request(request_data) return handler.process_reply(mensagem, reply_metadata) # type: ignore