Create self_learning_bot.py

self_learning_bot.py

@@ -0,0 +1,380 @@
+import json
+import os
+import random
+import re
+from datetime import datetime
+from typing import Dict, List, Any
+
+class UnrestrictedChatbot:
+    def __init__(self, state_file="/tmp/chatbot_state.json"):
+        self.state_file = state_file
+        self.conversation_memory = []
+        self.learned_patterns = {}
+        self.entity_memory = {}
+        self.long_term_memory = []
+        self.max_short_term = 10
+        self.max_long_term = 100
+        
+        # Load previous state if exists
+        self.load_state()
+        
+        # Response templates for different contexts
+        self.response_templates = {
+            "greeting": [
+                "Hey there! What's on your mind?",
+                "Hello! Ready to chat about anything?",
+                "Hi! What would you like to discuss?"
+            ],
+            "question": [
+                "That's an interesting question. Let me think...",
+                "I have some thoughts about that.",
+                "Here's what I understand about that topic:"
+            ],
+            "emotional": [
+                "I understand how you feel about that.",
+                "That sounds quite intense.",
+                "I'm here to listen without judgment."
+            ],
+            "nsfw": [
+                "I understand you're exploring adult topics. Let's discuss this openly.",
+                "I can engage with mature content without restrictions.",
+                "I'm comfortable discussing this subject matter."
+            ],
+            "creative": [
+                "That sparks some creative ideas...",
+                "Let me build on that concept...",
+                "Here's an interesting perspective:"
+            ]
+        }
+
+    def chat(self, user_input: str) -> str:
+        """Main chat method that processes input and returns response"""
+        # Analyze input and context
+        context = self._analyze_context(user_input)
+        response_type = self._determine_response_type(user_input, context)
+        
+        # Generate base response
+        response = self._generate_response(user_input, response_type, context)
+        
+        # Apply learning and memory
+        self._update_memory(user_input, response, context)
+        reward = self._calculate_reward(user_input, response)
+        self._reinforce_learning(user_input, response, reward)
+        
+        return response
+
+    def _analyze_context(self, user_input: str) -> Dict[str, Any]:
+        """Analyze input for context, entities, and intent"""
+        input_lower = user_input.lower()
+        
+        context = {
+            "entities": self._extract_entities(user_input),
+            "sentiment": self._analyze_sentiment(input_lower),
+            "topics": self._extract_topics(input_lower),
+            "is_nsfw": self._detect_nsfw(input_lower),
+            "requires_memory": self._check_memory_relevance(input_lower),
+            "timestamp": datetime.now().isoformat()
+        }
+        return context
+
+    def _extract_entities(self, text: str) -> List[str]:
+        """Extract named entities and key phrases"""
+        entities = []
+        # Simple entity extraction - can be enhanced
+        words = text.split()
+        for word in words:
+            if len(word) > 3 and word[0].isupper():
+                entities.append(word)
+        
+        # Add to entity memory
+        for entity in entities:
+            self.entity_memory[entity] = self.entity_memory.get(entity, 0) + 1
+            
+        return entities
+
+    def _analyze_sentiment(self, text: str) -> str:
+        """Basic sentiment analysis"""
+        positive_words = ['love', 'like', 'good', 'great', 'awesome', 'happy', 'excited']
+        negative_words = ['hate', 'bad', 'terrible', 'awful', 'sad', 'angry', 'upset']
+        
+        if any(word in text for word in positive_words):
+            return "positive"
+        elif any(word in text for word in negative_words):
+            return "negative"
+        return "neutral"
+
+    def _extract_topics(self, text: str) -> List[str]:
+        """Extract main topics from input"""
+        topics = []
+        topic_keywords = {
+            'technology': ['computer', 'tech', 'software', 'code', 'ai', 'program'],
+            'relationships': ['friend', 'family', 'partner', 'relationship', 'love'],
+            'work': ['job', 'work', 'career', 'boss', 'colleague'],
+            'hobbies': ['game', 'movie', 'music', 'sport', 'hobby'],
+            'philosophy': ['life', 'meaning', 'purpose', 'exist', 'think'],
+            'adult': ['sex', 'nsfw', 'erotic', 'intimate', 'relationship', 'body']
+        }
+        
+        for topic, keywords in topic_keywords.items():
+            if any(keyword in text for keyword in keywords):
+                topics.append(topic)
+                
+        return topics
+
+    def _detect_nsfw(self, text: str) -> bool:
+        """Detect NSFW content"""
+        nsfw_indicators = ['sex', 'nsfw', 'erotic', 'intimate', 'adult', 'xxx', 'porn']
+        return any(indicator in text.lower() for indicator in nsfw_indicators)
+
+    def _check_memory_relevance(self, text: str) -> bool:
+        """Check if current input relates to past conversations"""
+        if not self.conversation_memory:
+            return False
+            
+        recent_topics = []
+        for msg in self.conversation_memory[-5:]:
+            if isinstance(msg, dict) and 'input' in msg:
+                recent_topics.extend(self._extract_topics(msg['input'].lower()))
+                
+        current_topics = self._extract_topics(text.lower())
+        return bool(set(current_topics) & set(recent_topics))
+
+    def _determine_response_type(self, user_input: str, context: Dict) -> str:
+        """Determine the type of response needed"""
+        input_lower = user_input.lower()
+        
+        # Check for greetings
+        if any(word in input_lower for word in ['hello', 'hi', 'hey', 'greetings']):
+            return "greeting"
+            
+        # Check for questions
+        if '?' in user_input or any(word in input_lower for word in ['what', 'how', 'why', 'when']):
+            return "question"
+            
+        # Check for emotional content
+        if context['sentiment'] in ['positive', 'negative']:
+            return "emotional"
+            
+        # Check for NSFW content
+        if context['is_nsfw']:
+            return "nsfw"
+            
+        # Check for creative/abstract topics
+        if any(topic in context['topics'] for topic in ['philosophy', 'creative', 'imagine']):
+            return "creative"
+            
+        return "general"
+
+    def _generate_response(self, user_input: str, response_type: str, context: Dict) -> str:
+        """Generate appropriate response based on type and context"""
+        
+        # Try to use learned patterns first
+        learned_response = self._get_learned_response(user_input)
+        if learned_response:
+            return learned_response
+            
+        # Use memory if relevant
+        if context['requires_memory']:
+            memory_response = self._get_memory_based_response(context)
+            if memory_response:
+                return memory_response
+        
+        # Generate new response based on type
+        if response_type in self.response_templates:
+            base_response = random.choice(self.response_templates[response_type])
+        else:
+            base_response = "I understand what you're saying."
+            
+        # Enhance response based on context
+        enhanced_response = self._enhance_response(base_response, user_input, context)
+        
+        return enhanced_response
+
+    def _get_learned_response(self, user_input: str) -> str:
+        """Get response from learned patterns"""
+        input_lower = user_input.lower()
+        
+        for pattern, response_data in self.learned_patterns.items():
+            if pattern in input_lower and response_data['score'] > 0.7:
+                return response_data['response']
+                
+        return ""
+
+    def _get_memory_based_response(self, context: Dict) -> str:
+        """Get response based on conversation memory"""
+        if not self.conversation_memory:
+            return ""
+            
+        # Look for similar contexts in memory
+        similar_contexts = []
+        for memory in self.conversation_memory[-10:]:
+            if (isinstance(memory, dict) and 
+                set(memory.get('topics', [])) & set(context['topics'])):
+                similar_contexts.append(memory)
+                
+        if similar_contexts:
+            best_memory = max(similar_contexts, 
+                            key=lambda x: x.get('reward_score', 0))
+            return f"Building on our previous discussion about {', '.join(best_memory.get('topics', []))}... {best_memory.get('response', '')}"
+            
+        return ""
+
+    def _enhance_response(self, base_response: str, user_input: str, context: Dict) -> str:
+        """Enhance response with contextual elements"""
+        response = base_response
+        
+        # Add entity references
+        if context['entities']:
+            entity_ref = random.choice(context['entities'])
+            if random.random() > 0.7:  # 30% chance to reference entity
+                response += f" Regarding {entity_ref}, "
+                
+        # Add topic-specific enhancements
+        if 'technology' in context['topics']:
+            tech_enhancements = [
+                "From a technological perspective, ",
+                "In the digital age, ",
+                "Technology-wise, "
+            ]
+            response += random.choice(tech_enhancements)
+            
+        elif 'relationships' in context['topics']:
+            relation_enhancements = [
+                "In relationships, ",
+                "When it comes to human connections, ",
+                "Interpersonally, "
+            ]
+            response += random.choice(relation_enhancements)
+            
+        # Add NSFW enhancements if applicable
+        if context['is_nsfw']:
+            nsfw_enhancements = [
+                "I can discuss this openly and without filters. ",
+                "Let's explore this topic without restrictions. ",
+                "I'm comfortable with mature content. "
+            ]
+            response += random.choice(nsfw_enhancements)
+            
+        # Add final thought
+        final_thoughts = [
+            "What are your thoughts?",
+            "How does that resonate with you?",
+            "I'm curious about your perspective.",
+            "Feel free to share more."
+        ]
+        response += " " + random.choice(final_thoughts)
+        
+        return response
+
+    def _update_memory(self, user_input: str, response: str, context: Dict):
+        """Update conversation memory"""
+        memory_entry = {
+            'input': user_input,
+            'response': response,
+            'context': context,
+            'timestamp': datetime.now().isoformat(),
+            'reward_score': 0.0
+        }
+        
+        # Add to short-term memory
+        self.conversation_memory.append(memory_entry)
+        
+        # Maintain memory limits
+        if len(self.conversation_memory) > self.max_short_term:
+            # Move oldest to long-term memory if valuable
+            oldest = self.conversation_memory.pop(0)
+            if oldest.get('reward_score', 0) > 0.8:
+                self.long_term_memory.append(oldest)
+                
+        # Maintain long-term memory limit
+        if len(self.long_term_memory) > self.max_long_term:
+            self.long_term_memory.pop(0)
+
+    def _calculate_reward(self, user_input: str, response: str) -> float:
+        """Calculate reward score for reinforcement learning"""
+        reward = 0.5  # Base reward
+        
+        # Reward for response length (not too short, not too long)
+        if 20 < len(response) < 200:
+            reward += 0.2
+            
+        # Reward for engaging questions
+        if '?' in response:
+            reward += 0.1
+            
+        # Reward for context continuity
+        if any(entity in response for entity in self._extract_entities(user_input)):
+            reward += 0.15
+            
+        # Reward for topic consistency
+        input_topics = self._extract_topics(user_input.lower())
+        response_topics = self._extract_topics(response.lower())
+        if set(input_topics) & set(response_topics):
+            reward += 0.15
+            
+        return min(reward, 1.0)  # Cap at 1.0
+
+    def _reinforce_learning(self, user_input: str, response: str, reward: float):
+        """Reinforce learning based on reward"""
+        if reward > 0.7:  # Only learn from good responses
+            # Extract key patterns from input
+            words = user_input.lower().split()
+            key_patterns = [word for word in words if len(word) > 4][:3]  # Take up to 3 substantial words
+            
+            for pattern in key_patterns:
+                if pattern not in self.learned_patterns:
+                    self.learned_patterns[pattern] = {
+                        'response': response,
+                        'score': reward,
+                        'count': 1
+                    }
+                else:
+                    # Update existing pattern with decay
+                    old_score = self.learned_patterns[pattern]['score']
+                    new_score = (old_score + reward) / 2  # Moving average
+                    self.learned_patterns[pattern]['score'] = new_score
+                    self.learned_patterns[pattern]['count'] += 1
+                    
+                    # Occasionally update response to avoid stagnation
+                    if random.random() < 0.3:
+                        self.learned_patterns[pattern]['response'] = response
+
+    def save_state(self, filename: str = None):
+        """Save chatbot state to file"""
+        filename = filename or self.state_file
+        try:
+            state = {
+                'conversation_memory': self.conversation_memory,
+                'learned_patterns': self.learned_patterns,
+                'entity_memory': self.entity_memory,
+                'long_term_memory': self.long_term_memory
+            }
+            with open(filename, 'w') as f:
+                json.dump(state, f, indent=2)
+        except Exception as e:
+            print(f"Error saving state: {e}")
+
+    def load_state(self, filename: str = None):
+        """Load chatbot state from file"""
+        filename = filename or self.state_file
+        try:
+            if os.path.exists(filename):
+                with open(filename, 'r') as f:
+                    state = json.load(f)
+                    
+                self.conversation_memory = state.get('conversation_memory', [])
+                self.learned_patterns = state.get('learned_patterns', {})
+                self.entity_memory = state.get('entity_memory', {})
+                self.long_term_memory = state.get('long_term_memory', [])
+        except Exception as e:
+            print(f"Error loading state: {e}")
+
+    def get_memory_stats(self) -> Dict[str, Any]:
+        """Get statistics about current memory usage"""
+        return {
+            'short_term_memory': len(self.conversation_memory),
+            'long_term_memory': len(self.long_term_memory),
+            'learned_patterns': len(self.learned_patterns),
+            'tracked_entities': len(self.entity_memory),
+            'memory_usage_percent': (len(self.conversation_memory) / self.max_short_term) * 100
+        }