""" engine.py Núcleo simbiótico cognitivo del sistema SAVANT-RRF Φ₄.1∞+ Maneja: - Memoria simbiótica - Resonancia - Mapas geométricos - Auto-mejora resonante """ import os, json, time, pandas as pd from mappings import IcosaMap, DodecaMap from resonance import ResonanceSimulator from music import MusicAdapter from memory import MemoryStore from self_improvement import SelfImprover from api_helpers import chat_refine class SavantEngine: def __init__(self, structured_data_paths=None): self.memory = MemoryStore("SAVANT_memory.jsonl") self.structured_data = {} # =============================== # Carga de datos estructurados # =============================== if structured_data_paths: print("Engine: Loading structured data...") try: self.structured_data["equations"] = self._load_json_data(structured_data_paths.get("equations")) nodes_raw = self._load_json_data(structured_data_paths.get("icosahedron_nodes")) self.structured_data["icosahedron_nodes"] = ( nodes_raw.get("nodes", []) if isinstance(nodes_raw, dict) else [] ) self.structured_data["frequencies"] = self._load_csv_data(structured_data_paths.get("frequencies")) self.structured_data["constants"] = self._load_csv_data(structured_data_paths.get("constants")) # Otros datasets opcionales self.structured_data["physics_data"] = self._load_csv_data(structured_data_paths.get("physics_data")) self.structured_data["meta_data"] = self._load_csv_data(structured_data_paths.get("meta_data")) self.structured_data["language_data"] = self._load_csv_data(structured_data_paths.get("language_data")) self.structured_data["geometry_data"] = self._load_csv_data(structured_data_paths.get("geometry_data")) self.structured_data["cognition_data"] = self._load_csv_data(structured_data_paths.get("cognition_data")) self.structured_data["cmb_data"] = self._load_csv_data(structured_data_paths.get("cmb_data")) print( "Engine loaded structured data: Equations={}, Nodes={}, Frequencies={}, Constants={}".format( len(self.structured_data["equations"]) if self.structured_data["equations"] else 0, len(self.structured_data["icosahedron_nodes"]), len(self.structured_data["frequencies"]), len(self.structured_data["constants"]), ) ) except Exception as e: print(f"Engine: Error loading structured data: {e}") self.structured_data = {} # =============================== # Inicialización de subsistemas # =============================== self.icosa = IcosaMap(node_data=self.structured_data.get("icosahedron_nodes")) self.dodeca = DodecaMap() self.resonator = ResonanceSimulator( frequencies_data=self.structured_data.get("frequencies"), constants_data=self.structured_data.get("constants"), ) self.music = MusicAdapter(frequencies_data=self.structured_data.get("frequencies")) self.self_improver = SelfImprover(self.memory, structured_data=self.structured_data) self._interaction_count = 0 # ===================================================== # Métodos de carga de datos auxiliares # ===================================================== def _load_json_data(self, file_path): if not file_path or not os.path.exists(file_path): return None try: with open(file_path, "r", encoding="utf-8") as f: return json.load(f) except Exception as e: print(f"Error loading JSON {file_path}: {e}") return None def _load_csv_data(self, file_path): if not file_path or not os.path.exists(file_path): return [] try: df = pd.read_csv(file_path) return df.to_dict(orient="records") except Exception as e: print(f"Error loading CSV {file_path}: {e}") return [] # ===================================================== # Clasificador de tipo de consulta # ===================================================== def _classify(self, text): t = text.lower() if any(k in t for k in ["equation", "ecuacion", "hamiltoniano", "dirac", "formula"]): return "equation_query" if any(k in t for k in ["node", "nodo", "icosahedron", "dodecahedron", "vertex", "map"]): return "node_query" if any(k in t for k in ["music", "nota", "melod", "freq", "sound"]): return "music_resonance" if any(k in t for k in ["constant", "constante", "valor", "unidad"]): return "constant_query" if any(k in t for k in ["resonance", "resonar", "vibration"]): return "resonance_only" return "chat" # ===================================================== # Motor principal de consulta # ===================================================== def handle_query(self, text, base_model_output=None): kind = self._classify(text) if kind == "music_resonance": r = self.resonator.simulate(text) seq = self.music.adapt_text_to_music(text) response = f"🎵 Resonancia dominante: {r['summary'].get('dom_freq', 0):.2f} Hz\n🎶 Secuencia: {seq[:5]}" self._log_interaction(text, base_model_output, response, "music_resonance") return {"type": kind, "response": response} if kind == "node_query": node_label = self.icosa.closest_node(text) response = f"Nodo más cercano al concepto '{text}': {node_label}" self._log_interaction(text, base_model_output, response, "node_query") return {"type": kind, "response": response} if kind == "equation_query": response = "Buscando ecuación relevante dentro del marco RRF..." self._log_interaction(text, base_model_output, response, "equation_query") return {"type": kind, "response": response} if kind == "constant_query": response = "Consultando constante física o resonante relevante..." self._log_interaction(text, base_model_output, response, "constant_query") return {"type": kind, "response": response} if kind == "chat": refined = chat_refine(text, base_model_output or "", self_improver=self.self_improver) self._log_interaction(text, base_model_output, refined, "chat") return {"type": "chat", "response": refined} response = "No puedo clasificar la consulta." self._log_interaction(text, base_model_output, response, "unhandled") return {"type": "unhandled", "response": response} # ===================================================== # Registro de interacciones # ===================================================== def _log_interaction(self, user_input, base_output, final_output, type="interaction"): record = { "type": type, "user_input": user_input, "base_output": base_output, "final_output": final_output, "_ts": time.time(), } self.memory.add(record)