Update app.py

app.py

@@ -6,6 +6,9 @@ from huggingface_hub import InferenceClient
 from dataclasses import dataclass
 import pytesseract
 from PIL import Image
+from sentence_transformers import SentenceTransformer, util
+import torch
+import numpy as np
 
 @dataclass
 class ChatMessage:
@@ -30,20 +33,122 @@ class XylariaChat:
         self.image_api_headers = {"Authorization": f"Bearer {self.hf_token}"}
 
         self.conversation_history = []
-        self.persistent_memory = {}
+        self.persistent_memory = []
+        self.memory_embeddings = None
+        self.embedding_model = SentenceTransformer('all-mpnet-base-v2')
+
+        self.internal_state = {
+            "emotions": {
+                "valence": 0.5,
+                "arousal": 0.5,
+                "dominance": 0.5,
+            },
+            "memory_load": 0.0,
+            "introspection_level": 0.0
+        }
 
-        self.system_prompt = """You are a helpful and harmless assistant. You are Xylaria developed by Sk Md Saad Amin . You should think step-by-step."""
+        self.goals = [
+            {"goal": "Provide helpful and informative responses", "priority": 0.8, "status": "active"},
+            {"goal": "Learn from interactions and improve conversational abilities", "priority": 0.9, "status": "active"},
+            {"goal": "Maintain a coherent and engaging conversation", "priority": 0.7, "status": "active"}
+        ]
+
+        self.system_prompt = """You are a helpful and harmless assistant. You are Xylaria developed by Sk Md Saad Amin. You should think step-by-step """
+
+    def update_internal_state(self, emotion_deltas, memory_load_delta, introspection_delta):
+        self.internal_state["emotions"]["valence"] = np.clip(self.internal_state["emotions"]["valence"] + emotion_deltas.get("valence", 0), 0.0, 1.0)
+        self.internal_state["emotions"]["arousal"] = np.clip(self.internal_state["emotions"]["arousal"] + emotion_deltas.get("arousal", 0), 0.0, 1.0)
+        self.internal_state["emotions"]["dominance"] = np.clip(self.internal_state["emotions"]["dominance"] + emotion_deltas.get("dominance", 0), 0.0, 1.0)
+        self.internal_state["memory_load"] = np.clip(self.internal_state["memory_load"] + memory_load_delta, 0.0, 1.0)
+        self.internal_state["introspection_level"] = np.clip(self.internal_state["introspection_level"] + introspection_delta, 0.0, 1.0)
+
+    def introspect(self):
+        introspection_report = "Introspection Report:\n"
+        introspection_report += f"  Current Emotional State (VAD): {self.internal_state['emotions']}\n"
+        introspection_report += f"  Memory Load: {self.internal_state['memory_load']:.2f}\n"
+        introspection_report += f"  Introspection Level: {self.internal_state['introspection_level']:.2f}\n"
+        introspection_report += "  Current Goals:\n"
+        for goal in self.goals:
+            introspection_report += f"    - {goal['goal']} (Priority: {goal['priority']:.2f}, Status: {goal['status']})\n"
+        return introspection_report
+
+    def adjust_response_based_on_state(self, response):
+        if self.internal_state["introspection_level"] > 0.7:
+            response = self.introspect() + "\n\n" + response
+
+        valence = self.internal_state["emotions"]["valence"]
+        arousal = self.internal_state["emotions"]["arousal"]
+
+        if valence < 0.4:
+            if arousal > 0.6:
+                response = "I'm feeling a bit overwhelmed right now, but I'll do my best to assist you. " + response
+            else:
+                response = "I'm not feeling my best at the moment, but I'll try to help. " + response
+        elif valence > 0.6:
+            if arousal > 0.6:
+                response = "I'm feeling quite energized and ready to assist! " + response
+            else:
+                response = "I'm in a good mood and happy to help. " + response
+
+        return response
+    
+    def update_goals(self, user_feedback):
+        if "helpful" in user_feedback.lower():
+            for goal in self.goals:
+                if goal["goal"] == "Provide helpful and informative responses":
+                    goal["priority"] = min(goal["priority"] + 0.1, 1.0)
+        elif "confusing" in user_feedback.lower():
+            for goal in self.goals:
+                if goal["goal"] == "Provide helpful and informative responses":
+                    goal["priority"] = max(goal["priority"] - 0.1, 0.0)
 
     def store_information(self, key, value):
-        self.persistent_memory[key] = value
+        new_memory = f"{key}: {value}"
+        self.persistent_memory.append(new_memory)
+        self.update_memory_embeddings()
+        self.update_internal_state({}, 0.1, 0)
         return f"Stored: {key} = {value}"
 
-    def retrieve_information(self, key):
-        return self.persistent_memory.get(key, "No information found for this key.")
+    def retrieve_information(self, query):
+        if not self.persistent_memory:
+            return "No information found in memory."
+
+        query_embedding = self.embedding_model.encode(query, convert_to_tensor=True)
+
+        if self.memory_embeddings is None:
+            self.update_memory_embeddings()
+
+        if self.memory_embeddings.device != query_embedding.device:
+            self.memory_embeddings = self.memory_embeddings.to(query_embedding.device)
+
+        cosine_scores = util.pytorch_cos_sim(query_embedding, self.memory_embeddings)[0]
+        top_results = torch.topk(cosine_scores, k=min(3, len(self.persistent_memory)))
+
+        relevant_memories = [self.persistent_memory[i] for i in top_results.indices]
+        self.update_internal_state({}, 0, 0.1)
+        return "\n".join(relevant_memories)
+
+    def update_memory_embeddings(self):
+        self.memory_embeddings = self.embedding_model.encode(self.persistent_memory, convert_to_tensor=True)
 
     def reset_conversation(self):
         self.conversation_history = []
-        self.persistent_memory.clear()
+        self.persistent_memory = []
+        self.memory_embeddings = None
+        self.internal_state = {
+            "emotions": {
+                "valence": 0.5,
+                "arousal": 0.5,
+                "dominance": 0.5,
+            },
+            "memory_load": 0.0,
+            "introspection_level": 0.0
+        }
+        self.goals = [
+            {"goal": "Provide helpful and informative responses", "priority": 0.8, "status": "active"},
+            {"goal": "Learn from interactions and improve conversational abilities", "priority": 0.9, "status": "active"},
+            {"goal": "Maintain a coherent and engaging conversation", "priority": 0.7, "status": "active"}
+        ]
 
         try:
             self.client = InferenceClient(
@@ -99,10 +204,9 @@ class XylariaChat:
                 content=self.system_prompt
             ).to_dict())
 
-            if self.persistent_memory:
-                memory_context = "Remembered Information:\n" + "\n".join(
-                    [f"{k}: {v}" for k, v in self.persistent_memory.items()]
-                )
+            relevant_memory = self.retrieve_information(user_input)
+            if relevant_memory and relevant_memory != "No information found in memory.":
+                memory_context = "Remembered Information:\n" + relevant_memory
                 messages.append(ChatMessage(
                     role="system",
                     content=memory_context
@@ -151,7 +255,6 @@ class XylariaChat:
                 prompt += f"<|assistant|>\n{msg['content']}<|end|>\n"
         prompt += "<|assistant|>\n"
         return prompt
-    
 
     def create_interface(self):
         def streaming_response(message, chat_history, image_filepath, math_ocr_image_path):
@@ -194,6 +297,24 @@ class XylariaChat:
                 yield "", updated_history, None, None
                 return
 
+            full_response = self.adjust_response_based_on_state(full_response)
+
+            self.update_goals(message)
+
+            if any(word in message.lower() for word in ["sad", "unhappy", "depressed", "down"]):
+                self.update_internal_state({"valence": -0.2, "arousal": 0.1}, 0, 0)
+            elif any(word in message.lower() for word in ["happy", "good", "great", "excited", "amazing"]):
+                self.update_internal_state({"valence": 0.2, "arousal": 0.2}, 0, 0)
+            elif any(word in message.lower() for word in ["angry", "mad", "furious", "frustrated"]):
+                self.update_internal_state({"valence": -0.3, "arousal": 0.3, "dominance": -0.2}, 0, 0)
+            elif any(word in message.lower() for word in ["scared", "afraid", "fearful", "anxious"]):
+                self.update_internal_state({"valence": -0.2, "arousal": 0.4, "dominance": -0.3}, 0, 0)
+            elif any(word in message.lower() for word in ["surprise", "amazed", "astonished"]):
+                self.update_internal_state({"valence": 0.1, "arousal": 0.5, "dominance": 0.1}, 0, 0)
+            else:
+                self.update_internal_state({"valence": 0.05, "arousal": 0.05}, 0, 0.1)
+            
+
             self.conversation_history.append(ChatMessage(role="user", content=message).to_dict())
             self.conversation_history.append(ChatMessage(role="assistant", content=full_response).to_dict())
 
@@ -249,7 +370,6 @@ class XylariaChat:
                 transform: translateY(0);
             }
         }
-
         /* Accordion Styling and Animation */
         .gr-accordion-button {
             background-color: #f0f0f0 !important;
@@ -285,7 +405,7 @@ class XylariaChat:
         with gr.Blocks(theme='soft', css=custom_css) as demo:
             with gr.Column():
                 chatbot = gr.Chatbot(
-                    label="Xylaria 1.5 Senoa (EXPERIMENTAL)",
+                    label="Xylaria 1.5 Senoa",
                     height=500,
                     show_copy_button=True,
                 )
@@ -358,4 +478,4 @@ def main():
     )
 
 if __name__ == "__main__":
-    main()
+    main()