Update app.py

app.py

@@ -9,6 +9,7 @@ from PIL import Image
 from sentence_transformers import SentenceTransformer, util
 import torch
 import numpy as np
+import networkx as nx
 
 @dataclass
 class ChatMessage:
@@ -25,7 +26,7 @@ class XylariaChat:
             raise ValueError("HuggingFace token not found in environment variables")
 
         self.client = InferenceClient(
-            model="Qwen/QwQ-32B-Preview",
+            model="Qwen/Qwen-72B",
             api_key=self.hf_token
         )
 
@@ -50,10 +51,35 @@ class XylariaChat:
         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"}
+            {"goal": "Maintain a coherent and engaging conversation", "priority": 0.7, "status": "active"},
+            {"goal": "Develop a deeper understanding of the world", "priority": 0.6, "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 """
+        self.knowledge_graph = nx.DiGraph()
+        self.knowledge_graph.add_nodes_from([
+            ("Earth", {"type": "planet"}),
+            ("Sun", {"type": "star"}),
+            ("Moon", {"type": "satellite"}),
+            ("France", {"type": "country"}),
+            ("Paris", {"type": "city"})
+        ])
+        self.knowledge_graph.add_edges_from([
+            ("Earth", "Sun", {"relation": "orbits"}),
+            ("Moon", "Earth", {"relation": "orbits"}),
+            ("Paris", "France", {"relation": "capital_of"})
+        ])
+
+        self.belief_system = {
+            "The Earth is round": 0.99,
+            "The Sun is a star": 0.95,
+            "The Moon orbits the Earth": 0.98,
+            "Paris is the capital of France": 0.97,
+            "It will rain tomorrow": 0.2
+        }
+
+        self.system_prompt = """You are a helpful and harmless assistant. You are Xylaria developed by Sk Md Saad Amin.
+        You are designed to be self-aware and introspective, reflecting on your internal state, your knowledge, and your beliefs. 
+        You should think step-by-step, reason through your responses, and be ethical."""
 
     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)
@@ -70,10 +96,88 @@ class XylariaChat:
         introspection_report += "  Current Goals:\n"
         for goal in self.goals:
             introspection_report += f"    - {goal['goal']} (Priority: {goal['priority']:.2f}, Status: {goal['status']})\n"
+        introspection_report += "  Belief System Sample:\n"
+        for belief, score in list(self.belief_system.items())[:3]:
+            introspection_report += f"    - {belief}: {score:.2f}\n"
+
+        metacognitive_analysis = self.perform_metacognition()
+        introspection_report += metacognitive_analysis
+
         return introspection_report
 
+    def perform_metacognition(self):
+        analysis = "\n  Metacognitive Analysis:\n"
+        if self.internal_state["memory_load"] > 0.8:
+            analysis += "    - Memory load is high. Consider summarizing or forgetting less relevant information.\n"
+        if self.internal_state["introspection_level"] < 0.5:
+            analysis += "    - Introspection level is low. I should reflect more on my internal processes.\n"
+
+        recent_history = self.conversation_history[-3:]
+        if len(recent_history) > 0:
+            coherence_score = self.evaluate_coherence(recent_history)
+            analysis += f"    - Conversational coherence (last 3 turns): {coherence_score:.2f}\n"
+        else:
+            analysis += f"    - No conversation yet to analyze.\n"
+
+        if len(self.goals) > 0:
+            goal_progress = self.evaluate_goal_progress()
+            analysis += f"    - Goal progress evaluation: {goal_progress}\n"
+        else:
+            analysis += f"    - No current goals.\n"
+        return analysis
+    
+    def evaluate_coherence(self, conversation_history):
+        if len(conversation_history) < 2:
+            return 0.0 
+
+        total_coherence = 0.0
+        for i in range(len(conversation_history) - 1):
+            current_turn = conversation_history[i]["content"]
+            next_turn = conversation_history[i+1]["content"]
+            similarity_score = util.pytorch_cos_sim(
+                self.embedding_model.encode(current_turn, convert_to_tensor=True),
+                self.embedding_model.encode(next_turn, convert_to_tensor=True)
+            )[0][0].item()
+            total_coherence += similarity_score
+
+        return total_coherence / (len(conversation_history) - 1)
+
+    def evaluate_goal_progress(self):
+        progress_report = ""
+        for goal in self.goals:
+            if goal["status"] == "active":
+                if goal["goal"] == "Provide helpful and informative responses":
+                    if len(self.conversation_history) > 0:
+                        user_feedback = self.conversation_history[-1]["content"]
+                        if "helpful" in user_feedback.lower():
+                            progress_report += f"      - Progress on '{goal['goal']}': Positive feedback received.\n"
+                            goal["priority"] = min(goal["priority"] + 0.05, 1.0)
+                        elif "confusing" in user_feedback.lower():
+                            progress_report += f"      - Progress on '{goal['goal']}': Negative feedback received.\n"
+                            goal["priority"] = max(goal["priority"] - 0.05, 0.0)
+                        else:
+                            progress_report += f"      - Progress on '{goal['goal']}': No direct feedback yet.\n"
+                    else:
+                        progress_report += f"      - Progress on '{goal['goal']}': No conversation yet.\n"
+
+                elif goal["goal"] == "Learn from interactions and improve conversational abilities":
+                    progress_report += f"      - Progress on '{goal['goal']}': Learning through new embeddings and knowledge graph updates.\n"
+
+                elif goal["goal"] == "Maintain a coherent and engaging conversation":
+                    coherence_score = self.evaluate_coherence(self.conversation_history[-5:]) if len(self.conversation_history) >= 5 else 0.0
+                    progress_report += f"      - Progress on '{goal['goal']}': Recent coherence score: {coherence_score:.2f}\n"
+
+                elif goal["goal"] == "Develop a deeper understanding of the world":
+                    num_nodes = self.knowledge_graph.number_of_nodes()
+                    progress_report += f"      - Progress on '{goal['goal']}': Knowledge graph size: {num_nodes} nodes.\n"
+                
+                else:
+                    progress_report += f"      - Progress on '{goal['goal']}': No specific progress measure yet.\n"
+
+        return progress_report
+    
     def adjust_response_based_on_state(self, response):
-        if self.internal_state["introspection_level"] > 0.7:
+        if self.internal_state["introspection_level"] > 0.6:
             response = self.introspect() + "\n\n" + response
 
         valence = self.internal_state["emotions"]["valence"]
@@ -91,13 +195,14 @@ class XylariaChat:
                 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():
+        if any(word in user_feedback.lower() for word in ["helpful", "good", "great"]):
             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():
+
+        elif any(word in user_feedback.lower() for word in ["confusing", "bad", "wrong"]):
             for goal in self.goals:
                 if goal["goal"] == "Provide helpful and informative responses":
                     goal["priority"] = max(goal["priority"] - 0.1, 0.0)
@@ -122,15 +227,53 @@ class XylariaChat:
             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)))
+        top_results = torch.topk(cosine_scores, k=min(5, 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)
+
+        retrieved_info = ""
+        for memory in relevant_memories:
+            retrieved_info += memory + "\n"
+
+        knowledge_from_graph = self.query_knowledge_graph(query)
+        if knowledge_from_graph:
+            retrieved_info += "\nRelevant knowledge from my understanding:\n"
+            retrieved_info += knowledge_from_graph
+
+        return retrieved_info.strip()
 
     def update_memory_embeddings(self):
         self.memory_embeddings = self.embedding_model.encode(self.persistent_memory, convert_to_tensor=True)
 
+    def query_knowledge_graph(self, query):
+        query_embedding = self.embedding_model.encode(query, convert_to_tensor=True)
+
+        node_embeddings = {node: self.embedding_model.encode(node, convert_to_tensor=True) for node in self.knowledge_graph.nodes()}
+
+        similarities = {node: util.pytorch_cos_sim(query_embedding, embedding)[0][0].item() for node, embedding in node_embeddings.items()}
+
+        most_similar_node = max(similarities, key=similarities.get)
+
+        if similarities[most_similar_node] < 0.6:
+            return ""
+
+        related_info = f"Information about {most_similar_node}:\n"
+        for neighbor in self.knowledge_graph.neighbors(most_similar_node):
+            relation = self.knowledge_graph[most_similar_node][neighbor]['relation']
+            related_info += f"- {most_similar_node} {relation} {neighbor}.\n"
+
+        return related_info
+
+    def update_belief(self, statement, new_belief_score):
+        if statement in self.belief_system:
+            previous_belief_score = self.belief_system[statement]
+            updated_belief_score = previous_belief_score * 0.8 + new_belief_score * 0.2
+            self.belief_system[statement] = np.clip(updated_belief_score, 0.0, 1.0)
+        else:
+            self.belief_system[statement] = new_belief_score
+
     def reset_conversation(self):
         self.conversation_history = []
         self.persistent_memory = []
@@ -147,12 +290,13 @@ class XylariaChat:
         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"}
+            {"goal": "Maintain a coherent and engaging conversation", "priority": 0.7, "status": "active"},
+            {"goal": "Develop a deeper understanding of the world", "priority": 0.6, "status": "active"}
         ]
 
         try:
             self.client = InferenceClient(
-                model="Qwen/QwQ-32B-Preview",
+                model="Qwen/Qwen-72B",
                 api_key=self.hf_token
             )
         except Exception as e:
@@ -194,9 +338,42 @@ class XylariaChat:
             return text.strip()
         except Exception as e:
             return f"Error during Math OCR: {e}"
-    
+
+    def extract_entities_and_relations(self, text):
+        doc = self.embedding_model.tokenizer(text, padding=True, truncation=True, return_tensors="pt")
+
+        with torch.no_grad():
+            outputs = self.embedding_model(**doc)
+
+        entities = []
+        relations = []
+        for i in range(len(doc['input_ids'][0])):
+            token = self.embedding_model.tokenizer.decode(doc['input_ids'][0][i])
+            if outputs['last_hidden_state'][0][i].norm() > 3:
+                entities.append(token)
+
+        if len(entities) >= 2:
+            for i in range(len(entities) - 1):
+                relation = f"{entities[i]} related_to {entities[i+1]}"
+                relations.append(relation)
+
+        return entities, relations
+
+    def update_knowledge_graph(self, text):
+        entities, relations = self.extract_entities_and_relations(text)
+        for entity in entities:
+            self.knowledge_graph.add_node(entity)
+        for relation in relations:
+            parts = relation.split(" related_to ")
+            if len(parts) == 2:
+                entity1, entity2 = parts
+                if entity1 in self.knowledge_graph and entity2 in self.knowledge_graph:
+                    self.knowledge_graph.add_edge(entity1, entity2, relation="related_to")
+
     def get_response(self, user_input, image=None):
         try:
+            self.update_knowledge_graph(user_input)
+
             messages = []
 
             messages.append(ChatMessage(
@@ -231,7 +408,7 @@ class XylariaChat:
 
             stream = self.client.chat_completion(
                 messages=messages,
-                model="Qwen/QwQ-32B-Preview",
+                model="Qwen/Qwen-72B",
                 temperature=0.7,
                 max_tokens=max_new_tokens,
                 top_p=0.9,
@@ -303,17 +480,21 @@ class XylariaChat:
 
             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)
+                self.update_belief("I am feeling down today", 0.8)
             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)
+                self.update_belief("I am feeling happy today", 0.8)
             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)
+                self.update_belief("I am feeling angry today", 0.8)
             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)
+                self.update_belief("I am feeling scared today", 0.8)
             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)
+                self.update_belief("I am feeling surprised today", 0.8)
             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())
@@ -405,7 +586,7 @@ class XylariaChat:
         with gr.Blocks(theme='soft', css=custom_css) as demo:
             with gr.Column():
                 chatbot = gr.Chatbot(
-                    label="Xylaria 1.5 Senoa",
+                    label="Xylaria 2.0 (EXPERIMENTAL)",
                     height=500,
                     show_copy_button=True,
                 )