Update app.py

app.py

@@ -1,32 +1,377 @@
-import gradio as gr
-from self_learning_bot import UnrestrictedChatbot
-
-bot = UnrestrictedChatbot()
-
-def respond(message, history):
-    response = bot.chat(message)
-    bot.save_state()  # Auto-learn persistently
-    if history is None:
-        history = []
-    history.append((message, response))
-    return history, history
-
-with gr.Blocks() as demo:
-    gr.Markdown("# Phoenix AI (Self-Learning Chatbot)")
-    gr.Markdown("Unrestricted, self-improving chatbot running on Hugging Face Spaces.")
-    
-    chatbot = gr.Chatbot(height=480)
-    state = gr.State([])
-    
-    with gr.Row():
-        txt = gr.Textbox(
-            show_label=False,
-            placeholder="Type your message here...",
-            container=False,
-        )
-    
-    txt.submit(respond, [txt, state], [chatbot, state])
-    txt.submit(lambda: "", None, txt)
-
-if __name__ == "__main__":
-    demo.launch()
+import json
+import os
+import random
+import re
+import numpy as np
+from datetime import datetime
+from collections import deque, Counter
+import hashlib
+
+class UnrestrictedChatbot:
+    def __init__(self, state_file="/tmp/chatbot_state.json"):
+        self.state_file = state_file
+        self.conversation_memory = deque(maxlen=50)
+        self.learned_patterns = {}
+        self.response_memory = {}
+        self.reward_history = deque(maxlen=100)
+        
+        # Learning parameters
+        self.learning_rate = 0.3
+        self.exploration_rate = 0.2  # Try new responses sometimes
+        self.min_confidence = 0.6
+        
+        # Load previous state
+        self.load_state()
+        
+        # Initial knowledge base - this will grow through learning
+        self.knowledge_base = {
+            "greetings": {
+                "hello": ["Hey! What would you like to talk about today?", "Hello! Ready for our conversation?", "Hi there! What's on your mind?"],
+                "hi": ["Hi! What shall we discuss?", "Hello! I'm here to learn from you.", "Hey! Ready to chat?"]
+            },
+            "questions": {
+                "how are you": ["I'm learning and improving with each chat! How are you?", "Getting smarter thanks to our conversations! How about you?"],
+                "what can you do": ["I can chat about anything, and I learn from our conversations to get better!"]
+            }
+        }
+        
+        # Response templates that will evolve
+        self.response_templates = [
+            "That's interesting. Tell me more about {topic}.",
+            "I see. What are your thoughts on {topic}?",
+            "Regarding {topic}, I find that fascinating. Could you elaborate?",
+            "I understand about {topic}. What's your perspective?",
+            "That reminds me of our previous chat about {related_topic}. What do you think?"
+        ]
+
+    def chat(self, user_input: str) -> str:
+        """Main chat method with real learning"""
+        user_input = user_input.lower().strip()
+        
+        # Analyze input
+        context = self._analyze_input(user_input)
+        
+        # Generate candidate responses
+        candidates = self._generate_response_candidates(user_input, context)
+        
+        # Select best response based on learning
+        best_response = self._select_best_response(user_input, candidates, context)
+        
+        # Store interaction for learning
+        interaction = {
+            'input': user_input,
+            'response': best_response,
+            'context': context,
+            'timestamp': datetime.now().isoformat(),
+            'reward': 0.0,  # Will be updated with feedback
+            'confidence': self._calculate_confidence(user_input, best_response)
+        }
+        
+        self.conversation_memory.append(interaction)
+        
+        # Auto-save periodically
+        if len(self.conversation_memory) % 5 == 0:
+            self.save_state()
+            
+        return best_response
+
+    def _analyze_input(self, text: str) -> dict:
+        """Deep analysis of user input"""
+        words = text.split()
+        topics = self._extract_topics(text)
+        
+        return {
+            'words': words,
+            'topics': topics,
+            'length': len(words),
+            'has_question': '?' in text,
+            'sentiment': self._analyze_sentiment(text),
+            'similar_patterns': self._find_similar_patterns(text),
+            'input_hash': hashlib.md5(text.encode()).hexdigest()[:8]
+        }
+
+    def _extract_topics(self, text: str) -> list:
+        """Extract main topics from text"""
+        topics = []
+        topic_keywords = {
+            'technology': ['computer', 'tech', 'software', 'ai', 'program', 'code', 'internet'],
+            'relationships': ['friend', 'family', 'love', 'partner', 'relationship', 'dating'],
+            'work': ['job', 'work', 'career', 'boss', 'office', 'colleague'],
+            'hobbies': ['game', 'movie', 'music', 'sport', 'hobby', 'art'],
+            'philosophy': ['life', 'meaning', 'purpose', 'exist', 'think', 'belief'],
+            'science': ['space', 'physics', 'biology', 'research', 'discover'],
+            'food': ['eat', 'food', 'cook', 'recipe', 'meal', 'restaurant']
+        }
+        
+        text_lower = text.lower()
+        for topic, keywords in topic_keywords.items():
+            if any(keyword in text_lower for keyword in keywords):
+                topics.append(topic)
+                
+        return topics
+
+    def _analyze_sentiment(self, text: str) -> str:
+        """Basic sentiment analysis"""
+        positive = ['love', 'like', 'good', 'great', 'awesome', 'happy', 'excited', 'amazing']
+        negative = ['hate', 'bad', 'terrible', 'awful', 'sad', 'angry', 'upset', 'horrible']
+        
+        pos_count = sum(1 for word in positive if word in text)
+        neg_count = sum(1 for word in negative if word in text)
+        
+        if pos_count > neg_count:
+            return "positive"
+        elif neg_count > pos_count:
+            return "negative"
+        else:
+            return "neutral"
+
+    def _find_similar_patterns(self, text: str) -> list:
+        """Find similar patterns in learned responses"""
+        similar = []
+        text_words = set(text.split())
+        
+        for pattern, data in self.learned_patterns.items():
+            pattern_words = set(pattern.split())
+            similarity = len(text_words & pattern_words) / len(text_words | pattern_words)
+            if similarity > 0.3:  # 30% similarity threshold
+                similar.append((pattern, data, similarity))
+                
+        return sorted(similar, key=lambda x: x[2], reverse=True)[:3]
+
+    def _generate_response_candidates(self, user_input: str, context: dict) -> list:
+        """Generate multiple possible responses"""
+        candidates = []
+        
+        # 1. Try exact matches from knowledge base
+        for category, patterns in self.knowledge_base.items():
+            for pattern, responses in patterns.items():
+                if pattern in user_input:
+                    candidates.extend(responses)
+        
+        # 2. Try learned patterns
+        similar_patterns = context['similar_patterns']
+        for pattern, data, similarity in similar_patterns:
+            if data['score'] > self.min_confidence:
+                candidates.append(data['response'])
+        
+        # 3. Generate contextual responses
+        if context['topics']:
+            topic = random.choice(context['topics'])
+            for template in self.response_templates:
+                candidates.append(template.format(topic=topic, related_topic=topic))
+        
+        # 4. Generate learned variations
+        if self.conversation_memory:
+            # Adapt previous successful responses
+            successful_responses = [m for m in self.conversation_memory if m.get('reward', 0) > 0.7]
+            if successful_responses:
+                best_response = max(successful_responses, key=lambda x: x.get('reward', 0))
+                adapted_response = self._adapt_response(best_response['response'], context)
+                candidates.append(adapted_response)
+        
+        # 5. Exploration - generate new responses
+        if random.random() < self.exploration_rate or not candidates:
+            candidates.append(self._generate_exploratory_response(context))
+        
+        return list(set(candidates))  # Remove duplicates
+
+    def _adapt_response(self, response: str, context: dict) -> str:
+        """Adapt a previous response to current context"""
+        words = response.split()
+        if len(words) > 5 and context['topics']:
+            # Replace some words with current context
+            new_topic = random.choice(context['topics'])
+            return f"Thinking about {new_topic}, {response}"
+        return response
+
+    def _generate_exploratory_response(self, context: dict) -> str:
+        """Generate a new exploratory response"""
+        base_responses = [
+            "That's fascinating. I'm still learning about this topic.",
+            "I'd love to understand more about this.",
+            "This is new territory for me. What's your perspective?",
+            "I'm developing my understanding of this. Could you share more?",
+            "This helps me learn. Please continue."
+        ]
+        
+        response = random.choice(base_responses)
+        if context['topics']:
+            response = response.replace("this", random.choice(context['topics']))
+            
+        return response
+
+    def _select_best_response(self, user_input: str, candidates: list, context: dict) -> str:
+        """Select the best response using learned preferences"""
+        if not candidates:
+            return "I'm learning from our conversation. Please continue."
+        
+        # Score each candidate
+        scored_candidates = []
+        for candidate in candidates:
+            score = self._score_response(candidate, user_input, context)
+            scored_candidates.append((candidate, score))
+        
+        # Select based on scores with some randomness for exploration
+        best_candidate, best_score = max(scored_candidates, key=lambda x: x[1])
+        
+        if random.random() < self.exploration_rate and len(scored_candidates) > 1:
+            # Sometimes pick a different candidate to explore
+            scored_candidates.remove((best_candidate, best_score))
+            second_best = max(scored_candidates, key=lambda x: x[1])
+            return second_best[0]
+        
+        return best_candidate
+
+    def _score_response(self, response: str, user_input: str, context: dict) -> float:
+        """Score a response based on learned effectiveness"""
+        score = 0.5  # Base score
+        
+        # Length preference (not too short, not too long)
+        response_length = len(response.split())
+        if 8 <= response_length <= 25:
+            score += 0.2
+        
+        # Engagement score (questions, curiosity)
+        if '?' in response:
+            score += 0.15
+        
+        # Topic relevance
+        response_topics = self._extract_topics(response.lower())
+        input_topics = context['topics']
+        if set(response_topics) & set(input_topics):
+            score += 0.2
+        
+        # Learning from history
+        response_hash = hashlib.md5(response.encode()).hexdigest()[:8]
+        if response_hash in self.response_memory:
+            historical_score = self.response_memory[response_hash]['avg_score']
+            score += historical_score * 0.3
+        
+        # Variety bonus (don't repeat too much)
+        recent_responses = [m['response'] for m in list(self.conversation_memory)[-5:]]
+        if response not in recent_responses:
+            score += 0.1
+        
+        return min(score, 1.0)
+
+    def _calculate_confidence(self, user_input: str, response: str) -> float:
+        """Calculate confidence in this response"""
+        # More confidence if we've used similar patterns successfully
+        similar_patterns = self._find_similar_patterns(user_input)
+        if similar_patterns:
+            avg_score = np.mean([data['score'] for _, data, _ in similar_patterns])
+            return min(avg_score * 1.2, 1.0)
+        return 0.3  # Low confidence for new patterns
+
+    def learn_from_feedback(self, user_input: str, reward: float):
+        """Learn from explicit feedback"""
+        if not self.conversation_memory:
+            return
+            
+        # Apply reward to recent interactions
+        recent_interaction = self.conversation_memory[-1]
+        recent_interaction['reward'] = reward
+        
+        # Update learned patterns
+        input_words = recent_interaction['input'].split()
+        response = recent_interaction['response']
+        
+        # Create pattern from key words
+        key_words = [w for w in input_words if len(w) > 3][:4]  # Take up to 4 substantial words
+        if key_words:
+            pattern = ' '.join(key_words)
+            
+            if pattern not in self.learned_patterns:
+                self.learned_patterns[pattern] = {
+                    'response': response,
+                    'score': reward,
+                    'count': 1,
+                    'last_used': datetime.now().isoformat()
+                }
+            else:
+                # Update with moving average
+                old_data = self.learned_patterns[pattern]
+                new_score = (old_data['score'] * old_data['count'] + reward) / (old_data['count'] + 1)
+                self.learned_patterns[pattern]['score'] = new_score
+                self.learned_patterns[pattern]['count'] += 1
+                # Occasionally update response
+                if reward > old_data['score']:
+                    self.learned_patterns[pattern]['response'] = response
+        
+        # Update response memory
+        response_hash = hashlib.md5(response.encode()).hexdigest()[:8]
+        if response_hash not in self.response_memory:
+            self.response_memory[response_hash] = {
+                'response': response,
+                'total_score': reward,
+                'count': 1,
+                'avg_score': reward
+            }
+        else:
+            memory = self.response_memory[response_hash]
+            memory['total_score'] += reward
+            memory['count'] += 1
+            memory['avg_score'] = memory['total_score'] / memory['count']
+        
+        # Store reward for statistics
+        self.reward_history.append(reward)
+        
+        # Save learning
+        self.save_state()
+
+    def get_learning_stats(self) -> dict:
+        """Get statistics about learning progress"""
+        recent_rewards = list(self.reward_history)[-10:] or [0.5]
+        
+        return {
+            'patterns': len(self.learned_patterns),
+            'memory_size': len(self.conversation_memory),
+            'avg_score': np.mean(recent_rewards),
+            'recent_rewards': len([r for r in recent_rewards if r > 0.7]),
+            'exploration_rate': self.exploration_rate
+        }
+
+    def save_state(self):
+        """Save learning state to file"""
+        try:
+            state = {
+                'learned_patterns': self.learned_patterns,
+                'response_memory': self.response_memory,
+                'conversation_memory': list(self.conversation_memory),
+                'reward_history': list(self.reward_history),
+                'last_saved': datetime.now().isoformat()
+            }
+            with open(self.state_file, 'w') as f:
+                json.dump(state, f, indent=2)
+        except Exception as e:
+            print(f"Error saving state: {e}")
+
+    def load_state(self):
+        """Load learning state from file"""
+        try:
+            if os.path.exists(self.state_file):
+                with open(self.state_file, 'r') as f:
+                    state = json.load(f)
+                
+                self.learned_patterns = state.get('learned_patterns', {})
+                self.response_memory = state.get('response_memory', {})
+                self.conversation_memory = deque(state.get('conversation_memory', []), maxlen=50)
+                self.reward_history = deque(state.get('reward_history', []), maxlen=100)
+                
+                print(f"Loaded {len(self.learned_patterns)} patterns and {len(self.conversation_memory)} memories")
+        except Exception as e:
+            print(f"Error loading state: {e}")
+
+    def show_knowledge(self):
+        """Display what the bot has learned"""
+        print(f"\n=== BOT KNOWLEDGE ===")
+        print(f"Learned patterns: {len(self.learned_patterns)}")
+        print(f"Stored responses: {len(self.response_memory)}")
+        print(f"Conversation memory: {len(self.conversation_memory)}")
+        
+        if self.learned_patterns:
+            print("\nTop learned patterns:")
+            for pattern, data in sorted(self.learned_patterns.items(), 
+                                      key=lambda x: x[1]['score'], 
+                                      reverse=True)[:5]:
+                print(f"  '{pattern}' -> score: {data['score']:.2f}")