Spaces:
Running
Running
| import json | |
| import threading | |
| import re | |
| from loguru import logger | |
| from typing import List, Dict, Any, Optional | |
| try: | |
| from modules.database import Database | |
| except ImportError: | |
| from database import Database | |
| class PersonaTracker: | |
| """ | |
| Rastreador de Persona em Background (Character.AI style LTM). | |
| Analisa as conversas recentes do usuário silenciosamente e extrai | |
| seus traços de personalidade, gostos e emoções no banco de dados. | |
| """ | |
| def __init__(self, db: Database, llm_client: Any): | |
| """ | |
| Args: | |
| db (Database): Instância do banco de dados (database.py) | |
| llm_client (Any): Instância do cliente LLM (ex: MultiLLMClient) | |
| """ | |
| self.db = db | |
| self.llm_client = llm_client | |
| self.processing_users = set() | |
| def track_background(self, numero_usuario: str, historico_recente: List[Dict[str, str]]) -> None: | |
| """ | |
| Dispara a análise de persona em background para não bloquear a resposta do bot. | |
| Args: | |
| numero_usuario: ID ou número do usuário. | |
| historico_recente: Lista de dicionários {'role': '...', 'content': '...'} com as últimas mensagens do usuário. | |
| """ | |
| if numero_usuario in self.processing_users: | |
| return # Já está a ser analisado neste momento | |
| if not historico_recente or len(historico_recente) < 3: | |
| return # Muito pouco contexto para extrair algo útil | |
| self.processing_users.add(numero_usuario) | |
| thread = threading.Thread( | |
| target=self._analyze_and_save, | |
| args=(numero_usuario, historico_recente), | |
| daemon=True | |
| ) | |
| thread.start() | |
| def _analyze_and_save(self, numero_usuario: str, historico: List[Dict[str, str]]) -> None: | |
| """Método interno que roda na Thread.""" | |
| try: | |
| # Recupera a persona atual para o LLM saber o que já sabemos | |
| persona_atual = self.db.recuperar_persona(numero_usuario) or {} | |
| # Formata histórico apenas com as falas do usuário | |
| user_messages = [msg['content'] for msg in historico if msg.get('role') == 'user'] | |
| if not user_messages: | |
| return | |
| historico_texto = "\n".join([f"User: {msg}" for msg in user_messages[-10:]]) # Últimas 10 msg | |
| perfil_atual_str = json.dumps(persona_atual, ensure_ascii=False) if persona_atual else "Ainda não definido." | |
| prompt = f"""Você é um analista comportamental focado em rastreamento de persona (Long-Term Memory). | |
| Analise as mensagens recentes deste usuário e atualize/extraia o seu perfil. | |
| [PERFIL ATUAL NO BANCO DE DADOS] | |
| {perfil_atual_str} | |
| [MENSAGENS RECENTES] | |
| {historico_texto} | |
| EXTRAIA/ATUALIZE os seguintes traços com base APENAS nas mensagens recentes e no perfil atual. Mantenha os traços do perfil atual que não foram contraditórios. | |
| Seja CONCISO. Use bullet points curtos na sua mente e preencha os campos em formato JSON estrito. | |
| Retorne APENAS um JSON válido. É OBRIGATÓRIO USAR ASPAS DUPLAS NAS CHAVES E NOS VALORES ("chave": "valor"): | |
| {{ | |
| "personalidade": "Resumo calmo, agressivo, divertido, direto, etc.", | |
| "vicios_linguagem": "Expressões ou gírias que ele usa muito.", | |
| "gostos": "O que ele demonstrou gostar ou tópicos de interesse.", | |
| "desgostos": "O que o irrita, o que ele odeia.", | |
| "emocional": "Traços emocionais, forças ou gatilhos/fraquezas." | |
| }} | |
| """ | |
| # Chama o LLM (garante formato json) | |
| # Agora retorna (resposta, modelo_usado) ou apenas resposta | |
| response_raw = self.llm_client.generate(prompt, []) | |
| if isinstance(response_raw, tuple): | |
| response_json_str = response_raw[0] | |
| else: | |
| response_json_str = response_raw | |
| if not response_json_str: | |
| return | |
| # Extrai o JSON (Robusto contra texto extra, markdown e quebras parciais) | |
| response_clean = response_json_str.strip() | |
| # 1. Localiza o início do JSON, permitindo quebras (truncado) | |
| if '{' in response_clean: | |
| start_pts = response_clean.find('{') | |
| end_pts = response_clean.rfind('}') | |
| if end_pts > start_pts: | |
| response_clean = response_clean[start_pts:end_pts+1] | |
| else: | |
| response_clean = response_clean[start_pts:] # Caso esteja truncado sem o '}' | |
| # 2. Normalização agressiva de caracteres | |
| response_clean = response_clean.replace('\r', '').replace('\n', ' ') | |
| response_clean = re.sub(r'\s+', ' ', response_clean) # Remove múltiplos espaços | |
| response_clean = re.sub(r'\\+', r'\\', response_clean) | |
| # Tenta converter aspas simples em duplas para chaves/valores | |
| response_clean = re.sub(r"(?<![a-zA-Z])'|'(?![a-zA-Z])", '"', response_clean) | |
| response_clean = response_clean.replace('""', '"') | |
| dados_extraidos = {} | |
| parsed_success = False | |
| try: | |
| # Se houver chaves json "sujas" (ex: { personalidade: "x" } ao invés de {"personalidade": "x"}) | |
| rc_temp = re.sub(r'([{,]\s*)([a-zA-Z_]+)\s*:', r'\g<1>"\g<2>":', response_clean) | |
| dados_extraidos = json.loads(rc_temp) | |
| parsed_success = True | |
| except json.JSONDecodeError: | |
| # Fallback extremo 1: tenta reconstruir dicionário com ast | |
| import ast | |
| try: | |
| ast_clean = response_clean.replace('\n', '') | |
| dados_extraidos = ast.literal_eval(ast_clean) | |
| if isinstance(dados_extraidos, dict): | |
| parsed_success = True | |
| except Exception: | |
| pass | |
| # Fallback extremo 2: Modo de extração de emergência (Regex Direto) | |
| # Ideal para '{ personalidade: Direto, irônico, vicioslinguagem: orroh, gostos: -, ... }' | |
| if not parsed_success or not isinstance(dados_extraidos, dict): | |
| logger.warning(f"Iniciando MODO DE EMERGÊNCIA Regex para Persona de {numero_usuario}...") | |
| dados_extraidos = {} | |
| chaves_busca = ["personalidade", "vicios_linguagem", "vicioslinguagem", "gostos", "desgostos", "emocional"] | |
| # Regex para encontrar "chave: valor (até encontrar outra chave ou o fim)" | |
| lookahead = "|".join(chaves_busca) | |
| for chave in chaves_busca: | |
| # Pattern que ignora aspas nas chaves e valores, parando na próxima chave conhecida | |
| pattern = re.compile(rf"['\"]?{chave}['\"]?\s*[:=]\s*(.*?)(?=(?:{lookahead})['\"]?\s*[:=]|$)", re.IGNORECASE | re.DOTALL) | |
| match = pattern.search(response_clean) | |
| if match: | |
| val = match.group(1).strip() | |
| # Limpeza radical de muletas de JSON (aspas, vírgulas no fim, chaves) | |
| val = re.sub(r'^[\s\'"{\[:]+|[\s\'"}\],:]+$', '', val).strip() | |
| if val and len(val) > 1: | |
| real_key = "vicios_linguagem" if chave == "vicioslinguagem" else chave | |
| dados_extraidos[real_key] = val | |
| if not dados_extraidos: | |
| # Se falhou tudo, mas temos a string, tentamos pelo menos salvar a string bruta como nota | |
| logger.warning(f"Falha total no Parser JSON do Persona Tracker para {numero_usuario}. Salvando payload bruto como nota.") | |
| dados_extraidos = {"personalidade": response_json_str[:200]} | |
| parsed_success = True | |
| # Limpa chaves inválidas | |
| chaves_validas = ["personalidade", "vicios_linguagem", "gostos", "desgostos", "emocional"] | |
| campos_atualizar = {k: str(v) for k, v in dados_extraidos.items() if k in chaves_validas} | |
| if campos_atualizar: | |
| sucesso = self.db.atualizar_persona(numero_usuario, campos_atualizar) | |
| if sucesso: | |
| logger.info(f"✅ Persona LTM atualizada para o usuário {numero_usuario} em background.") | |
| else: | |
| logger.warning(f"Falha ao salvar a persona no banco para {numero_usuario}.") | |
| except json.JSONDecodeError: | |
| logger.warning(f"Falha no Parser JSON do Persona Tracker para {numero_usuario}.") | |
| except Exception as e: | |
| logger.error(f"Erro no Persona Tracker background: {e}") | |
| finally: | |
| if numero_usuario in self.processing_users: | |
| self.processing_users.remove(numero_usuario) | |