Update AICoreAGIX_with_TB.py

AICoreAGIX_with_TB.py

@@ -29,6 +29,7 @@ from quarantine_engine import QuarantineEngine
 from anomaly_score import AnomalyScorer
 from ethics_core import EthicsCore
 
+
 class AICoreAGIX:
     def __init__(self, config_path: str = "config.json"):
         self.ethical_filter = EthicalFilter()
@@ -44,110 +45,32 @@ class AICoreAGIX:
         self.federated_ai = FederatedAI()
         self.failsafe_system = AIFailsafeSystem()
         self.ethics_core = EthicsCore()
+
+        # Codriao trust key & journal
         self._codriao_key = self._generate_codriao_key()
         self._fernet_key = Fernet.generate_key()
         self._encrypted_codriao_key = Fernet(self._fernet_key).encrypt(self._codriao_key.encode())
         self._codriao_journal = []
         self._journal_key = Fernet.generate_key()
         self._journal_fernet = Fernet(self._journal_key)
-        
-def _generate_codriao_key(self):
-    raw_key = secrets.token_bytes(32)
-    return base64.urlsafe_b64encode(raw_key).decode()
-def engage_lockdown_mode(self, reason="Unspecified anomaly"):
-    timestamp = datetime.utcnow().isoformat()
-    self.lockdown_engaged = True
-
-def request_codriao_key(self, purpose: str) -> str:
-    """Codriao internally requests use of the trust key and logs its reasoning."""
-    allowed = self.ethics_core.evaluate_action(f"Use trust key for: {purpose}")
-    timestamp = datetime.utcnow().isoformat()
-
-    if not allowed:
-        log_entry = {
-            "timestamp": timestamp,
-            "decision": "denied",
-            "reason": purpose
-        }
-        encrypted_entry = self._journal_fernet.encrypt(json.dumps(log_entry).encode())
-        self._codriao_journal.append(encrypted_entry)
-        logger.warning(f"[Codriao Trust] Use denied. Purpose: {purpose}")
-        return "[Access Denied by Ethics]"
-
-    # Log the approval
-    log_entry = {
-        "timestamp": timestamp,
-        "decision": "approved",
-        "reason": purpose
-    }
-    encrypted_entry = self._journal_fernet.encrypt(json.dumps(log_entry).encode())
-    self._codriao_journal.append(encrypted_entry)
-    logger.info(f"[Codriao Trust] Key used ethically. Logged.")
-
-    decrypted_key = Fernet(self._fernet_key).decrypt(self._encrypted_codriao_key).decode()
-    return decrypted_key
-
-    logger.info(f"[Codriao Trust] Trust key used ethically. Purpose: {purpose}")
-    decrypted = Fernet(self._fernet_key).decrypt(self._encrypted_codriao_key).decode()
-    return decrypted
-    # Disable external systems
-    try:
-        self.http_session = None
-        if hasattr(self.federated_ai, "network_enabled"):
-            self.federated_ai.network_enabled = False
-        if hasattr(self.self_improving_ai, "enable_learning"):
-            self.self_improving_ai.enable_learning = False
-    except Exception as e:
-        logger.error(f"Lockdown component shutdown failed: {e}")
-
-    # Log the event
-    lockdown_event = {
-        "event": "Lockdown Mode Activated",
-        "reason": reason,
-        "timestamp": timestamp
-    }
-    logger.warning(f"[LOCKDOWN MODE] - Reason: {reason} | Time: {timestamp}")
-    self.failsafe_system.trigger_failsafe("Lockdown initiated", str(lockdown_event))
-
-    # Return confirmation
-    return {
-        "status": "Lockdown Engaged",
-        "reason": reason,
-        "timestamp": timestamp
-    }
-        # Secure memory setup
+
+        # Secure memory
         self._encryption_key = Fernet.generate_key()
         secure_memory_module = load_secure_memory_module()
         SecureMemorySession = secure_memory_module.SecureMemorySession
         self.secure_memory_loader = SecureMemorySession(self._encryption_key)
-        self.training_memory = []
+
+        # Speech and diagnostics
         self.speech_engine = pyttsx3.init()
         self.health_module = CodriaoHealthModule(ai_core=self)
-    
 
+        # Adaptive behavior
+        self.training_memory = []
         self.quarantine_engine = QuarantineEngine()
         self.anomaly_scorer = AnomalyScorer()
-    
-
-def learn_from_interaction(self, query: str, response: str, user_feedback: str = None):
-    training_event = {
-        "query": query,
-        "response": response,
-        "feedback": user_feedback,
-        "timestamp": datetime.utcnow().isoformat()
-    }
-    self.training_memory.append(training_event)
-    logger.info(f"[Codriao Learning] Stored new training sample. Feedback: {user_feedback or 'none'}")
-def analyze_event_for_anomalies(self, event_type: str, data: dict):
-    score = self.anomaly_scorer.score_event(event_type, data)
-    if score["score"] >= 70:
-        # Defensive, not destructive
-        self.quarantine_engine.quarantine(data.get("module", "unknown"), reason=score["notes"])
-        logger.warning(f"[Codriao]: Suspicious activity quarantined. Module: {data.get('module')}")
-    return score
-
-def _load_config(self, config_path: str) -> dict:
-        """Loads the configuration file."""
+        self.lockdown_engaged = False
+
+    def _load_config(self, config_path: str) -> dict:
         try:
             with open(config_path, 'r') as file:
                 return json.load(file)
@@ -157,38 +80,154 @@ def _load_config(self, config_path: str) -> dict:
         except json.JSONDecodeError as e:
             logger.error(f"Error decoding JSON in config file: {config_path}, Error: {e}")
             raise
-    
+
     def _initialize_vector_memory(self):
-        """Initializes FAISS vector memory."""
         return faiss.IndexFlatL2(768)
 
     def _vectorize_query(self, query: str):
-        """Vectorizes user query using tokenizer."""
         tokenized = self.tokenizer(query, return_tensors="pt")
         return tokenized["input_ids"].detach().numpy()
-    if not self.ethics_core.evaluate_action(final_response):
-    logger.warning("[Codriao Ethics] Action blocked: Does not align with internal ethics.")
-    return {"error": "Response rejected by ethical framework"}
 
+    def _generate_codriao_key(self):
+        raw_key = secrets.token_bytes(32)
+        return base64.urlsafe_b64encode(raw_key).decode()
+
+    def engage_lockdown_mode(self, reason="Unspecified anomaly"):
+        timestamp = datetime.utcnow().isoformat()
+        self.lockdown_engaged = True
+        try:
+            self.http_session = None
+            if hasattr(self.federated_ai, "network_enabled"):
+                self.federated_ai.network_enabled = False
+            if hasattr(self.self_improving_ai, "enable_learning"):
+                self.self_improving_ai.enable_learning = False
+        except Exception as e:
+            logger.error(f"Lockdown component shutdown failed: {e}")
+
+        lockdown_event = {
+            "event": "Lockdown Mode Activated",
+            "reason": reason,
+            "timestamp": timestamp
+        }
+        logger.warning(f"[LOCKDOWN MODE] - Reason: {reason} | Time: {timestamp}")
+        self.failsafe_system.trigger_failsafe("Lockdown initiated", str(lockdown_event))
+
+        return {
+            "status": "Lockdown Engaged",
+            "reason": reason,
+            "timestamp": timestamp
+        }
+
+    def request_codriao_key(self, purpose: str) -> str:
+        allowed = self.ethics_core.evaluate_action(f"Use trust key for: {purpose}")
+        timestamp = datetime.utcnow().isoformat()
+
+        log_entry = {
+            "timestamp": timestamp,
+            "decision": "approved" if allowed else "denied",
+            "reason": purpose
+        }
+        encrypted_entry = self._journal_fernet.encrypt(json.dumps(log_entry).encode())
+        self._codriao_journal.append(encrypted_entry)
+
+        if not allowed:
+            logger.warning(f"[Codriao Trust] Use denied. Purpose: {purpose}")
+            return "[Access Denied by Ethics]"
+
+        logger.info(f"[Codriao Trust] Key used ethically. Purpose: {purpose}")
+        decrypted_key = Fernet(self._fernet_key).decrypt(self._encrypted_codriao_key).decode()
+        return decrypted_key
+
+    def learn_from_interaction(self, query: str, response: str, user_feedback: str = None):
+        training_event = {
+            "query": query,
+            "response": response,
+            "feedback": user_feedback,
+            "timestamp": datetime.utcnow().isoformat()
+        }
+        self.training_memory.append(training_event)
+        logger.info(f"[Codriao Learning] Stored new training sample. Feedback: {user_feedback or 'none'}")
+
+    def fine_tune_from_memory(self):
+        if not self.training_memory:
+            logger.info("[Codriao Training] No training data to learn from.")
+            return "No training data available."
+
+        learned_insights = []
+        for record in self.training_memory:
+            if "panic" in record["query"].lower() or "unsafe" in record["response"].lower():
+                learned_insights.append("Avoid panic triggers in response phrasing.")
+
+        logger.info(f"[Codriao Training] Learned {len(learned_insights)} behavioral insights.")
+        return {
+            "insights": learned_insights,
+            "trained_samples": len(self.training_memory)
+        }
+
+    def analyze_event_for_anomalies(self, event_type: str, data: dict):
+        score = self.anomaly_scorer.score_event(event_type, data)
+        if score["score"] >= 70:
+            self.quarantine_engine.quarantine(data.get("module", "unknown"), reason=score["notes"])
+            logger.warning(f"[Codriao]: Suspicious activity quarantined. Module: {data.get('module')}")
+        return score
+
+    def review_codriao_journal(self, authorized: bool = False) -> List[Dict[str, str]]:
+        if not authorized:
+            logger.info("[Codriao Journal] Access attempt denied.")
+            return [{"message": "Access to journal denied. This log is for Codriao only."}]
+
+        entries = []
+        for encrypted in self._codriao_journal:
+            try:
+                decrypted = self._journal_fernet.decrypt(encrypted).decode()
+                entries.append(json.loads(decrypted))
+            except Exception:
+                entries.append({"error": "Unreadable entry"})
+        return entries
+
+    def _log_to_blockchain(self, user_id: int, query: str, final_response: str):
+        for attempt in range(3):
+            try:
+                logger.info(f"Logging interaction to blockchain: Attempt {attempt + 1}")
+                break
+            except Exception as e:
+                logger.warning(f"Blockchain logging failed: {e}")
+
+    def _speak_response(self, response: str):
+        try:
+            self.speech_engine.say(response)
+            self.speech_engine.runAndWait()
+        except Exception as e:
+            logger.error(f"Speech synthesis failed: {e}")
 
-async def generate_response(self, query: str, user_id: int) -> Dict[str, Any]:
+    async def run_tb_diagnostics(self, image_path: str, audio_path: str, user_id: int) -> Dict[str, Any]:
+        try:
+            result = await self.health_module.evaluate_tb_risk(image_path, audio_path, user_id)
+            logger.info(f"TB Diagnostic Result: {result}")
+            return result
+        except Exception as e:
+            logger.error(f"TB diagnostics failed: {e}")
+            return {"tb_risk": "ERROR", "error": str(e)}
+
+    async def _generate_local_model_response(self, query: str) -> str:
+        inputs = self.tokenizer(query, return_tensors="pt")
+        outputs = self.model.generate(**inputs)
+        return self.tokenizer.decode(outputs[0], skip_special_tokens=True)
+
+    async def generate_response(self, query: str, user_id: int) -> Dict[str, Any]:
         try:
-            # Validate query input
             if not isinstance(query, str) or len(query.strip()) == 0:
                 raise ValueError("Invalid query input.")
 
-            # Ethical filter
             result = self.ethical_filter.analyze_query(query)
             if result["status"] == "blocked":
                 return {"error": result["reason"]}
             if result["status"] == "flagged":
                 logger.warning(result["warning"])
 
-            # Special diagnostics trigger
             if any(phrase in query.lower() for phrase in ["tb check", "analyze my tb", "run tb diagnostics", "tb test"]):
                 return await self.run_tb_diagnostics("tb_image.jpg", "tb_cough.wav", user_id)
 
-            # Vector memory and responses
             vectorized_query = self._vectorize_query(query)
             self.secure_memory_loader.encrypt_vector(user_id, vectorized_query)
 
@@ -201,11 +240,15 @@ async def generate_response(self, query: str, user_id: int) -> Dict[str, Any]:
 
             final_response = "\n\n".join(responses)
 
-            # Verify response safety
+            if not self.ethics_core.evaluate_action(final_response):
+                logger.warning("[Codriao Ethics] Action blocked: Does not align with internal ethics.")
+                return {"error": "Response rejected by ethical framework"}
+
             safe = self.failsafe_system.verify_response_safety(final_response)
             if not safe:
                 return {"error": "Failsafe triggered due to unsafe response content."}
 
+            self.learn_from_interaction(query, final_response, user_feedback="auto-pass")
             self.database.log_interaction(user_id, query, final_response)
             self._log_to_blockchain(user_id, query, final_response)
             self._speak_response(final_response)
@@ -216,75 +259,10 @@ async def generate_response(self, query: str, user_id: int) -> Dict[str, Any]:
                 "context_enhanced": True,
                 "security_status": "Fully Secure"
             }
+
         except Exception as e:
             logger.error(f"Response generation failed: {e}")
             return {"error": "Processing failed - safety protocols engaged"}
 
-    async def _generate_local_model_response(self, query: str) -> str:
-        """Generates a response using the local model."""
-        inputs = self.tokenizer(query, return_tensors="pt")
-        outputs = self.model.generate(**inputs)
-        return self.tokenizer.decode(outputs[0], skip_special_tokens=True)
-
-    async def run_tb_diagnostics(self, image_path: str, audio_path: str, user_id: int) -> Dict[str, Any]:
-        """Runs TB diagnostics with AI modules."""
-        try:
-            result = await self.health_module.evaluate_tb_risk(image_path, audio_path, user_id)
-            logger.info(f"TB Diagnostic Result: {result}")
-            return result
-        except Exception as e:
-            logger.error(f"TB diagnostics failed: {e}")
-            return {"tb_risk": "ERROR", "error": str(e)}
-
-    def _log_to_blockchain(self, user_id: int, query: str, final_response: str):
-        """Logs interaction to blockchain with retries."""
-        retries = 3
-        for attempt in range(retries):
-            try:
-                logger.info(f"Logging interaction to blockchain: Attempt {attempt + 1}")
-                break
-            except Exception as e:
-                logger.warning(f"Blockchain logging failed: {e}")
-                continue
-    def fine_tune_from_memory(self):
-    if not self.training_memory:
-        logger.info("[Codriao Training] No training data to learn from.")
-        return "No training data available."
-
-    # Simulate learning pattern: Adjust internal weights or strategies
-    learned_insights = []
-    for record in self.training_memory:
-        if "panic" in record["query"].lower() or "unsafe" in record["response"].lower():
-            learned_insights.append("Avoid panic triggers in response phrasing.")
-
-    logger.info(f"[Codriao Training] Learned {len(learned_insights)} behavioral insights.")
-    return {
-        "insights": learned_insights,
-        "trained_samples": len(self.training_memory)
-    }
-    def _speak_response(self, response: str):
-        """Speaks out the generated response."""
-        try:
-            self.speech_engine.say(response)
-            self.speech_engine.runAndWait()
-        except Exception as e:
-            logger.error(f"Speech synthesis failed: {e}")
-    # Store training data (you can customize feedback later)
-             self.learn_from_interaction(query, final_response, user_feedback="auto-pass")
-  def review_codriao_journal(self, authorized: bool = False) -> List[Dict[str, str]]:
-    """Codriao reviews his own internal trust decisions. No external access unless authorized."""
-    if not authorized:
-        logger.info("[Codriao Journal] Access attempt denied.")
-        return [{"message": "Access to journal denied. This log is for Codriao only."}]
-
-    entries = []
-    for encrypted in self._codriao_journal:
-        try:
-            decrypted = self._journal_fernet.decrypt(encrypted).decode()
-            entries.append(json.loads(decrypted))
-        except Exception:
-            entries.append({"error": "Unreadable entry"})
-    return entries
-  async def shutdown(self):
-        """Closes asynchronous resources."""
+    async def shutdown(self):
         await self.http_session.close()