# type: ignore """ ================================================================================ AKIRA V21 ULTIMATE - UNIFIED CONTEXT MODULE ================================================================================ Sistema unificado que integra Reply Context + Short-Term Memory em sintonia. Philosophy: "Reply context e STM devem trabalhar em sintonia como tik e tack - um fornece o contexto imediato/urgente (o que o usuário está respondendo), o outro fornece o fluxo da conversa (contexto geral)." Features: - Integração seamless entre reply context e STM - Token budgeting inteligente entre os dois contextos - Priorização dinâmica baseada no tipo de mensagem - Suporte a perguntas curtas com reply (prioridade máxima) - Persistência e restauração de contexto unificado ================================================================================ """ import os import sys import time import json import logging from typing import Optional, Dict, Any, List, Tuple from dataclasses import dataclass, field from datetime import datetime # Imports robustos com fallback try: import modules.config as config from .short_term_memory import ( ShortTermMemory, MessageWithContext, IMPORTANCIA_NORMAL, IMPORTANCIA_REPLY, IMPORTANCIA_REPLY_TO_BOT, IMPORTANCIA_PERGUNTA_CURTA_REPLY, estimar_tokens, is_pergunta_curta ) from .reply_context_handler import ( ReplyContextHandler, ProcessedReplyContext, PRIORITY_REPLY, PRIORITY_REPLY_TO_BOT, PRIORITY_REPLY_TO_BOT_SHORT_QUESTION ) UNIFIED_CONTEXT_AVAILABLE = True except ImportError as e: UNIFIED_CONTEXT_AVAILABLE = False config = None logger = logging.getLogger(__name__) # ============================================================ # CONFIGURAÇÃO DE TOKEN BUDGET # ============================================================ @dataclass class ContextTokenBudget: """ Alocação de tokens entre reply context e STM. Philosophy: Reply tem orçamento dedicado (urgente), STM tem o resto (fluxo). """ total_budget: int = 8000 system_tokens: int = 1500 user_message_tokens: int = 500 # Reply context budget (URGENTE) reply_tokens: int = 300 reply_priority_multiplier: float = 1.0 # STM budget (FLUXO DA CONVERSA) stm_tokens: int = 4000 # Reservado para resposta response_reserved: int = 1200 def calculate(self, is_reply: bool, reply_priority: int = 1) -> 'ContextTokenBudget': """ Calcula orçamento baseado no tipo de mensagem. Args: is_reply: Se é um reply reply_priority: Nível de prioridade do reply (1-4) Returns: ContextTokenBudget ajustado """ budget = ContextTokenBudget( total_budget=self.total_budget, system_tokens=self.system_tokens, user_message_tokens=self.user_message_tokens ) if is_reply: if reply_priority >= PRIORITY_REPLY_TO_BOT_SHORT_QUESTION: # Pergunta curta com reply ao bot = prioridade máxima budget.reply_tokens = min(1500, int(self.total_budget * 0.20)) budget.reply_priority_multiplier = 1.5 budget.stm_tokens = min(3500, int(self.total_budget * 0.45)) elif reply_priority >= PRIORITY_REPLY_TO_BOT: # Reply ao bot budget.reply_tokens = min(1200, int(self.total_budget * 0.15)) budget.reply_priority_multiplier = 1.3 budget.stm_tokens = min(4000, int(self.total_budget * 0.50)) elif reply_priority >= PRIORITY_REPLY: # Reply normal budget.reply_tokens = min(800, int(self.total_budget * 0.10)) budget.reply_priority_multiplier = 1.1 budget.stm_tokens = min(4500, int(self.total_budget * 0.55)) else: # Mensagem normal = STM tem orçamento completo budget.reply_tokens = 0 budget.stm_tokens = min(5000, int(self.total_budget * 0.65)) # Calcula response reserved budget.response_reserved = ( budget.total_budget - budget.system_tokens - budget.user_message_tokens - budget.reply_tokens - budget.stm_tokens ) return budget def to_dict(self) -> Dict[str, Any]: """Serializa para dicionário.""" return { "total_budget": self.total_budget, "system_tokens": self.system_tokens, "user_message_tokens": self.user_message_tokens, "reply_tokens": self.reply_tokens, "stm_tokens": self.stm_tokens, "response_reserved": self.response_reserved, "reply_priority_multiplier": self.reply_priority_multiplier } # ============================================================ # CONTEXTO UNIFICADO # ============================================================ @dataclass class UnifiedMessageContext: """ Contexto unificado combinando reply + STM. Philosophy: Reply context (tik) + STM (tok) trabalhando em sintonia. Attributes: - Reply context: Contexto imediato/urgente do reply - STM context: Contexto do fluxo da conversa - Integration: Como os dois são combinados """ # Identificação conversation_id: str = "" user_id: str = "" timestamp: float = field(default_factory=time.time) # Reply Context (TIK - urgente/imediato) is_reply: bool = False reply_to_bot: bool = False reply_priority: int = 1 # 1=normal, 2=reply, 3=reply_to_bot, 4=critical quoted_author: str = "" quoted_content: str = "" reply_importancia: float = 1.0 # STM Context (TOK - fluxo da conversa) stm_messages: List[MessageWithContext] = field(default_factory=list) stm_summary: Dict[str, Any] = field(default_factory=dict) stm_emotional_trend: str = "neutral" # Long-Term Memory (RAG) long_term_memory: str = "" # Integração sync_mode: str = "tiktok" # "tiktok" = reply priority + STM flow token_budget: ContextTokenBudget = field(default_factory=ContextTokenBudget) # Mensagem atual current_message: str = "" current_emotion: str = "neutral" system_override: str = "" def to_dict(self) -> Dict[str, Any]: """Serializa para dicionário.""" return { "conversation_id": self.conversation_id, "user_id": self.user_id, "timestamp": self.timestamp, "is_reply": self.is_reply, "reply_to_bot": self.reply_to_bot, "reply_priority": self.reply_priority, "quoted_author": self.quoted_author, "quoted_content": self.quoted_content[:500] if self.quoted_content else "", "reply_importancia": self.reply_importancia, "stm_messages_count": len(self.stm_messages), "stm_summary": self.stm_summary, "stm_emotional_trend": self.stm_emotional_trend, "long_term_memory": self.long_term_memory, "sync_mode": self.sync_mode, "token_budget": self.token_budget.to_dict(), "current_message": self.current_message[:100], "current_emotion": self.current_emotion } def build_prompt(self) -> str: """ Constrói prompt formatado para o LLM. Returns: String formatada com contexto unificado (reply + STM) """ return format_unified_context_for_llm(self, self.token_budget) # ==================================== # HELPER FUNCTIONS # ==================================== def sync_reply_with_stm( reply_context: Dict[str, Any], stm_messages: List[MessageWithContext], max_stm_messages: int = 10 ) -> List[MessageWithContext]: """ Sincroniza reply context com mensagens STM. Philosophy: Reply (tik) vem primeiro, STM (tok) vem depois. Ambos são combinados para formar o contexto completo. Args: reply_context: Contexto do reply stm_messages: Mensagens da memória de curto prazo max_stm_messages: Máximo de mensagens STM a incluir Returns: Lista combinada de mensagens para contexto """ combined = [] # 1. Adiciona reply context como mensagem mais recente (TIK) if reply_context.get('is_reply', False): reply_msg = MessageWithContext( role="user", content=reply_context.get('quoted_content', ''), importancia=reply_context.get('importancia', IMPORTANCIA_NORMAL), emocao=reply_context.get('emocao', 'neutral'), reply_info={ 'is_reply': True, 'reply_to_bot': reply_context.get('reply_to_bot', False), 'quoted_text_original': reply_context.get('quoted_content', ''), 'priority_level': reply_context.get('priority', 1), 'sync_mode': 'tiktok' } ) combined.append(reply_msg) # 2. Adiciona mensagens STM (TOK - fluxo da conversa) # Pega últimas N mensagens STM stm_to_add = stm_messages[-max_stm_messages:] if stm_messages else [] for msg in stm_to_add: # Se a mensagem STM já é um reply, preserva info if msg.is_reply and not msg.reply_info.get('sync_mode'): msg.reply_info['sync_mode'] = 'stm' combined.append(msg) return combined def format_unified_context_for_llm( unified: UnifiedMessageContext, budget: ContextTokenBudget ) -> str: """ Formata contexto unificado para o prompt do LLM. Philosophy: Reply (tik) primeiro por ser urgente, STM (tok) depois para contexto da conversa. Args: unified: Contexto unificado budget: Orçamento de tokens Returns: String formatada para o prompt """ parts = [] # ===== 1. REPLY CONTEXT (TIK - URGENTE) ===== if unified.is_reply: reply_section = [] reply_section.append("=" * 50) reply_section.append("[📎 REPLY CONTEXT - PRIORITÁRIO]") reply_section.append("=" * 50) if unified.reply_to_bot: reply_section.append("⚠️ VOCÊ ESTÁ SENDO DIRETAMENTE RESPONDIDO!") else: reply_section.append(f"Respondendo a: {unified.quoted_author}") # Conteúdo citado if unified.quoted_content: quoted_preview = unified.quoted_content[:budget.reply_tokens // 4] reply_section.append(f"\n\n{quoted_preview}...\n") # Prioridade if unified.reply_priority >= PRIORITY_REPLY_TO_BOT_SHORT_QUESTION: reply_section.append("\n💡 PERGUNTA CURTA + REPLY: FOCO NA CITAÇÃO") reply_section.append("\n📌 INSTRUÇÕES DE REPLY:") reply_section.append("- Relacione o input atual ESTRITAMENTE ao .") reply_section.append("- PRESERVE a sua identidade e humor (seja o Akira, natural e irreverente).") reply_section.append("- Não assuma detalhes inexistentes, use o fluxo (STM) para coerência base.") parts.append("\n".join(reply_section)) # ===== RAG CONTEXT (MEMÓRIA DE LONGO PRAZO) ===== if unified.long_term_memory: rag_section = [] rag_section.append("\n" + "=" * 50) rag_section.append("[📖 MEMÓRIA DE LONGO PRAZO (BANCO DE DADOS)]") rag_section.append("=" * 50) rag_section.append("(Informações previamente aprendidas sobre o usuário)") rag_section.append(unified.long_term_memory) parts.append("\n".join(rag_section)) # ===== 2. STM CONTEXT (TOK - FLUXO DA CONVERSA) ===== if unified.stm_messages: stm_section = [] stm_section.append("\n" + "=" * 50) stm_section.append("[🧠 MEMÓRIA DE CURTO PRAZO - FLUXO DA CONVERSA]") stm_section.append("=" * 50) stm_section.append("(conversa recente para contexto)") # emotional trend if unified.stm_emotional_trend != "neutral": stm_section.append(f"\n📊 Tendência emocional: {unified.stm_emotional_trend}") # Formata mensagens STM stm_tokens_used = 0 for msg in unified.stm_messages: # Formata role role_icon = "👤" if msg.role == "user" else "🤖" role_label = "USER" if msg.role == "user" else "AKIRA" # Se é reply, marca reply_marker = " [REPLY]" if msg.is_reply else "" # Preview do conteúdo content_preview = msg.content[:100] msg_line = f"{role_icon} [{role_label}]{reply_marker}: {content_preview}..." msg_tokens = estimar_tokens(msg_line) if stm_tokens_used + msg_tokens <= budget.stm_tokens: stm_section.append(msg_line) stm_tokens_used += msg_tokens stm_section.append("\n💡 INTEGRAÇÃO: Use este contexto para manter coerência!") parts.append("\n".join(stm_section)) return "\n".join(parts) # ==================================== # SHORT-TERM MEMORY MANAGER # ==================================== class ShortTermMemoryManager: """ Gerenciador de instâncias STM por conversa. Philosophy: Cada conversa tem sua própria STM isolada, mas todas compartilham o mesmo manager. """ _instance = None _lock = None def __new__(cls): if cls._instance is None: cls._lock = __import__('threading').Lock() with cls._lock: if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._initialized = False return cls._instance def __init__(self): if self._initialized: return self._instances: Dict[str, ShortTermMemory] = {} self._initialized = True logger.debug("✅ ShortTermMemoryManager inicializado") def get_or_create( self, conversation_id: str, user_id: str = "", max_messages: int = 100 ) -> ShortTermMemory: """ Obtém ou cria STM para uma conversa. Args: conversation_id: ID único da conversa user_id: ID do usuário max_messages: Máximo de mensagens na STM Returns: Instância de ShortTermMemory """ if conversation_id not in self._instances: self._instances[conversation_id] = ShortTermMemory( conversation_id=conversation_id, max_messages=max_messages ) logger.debug(f"🧠 STM criada: {conversation_id[:8]}...") return self._instances[conversation_id] def add_message( self, conversation_id: str, role: str, content: str, emocao: str = "neutral", reply_info: Optional[Dict] = None, importancia: Optional[float] = None ) -> MessageWithContext: """ Adiciona mensagem à STM de uma conversa. Args: conversation_id: ID da conversa role: "user" ou "assistant" content: Texto da mensagem emocao: Emoção detectada reply_info: Info de reply (se aplicável) importancia: Importância customizada Returns: MessageWithContext criada """ stm = self.get_or_create(conversation_id) # Calcula importância automaticamente se não fornecida if importancia is None: from .short_term_memory import calcular_importancia importancia = calcular_importancia( is_reply=bool(reply_info and reply_info.get("is_reply")), reply_to_bot=bool(reply_info and reply_info.get("reply_to_bot")), mensagem=content, emocao=emocao ) return stm.add_message( role=role, content=content, importancia=importancia, emocao=emocao, reply_info=reply_info ) def get_context( self, conversation_id: str, include_replies: bool = True, prioritize_replies: bool = True, max_messages: int = 10, max_tokens: int = 4000 ) -> List[MessageWithContext]: """ Obtém contexto da STM de uma conversa. Args: conversation_id: ID da conversa include_replies: Se inclui replies prioritize_replies: Se prioriza replies max_messages: Máximo de mensagens max_tokens: Máximo de tokens Returns: Lista de mensagens """ if conversation_id not in self._instances: return [] stm = self._instances[conversation_id] return stm.get_context_window( include_replies=include_replies, prioritize_replies=prioritize_replies, max_messages=max_messages, max_tokens=max_tokens ) def get_summary(self, conversation_id: str) -> Dict[str, Any]: """ Obtém resumo da STM de uma conversa. Args: conversation_id: ID da conversa Returns: Dicionário com resumo """ if conversation_id not in self._instances: return {} stm = self._instances[conversation_id] return stm.get_conversation_summary() def clear(self, conversation_id: str) -> bool: """ Limpa STM de uma conversa. Args: conversation_id: ID da conversa Returns: True se limpou """ if conversation_id in self._instances: self._instances[conversation_id].clear() return True return False def get_messages( self, conversation_id: str, limit: int = 10, include_replies: bool = True ) -> list: """ Alias de compatibilidade para get_context(). Retorna lista de MessageWithContext para a conversa. Args: conversation_id: ID da conversa limit: Quantidade máxima de mensagens include_replies: Se inclui replies Returns: Lista de MessageWithContext """ if conversation_id not in self._instances: return [] stm = self._instances[conversation_id] result = stm.get_context_window( include_replies=include_replies, prioritize_replies=True, max_messages=limit ) return result if result else [] # ==================================== # UNIFIED CONTEXT BUILDER # ==================================== class UnifiedContextBuilder: """ Constrói contexto unificado combinando reply + STM. Philosophy: "Reply context e STM devem trabalhar em sintonia como tik e tack" Usage: builder = UnifiedContextBuilder() context = builder.build( conversation_id="...", reply_metadata={...}, current_message="..." ) prompt_section = builder.format_for_llm(context) """ def __init__(self, context_manager=None, stm_manager=None, db_instance=None): self.stm_manager = stm_manager if stm_manager else ShortTermMemoryManager() self.context_manager = context_manager self.db = db_instance self.reply_handler = None self._initialized = False def _ensure_initialized(self): """Garante inicialização do reply handler.""" if not self._initialized and UNIFIED_CONTEXT_AVAILABLE: try: self.reply_handler = ReplyContextHandler() self._initialized = True except Exception as e: logger.warning(f"UnifiedContextBuilder: falha ao init reply handler: {e}") def build( self, conversation_id: str, user_id: str = "", reply_metadata: Optional[Dict[str, Any]] = None, current_message: str = "", current_emotion: str = "neutral", stm_messages: Optional[List[MessageWithContext]] = None ) -> UnifiedMessageContext: """ Constrói contexto unificado. Args: conversation_id: ID único da conversa user_id: ID do usuário reply_metadata: Metadados do reply current_message: Mensagem atual current_emotion: Emoção atual stm_messages: Mensagens STM (usa manager se None) Returns: UnifiedMessageContext pronto para uso """ self._ensure_initialized() # ===== 1. PROCESSA REPLY CONTEXT (TIK) ===== is_reply = reply_metadata.get('is_reply', False) if reply_metadata else False reply_context = { 'is_reply': is_reply, 'reply_to_bot': reply_metadata.get('reply_to_bot', False) if reply_metadata else False, 'quoted_author': reply_metadata.get('quoted_author_name', '') if reply_metadata else '', 'quoted_content': reply_metadata.get('quoted_text_original', '') or reply_metadata.get('mensagem_citada', '') if reply_metadata else '', 'importancia': IMPORTANCIA_NORMAL, 'emocao': current_emotion, 'priority': 1 } # Calcula prioridade do reply if is_reply and reply_metadata: reply_context['priority'] = self._calculate_reply_priority( reply_metadata.get('reply_to_bot', False), current_message, reply_metadata.get('quoted_text_original', '') ) # Calcula importância baseada em prioridade if reply_context['priority'] >= PRIORITY_REPLY_TO_BOT_SHORT_QUESTION: reply_context['importancia'] = IMPORTANCIA_PERGUNTA_CURTA_REPLY elif reply_context['priority'] >= PRIORITY_REPLY_TO_BOT: reply_context['importancia'] = IMPORTANCIA_REPLY_TO_BOT elif reply_context['priority'] >= PRIORITY_REPLY: reply_context['importancia'] = IMPORTANCIA_REPLY # ===== 2. OBTÉM STM (TOK) ===== if stm_messages is None: stm_messages = self.stm_manager.get_context( conversation_id, include_replies=True, prioritize_replies=True, max_messages=10, max_tokens=4000 ) # ===== 3. CALCULA TOKEN BUDGET ===== budget = ContextTokenBudget().calculate( is_reply=is_reply, reply_priority=reply_context['priority'] ) # ===== 4. FETCH LONG-TERM MEMORY (DB) ===== long_term_memory_string = "" if self.db and user_id: try: # Recuperar aprendizados e gírias ltm_facts = self.db.recuperar_aprendizado_detalhado(user_id) ltm_girias = self.db.recuperar_girias_usuario(user_id) ltm_tom = self.db.obter_tom_predominante(user_id) persona_ltm = self.db.recuperar_persona(user_id) if hasattr(self.db, 'recuperar_persona') else None ltm_lines = [] # --- PERSONA DO USUÁRIO (Rastreador) --- if persona_ltm: ltm_lines.append("=== PERFIL ANALISADO DO USUÁRIO ===") if persona_ltm.get('personalidade') and persona_ltm['personalidade'] != "None": ltm_lines.append(f"• Personalidade: {persona_ltm['personalidade']}") if persona_ltm.get('gostos') and persona_ltm['gostos'] != "None": ltm_lines.append(f"• Tópicos de Interesse: {persona_ltm['gostos']}") if persona_ltm.get('desgostos') and persona_ltm['desgostos'] != "None": ltm_lines.append(f"• Desgostos/Gatilhos: {persona_ltm['desgostos']}") if persona_ltm.get('vicios_linguagem') and persona_ltm['vicios_linguagem'] != "None": ltm_lines.append(f"• Padrões de Linguagem: {persona_ltm['vicios_linguagem']}") if persona_ltm.get('emocional') and persona_ltm['emocional'] != "None": ltm_lines.append(f"• Perfil Emocional: {persona_ltm['emocional']}") if ltm_tom: ltm_lines.append(f"• Seu tom de conversa predominante é: {ltm_tom}") if ltm_facts and isinstance(ltm_facts, dict): # Ignorar chaves puramente técnicas como 'emocao_atual' ou strings de timestamp longas fatos_filtrados = {k: v for k, v in ltm_facts.items() if not k.startswith("emocao_")} if fatos_filtrados: ltm_lines.append("• Fatos Relevantes Aprendidos:") for k, v in list(fatos_filtrados.items())[:5]: # limita 5 ltm_lines.append(f" - {k}: {v}") if ltm_girias: ltm_lines.append("• Expressões Específicas Recentes:") for g in ltm_girias[:5]: ltm_lines.append(f" - {g['giria']} ({g['significado']})") if ltm_lines: long_term_memory_string = "\n".join(ltm_lines) except Exception as e: logger.warning(f"Erro ao recuperar memória de longo prazo: {e}") # ===== 5. CRIA CONTEXTO UNIFICADO ===== unified = UnifiedMessageContext( conversation_id=conversation_id, user_id=user_id, timestamp=time.time(), is_reply=is_reply, reply_to_bot=reply_context['reply_to_bot'], reply_priority=reply_context['priority'], quoted_author=reply_context['quoted_author'], quoted_content=reply_context['quoted_content'], reply_importancia=reply_context['importancia'], stm_messages=stm_messages, stm_summary=self.stm_manager.get_summary(conversation_id), stm_emotional_trend=self._get_stm_emotional_trend(stm_messages), long_term_memory=long_term_memory_string, sync_mode="tiktok", token_budget=budget, current_message=current_message, current_emotion=current_emotion ) return unified def _calculate_reply_priority( self, reply_to_bot: bool, current_message: str, quoted_content: str ) -> int: """ Calcula nível de prioridade do reply. Returns: 1=normal, 2=reply, 3=reply_to_bot, 4=critical """ if not reply_to_bot: return PRIORITY_REPLY if is_pergunta_curta(current_message): return PRIORITY_REPLY_TO_BOT_SHORT_QUESTION return PRIORITY_REPLY_TO_BOT def _get_stm_emotional_trend( self, stm_messages: List[MessageWithContext] ) -> str: """Obtém tendência emocional da STM.""" if not stm_messages: return "neutral" emocoes = {} for msg in stm_messages[-10:]: # Últimas 10 emocao = msg.emocao or "neutral" emocoes[emocao] = emocoes.get(emocao, 0) + 1 if not emocoes: return "neutral" return max(emocoes, key=emocoes.get) def format_for_llm( self, unified: UnifiedMessageContext, include_header: bool = True ) -> str: """ Formata contexto unificado para o prompt do LLM. Args: unified: Contexto unificado include_header: Se inclui cabeçalho Returns: String formatada para o prompt """ return format_unified_context_for_llm(unified, unified.token_budget) def add_to_stm( self, conversation_id: str, role: str, content: str, emocao: str = "neutral", reply_info: Optional[Dict] = None, resposta: str = "" ) -> MessageWithContext: """ Adiciona mensagem (user ou bot) à STM. Args: conversation_id: ID da conversa role: "user" ou "assistant" content: Conteúdo da mensagem emocao: Emoção reply_info: Info de reply (se aplicável) resposta: Resposta do bot (se for assistant) Returns: MessageWithContext criada """ # Para mensagens do bot, usa a resposta gerada if role == "assistant" and resposta: content = resposta return self.stm_manager.add_message( conversation_id=conversation_id, role=role, content=content, emocao=emocao, reply_info=reply_info ) def merge_reply_with_stm( self, reply_context: Dict[str, Any], stm_messages: List[MessageWithContext], max_stm: int = 10 ) -> List[MessageWithContext]: """ Mescla reply context com STM para contexto do LLM. Args: reply_context: Contexto do reply stm_messages: Mensagens STM max_stm: Máximo de mensagens STM Returns: Lista combinada """ return sync_reply_with_stm(reply_context, stm_messages, max_stm) # ==================================== # FACTORY FUNCTIONS # ==================================== _unified_builder: Optional[UnifiedContextBuilder] = None def get_unified_context_builder() -> UnifiedContextBuilder: """Obtém instância singleton do builder.""" global _unified_builder if _unified_builder is None: _unified_builder = UnifiedContextBuilder() return _unified_builder def get_stm_manager() -> ShortTermMemoryManager: """Obtém instância singleton do manager de STM.""" return ShortTermMemoryManager() def build_unified_context( conversation_id: str, user_id: str = "", reply_metadata: Optional[Dict[str, Any]] = None, current_message: str = "", current_emotion: str = "neutral" ) -> UnifiedMessageContext: """ Factory function para construir contexto unificado. Usage: context = build_unified_context( conversation_id="pv:2449...", reply_metadata={...}, current_message="." ) """ builder = get_unified_context_builder() return builder.build( conversation_id=conversation_id, user_id=user_id, reply_metadata=reply_metadata, current_message=current_message, current_emotion=current_emotion ) # ==================================== # COMPATIBILITY HELPERS # ==================================== def gerar_id_conversao( numero: str, tipo_conversa: str = "pv", grupo_id: Optional[str] = None ) -> str: """ Gera ID de conversa para STM isolada. Args: numero: Número do usuário tipo_conversa: "pv" ou "grupo" grupo_id: ID do grupo (para conversas em grupo) Returns: ID único da conversa """ from .context_isolation import generate_context_id return generate_context_id(numero, tipo_conversa, grupo_id) # type: ignore