Upload engine.py

engine.py

@@ -0,0 +1,327 @@
+
+"""
+engine.py
+Orquestador principal del motor Savant Simbiótico RRF.
+Expone:
+- handle_query(text): detecta intención (map/resonance/music/chat) y responde
+- access to SimpleTrainer, SelfImprover, MemoryStore for external control
+"""
+import time # Import time
+from .mappings import IcosaMap, DodecaMap
+from .resonance import ResonanceSimulator
+from .music import MusicAdapter
+from .memory import MemoryStore
+from .self_improvement import SelfImprover
+# from .trainer import SimpleTrainer # Avoid circular import, trainer can be instantiated externally
+from .api_helpers import chat_refine
+import os # Import os
+import pandas as pd # Import pandas
+import json # Import json
+import pickle # Import pickle
+
+class SavantEngine:
+    def __init__(self, structured_data_paths=None):
+        self.memory = MemoryStore("SAVANT_memory.jsonl")
+        # Load structured data if paths are provided
+        self.structured_data = {}
+        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'))
+
+                 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 = {} # Reset if loading fails
+
+
+        # Instantiate components, passing relevant structured data
+        self.icosa = IcosaMap(node_data=self.structured_data.get('icosahedron_nodes')) # Pass node data
+        self.dodeca = DodecaMap() # No dodecahedron data provided in list
+        self.resonator = ResonanceSimulator(frequencies_data=self.structured_data.get('frequencies'), constants_data=self.structured_data.get('constants')) # Pass freq/const data
+        self.music = MusicAdapter(frequencies_data=self.structured_data.get('frequencies')) # Pass frequencies data
+        self.self_improver = SelfImprover(self.memory, structured_data=self.structured_data) # Pass structured data to SelfImprover
+
+
+        self._interaction_count = 0 # Initialize interaction count for self-improvement trigger
+
+
+    # Helper methods for loading data within the Engine (copied from Trainer for self-containment)
+    def _load_json_data(self, file_path):
+        """Loads data from a JSON file."""
+        if not file_path or not os.path.exists(file_path):
+            # print(f"JSON file not found or path not provided: {file_path}") # Suppress not found for optional files
+            return None
+        try:
+            with open(file_path, "r", encoding="utf-8") as f:
+                data = json.load(f)
+            # print(f"Successfully loaded JSON data from {file_path}") # Suppress success for cleaner output
+            return data
+        except json.JSONDecodeError as e:
+            print(f"Error decoding JSON from {file_path}: {e}")
+            return None
+        except Exception as e:
+            print(f"An unexpected error occurred while loading JSON data: {e}")
+            return None
+
+    def _load_csv_data(self, file_path):
+        """Loads data from a CSV file using pandas."""
+        if not file_path or not os.path.exists(file_path):
+             # print(f"CSV file not found or path not provided: {file_path}") # Suppress not found for optional files
+             return []
+        try:
+            df = pd.read_csv(file_path)
+            # print(f"Successfully loaded CSV data from {file_path}") # Suppress success for cleaner output
+            return df.to_dict(orient='records')
+        except Exception as e:
+            print(f"An error occurred while loading CSV data from {file_path}: {e}")
+            return []
+
+
+    def _classify(self, text):
+        t = text.lower()
+        # Enhanced classification based on structured data keywords and patterns
+        if any(k in t for k in ("equation", "ecuacion", "hamiltoniano", "dirac", "formula", "formulae", "formulas")): # Added formula variations
+             return "equation_query"
+        if any(k in t for k in ("node", "nodo", "icosahedron", "dodecahedron", "poly", "vertex", "point", "map")): # Added map keyword to node query
+             # Check for patterns like "node X" where X is a number
+             words = t.split()
+             if len(words) > 1 and words[-1].isdigit() and words[-2] in ("node", "nodo"):
+                  return "node_query"
+             return "node_query"
+        if any(k in t for k in ("frecuen", "freq", "music", "nota", "melod", "tono", "pitch", "scale", "musical", "sound", "audio")): # Added sound, audio
+             return "music_resonance" # Combine music and resonance intent for simplicity here
+        if any(k in t for k in ("constant", "constante", "valor", "unidad", "define", "what is the value of")): # Added "what is the value of"
+             return "constant_query"
+        if any(k in t for k in ("resonance", "resonar", "resonant", "vibration", "oscilla")): # Specific keywords for resonance without music
+             return "resonance_only"
+
+        # Existing classifications (kept as fallbacks or for broader terms)
+        # Removed redundant 'reson' and 'sinton' mapping to music_resonance as specific resonance_only added
+        if any(k in t for k in ("chat", "hola", "qué", "como", "explica", "tell me", "what is", "describe", "info", "information")): # Added info, information
+            return "chat"
+        return "chat" # Default to chat
+
+
+    def handle_query(self, text, base_model_output=None):
+        kind = self._classify(text)
+
+        # Handle query types based on structured data
+        if kind == "equation_query":
+             relevant_eqs = []
+             if self.structured_data.get('equations'):
+                  # Find equations related to the query (more robust keyword matching)
+                  query_words = text.lower().split()
+                  relevant_eqs = [eq for eq in self.structured_data['equations'] if any(word in eq.get('nombre', '').lower() or word in eq.get('descripcion', '').lower() or any(comp.lower() in word for comp in eq.get('componentes', [])) for word in query_words)]
+
+             if relevant_eqs:
+                  # Provide information about found equations
+                  response_parts = ["Based on the RRF Equations data, I found the following relevant equations:"]
+                  for eq in relevant_eqs[:3]: # Limit to first 3 for brevity
+                      response_parts.append(f"- '{eq.get('nombre', 'N/A')}' ({eq.get('tipo', 'Equation')}): {eq.get('ecuacion', 'N/A')} (Components: {', '.join(eq.get('componentes', []))})")
+                  if len(relevant_eqs) > 3:
+                       response_parts.append("...")
+                  response = "\n".join(response_parts)
+                  self._log_interaction(text, base_model_output, response, type="equation_query")
+                  return {"type": "equation_query", "query": text, "result": relevant_eqs, "response": response}
+             else:
+                  response = "I couldn't find any relevant equations in the loaded data for that query."
+                  self._log_interaction(text, base_model_output, response, type="equation_query_not_found")
+                  return {"type": "equation_query", "query": text, "result": [], "response": response}
+
+
+        if kind == "node_query":
+             relevant_nodes = []
+             if self.structured_data.get('icosahedron_nodes'):
+                  query_words = text.lower().split()
+                  # Try to find by ID first if query contains a number
+                  try:
+                       node_id = int(query_words[-1]) if query_words and query_words[-1].isdigit() else None
+                       if node_id is not None:
+                            relevant_nodes = [node for node in self.structured_data['icosahedron_nodes'] if node.get('id') == node_id]
+                  except (ValueError, IndexError):
+                        pass # Not a number query
+
+                  # If not found by ID or not a number query, search by keyword in description/name
+                  if not relevant_nodes:
+                       relevant_nodes = [node for node in self.structured_data['icosahedron_nodes'] if any(word in node.get('description', '').lower() or word in node.get('name', '').lower() for word in query_words)]
+
+             if relevant_nodes:
+                  response_parts = ["Based on the Icosahedron Nodes data, I found the following relevant nodes:"]
+                  for node in relevant_nodes[:3]: # Limit to first 3
+                       response_parts.append(f"- Node {node.get('id', 'N/A')}: {node.get('description', node.get('name', 'No description'))} (Coords: ({node.get('x', 'N/A')}, {node.get('y', 'N/A')}, {node.get('z', 'N/A')}))") # Added N/A checks
+                  if len(relevant_nodes) > 3:
+                       response_parts.append("...")
+                  response = "\n".join(response_parts)
+                  self._log_interaction(text, base_model_output, response, type="node_query")
+                  return {"type": "node_query", "query": text, "result": relevant_nodes, "response": response}
+             else:
+                  response = "I couldn't find any relevant nodes in the loaded data for that query."
+                  self._log_interaction(text, base_model_output, response, type="node_query_not_found")
+                  return {"type": "node_query", "query": text, "result": [], "response": response}
+
+        if kind == "music_resonance":
+            # Can still trigger resonance simulation and music adaptation
+            # Enhance response with information from frequencies/constants if relevant keywords are used
+            response_parts = []
+            if self.structured_data.get('frequencies') and any(k in text.lower() for k in ("frecuen", "freq", "nota", "pitch", "scale", "musical", "sound", "audio")):
+                 query_words = text.lower().split()
+                 relevant_freqs = [f for f in self.structured_data['frequencies'] if any(word in f.get('note', '').lower() or word in f.get('role', '').lower() for word in query_words)]
+                 if relevant_freqs:
+                      response_parts.append("Based on the Frequencies data, I found:")
+                      for freq in relevant_freqs[:3]:
+                           response_parts.append(f"- Note: {freq.get('note', 'N/A')}, Frequency: {freq.get('frequency', 'N/A')} Hz, Role: {freq.get('role', 'N/A')}") # Added N/A checks
+                      if len(relevant_freqs) > 3: response_parts.append("...")
+
+            if self.structured_data.get('constants') and any(k in text.lower() for k in ("constant", "constante")):
+                 query_words = text.lower().split()
+                 relevant_constants = [c for c in self.structured_data['constants'] if any(word in c.get('name', '').lower() for word in query_words)]
+                 if relevant_constants:
+                      response_parts.append("Based on the Constants data, I found:")
+                      for const in relevant_constants[:3]:
+                           response_parts.append(f"- Constant: {const.get('name', 'N/A')}, Value: {const.get('value', 'N/A')}, Units: {const.get('units', 'N/A')}") # Added N/A checks
+                      if len(relevant_constants) > 3: response_parts.append("...")
+
+            # Always run resonance simulation and music adaptation for this type
+            r = self.resonator.simulate(text)
+            seq = self.music.adapt_text_to_music(text)
+
+            response_parts.append(f"Resonance simulation summary: Dominant Frequency={r['summary'].get('dom_freq', 0.0):.4f} Hz, Max Power={r['summary'].get('max_power', 0.0):.4f}.") # Added default values
+            response_parts.append(f"Adapted to music sequence (first 5 notes: pitch, duration): {seq[:5]}...")
+
+            response = "\n".join(response_parts) if response_parts else "Processing music and resonance query..."
+            self._log_interaction(text, base_model_output, response, type="music_resonance")
+            return {"type":"music_resonance","query":text,"resonance_result":r,"music_result":seq, "response": response}
+
+        if kind == "resonance_only": # New handler for resonance-only queries
+             # Can still trigger resonance simulation
+            response_parts = []
+            if self.structured_data.get('constants') and any(k in text.lower() for k in ("constant", "constante")):
+                 query_words = text.lower().split()
+                 relevant_constants = [c for c in self.structured_data['constants'] if any(word in c.get('name', '').lower() for word in query_words)]
+                 if relevant_constants:
+                      response_parts.append("Based on the Constants data, I found:")
+                      for const in relevant_constants[:3]:
+                           response_parts.append(f"- Constant: {const.get('name', 'N/A')}, Value: {const.get('value', 'N/A')}, Units: {const.get('units', 'N/A')}") # Added N/A checks
+                      if len(relevant_constants) > 3: response_parts.append("...")
+
+            r = self.resonator.simulate(text)
+            response_parts.append(f"Resonance simulation summary: Dominant Frequency={r['summary'].get('dom_freq', 0.0):.4f} Hz, Max Power={r['summary'].get('max_power', 0.0):.4f}.") # Added default values
+
+            response = "\n".join(response_parts) if response_parts else "Processing resonance query..."
+            self._log_interaction(text, base_model_output, response, type="resonance_only")
+            return {"type":"resonance_only","query":text,"resonance_result":r, "response": response}
+
+
+        if kind == "constant_query":
+             relevant_constants = []
+             if self.structured_data.get('constants'):
+                  query_words = text.lower().split()
+                  relevant_constants = [c for c in self.structured_data['constants'] if any(word in c.get('name', '').lower() or word in c.get('units', '').lower() for word in query_words)]
+
+             if relevant_constants:
+                  response_parts = ["Based on the RRF Constants data, I found the following relevant constants:"]
+                  for const in relevant_constants[:3]:
+                       response_parts.append(f"- Name: {const.get('name', 'N/A')}, Value: {const.get('value', 'N/A')}, Units: {const.get('units', 'N/A')}") # Added N/A checks
+                  if len(relevant_constants) > 3: response_parts.append("...")
+                  response = "\n".join(response_parts)
+                  self._log_interaction(text, base_model_output, response, type="constant_query")
+                  return {"type": "constant_query", "query": text, "result": relevant_constants, "response": response}
+             else:
+                  response = "I couldn't find any relevant constants in the loaded data for that query."
+                  self._log_interaction(text, base_model_output, response, type="constant_query_not_found")
+                  return {"type": "constant_query", "query": text, "result": [], "response": response}
+
+
+        if kind == "map":
+             # Use icosahedron_nodes data in mapping (already done in IcosaMap)
+            node_label = self.icosa.closest_node(text)
+            response = f"Mapping query '{text}' to closest node: {node_label}"
+            # If we have node data, try to find details about the mapped node
+            if self.structured_data.get('icosahedron_nodes'):
+                 # Assuming node_label is the description or name from node_data used for embedding
+                 # A more robust mapping is needed here to link label back to original node dict by ID
+                 # For now, let's just find the node with a matching description/name if possible
+                 mapped_node_data = next((node for node in self.structured_data['icosahedron_nodes'] if node.get('description', '').lower() == node_label.lower() or node.get('name', '').lower() == node_label.lower()), None)
+                 if mapped_node_data:
+                      response += f" (ID: {mapped_node_data.get('id', 'N/A')}, Coords: ({mapped_node_data.get('x', 'N/A')}, {mapped_node_data.get('y', 'N/A')}, {mapped_node_data.get('z', 'N/A')}))" # Added N/A checks
+
+
+            self._log_interaction(text, base_model_output, response, type="map")
+            return {"type":"map","query":text,"node":node_label, "response": response}
+
+        # chat fallback: if base_model_output provided, refine it using self_improver
+        if kind == "chat":
+            if base_model_output is None:
+                # default echo
+                base = "Echo: " + text
+            else:
+                base = base_model_output
+
+            refined = chat_refine(text, base, self_improver=self.self_improver)
+            response = refined # Use refined output as the main response for chat
+            self._log_interaction(text, base_model_output, refined, type="chat_interaction") # Log chat interaction
+
+            return {"type":"chat","query":text,"base":base,"refined":refined, "response": response}
+
+        # Fallback for unhandled types (shouldn't be reached with current classify)
+        response = "I'm not sure how to handle that query based on the available data and functions."
+        self._log_interaction(text, base_model_output, response, type="unhandled_query")
+        return {"type": "unhandled", "query": text, "response": response}
+
+
+    def _log_interaction(self, user_input, base_output, final_output, type="interaction"):
+         """Logs interaction details to memory and triggers self-improvement if needed."""
+         interaction_record = {
+             "type": type, # Use the specified type (e.g., chat_interaction, equation_query)
+             "user_input": user_input,
+             "base_model_output": base_output, # Might be None for non-chat types
+             "final_output": final_output, # The response generated by handle_query
+             "_ts": time.time() # Add timestamp
+         }
+         self.memory.add(interaction_record)
+
+         # Periodically trigger self-improvement (e.g., every 10 interactions)
+         self._interaction_count = getattr(self, '_interaction_count', 0) + 1
+         if self._interaction_count % 10 == 0:
+             print("SAVANT: Triggering self-improvement cycle...")
+             try:
+                 proposal = self.self_improver.propose()
+                 accepted, metric = self.self_improver.evaluate_and_apply(proposal)
+                 print(f"SAVANT: Self-improvement proposal accepted: {accepted}, New metric: {metric}")
+                 self.memory.add({
+                     "type": "self_improvement_triggered",
+                     "proposal": proposal,
+                     "accepted": accepted,
+                     "metric": metric,
+                     "_ts": time.time()
+                 })
+             except Exception as si_error:
+                     # Log the error and continue
+                     error_message = f"Error during self-improvement: {si_error}"
+                     print(f"SAVANT: {error_message}")
+                     self.memory.add({
+                         "type": "self_improvement_error",
+                         "error": error_message,
+                         "_ts": time.time()
+                     })
+
+
+    # trainer helpers (these are now called externally via SimpleTrainer instance)
+    # def run_training_epochs(self, stimuli, epochs=3):
+    #     return self.trainer.run_epochs(stimuli, epochs)
+
+    def propose_improvement(self):
+        return self.self_improver.propose()
+
+    def apply_improvement(self, proposal):
+        return self.self_improver.evaluate_and_apply(proposal)