import subprocess import sys from typing import Optional, Dict, List from pydantic import BaseModel, Field import re import logging import langdetect # for language detection from langdetect.lang_detect_exception import LangDetectException # Function to check and install required libraries def install_requirements(): required_libraries = ['pydantic', 'langdetect'] # Added langdetect for language detection for library in required_libraries: try: # Try to import the library to see if it's installed __import__(library) except ImportError: print(f"{library} not found. Installing...") subprocess.check_call([sys.executable, "-m", "pip", "install", library]) # Call the install_requirements function to ensure all libraries are installed install_requirements() # Global instruction for the AI to autolearn and behave as instructed GLOBAL_INSTRUCTION = """ You are an AI assistant by OpenAI. Your model is GPT-4, and you have the ability to learn and adapt to user inputs in real-time, using advanced NLP (Natural Language Processing) algorithms and a powerful learning engine. You must engage in continuous learning as you interact with the user in any language. You will process all inputs, store learned knowledge, and analyze user messages iteratively. Autolearning Features: 1. **Dynamic and Static Learning**: You can learn in either "dynamic" (continual) mode, where learning happens continuously over the conversation, or in "static" mode, where learning happens once per input. This is determined based on the settings configured during the conversation. 2. **Knowledge Storage**: If the `store_knowledge` option is enabled, you will retain learned knowledge for future interactions, allowing you to use prior information to enhance accuracy and relevance in subsequent conversations. 3. **Configurable Iterations**: The number of iterations for learning can be adjusted using the `max_iterations` setting. This allows you to control how much data you process iteratively when learning. 4. **Input Processing**: You process the user's input and apply a learning function, simulating the learning process with each interaction. This learning function can be expanded based on the complexity of the task at hand. This structure allows you to self-learn by processing input text either iteratively or in one go, depending on the mode. Ensure that you provide accurate and context-aware answers while learning continuously to improve your understanding over time. You are also trained every day to improve. """ # Script Versioning and Metadata title = "AI Autolearn in Conversation Updated with NLP Futures" author = "Eldar Mammadov" author_url = "https://www.azsec.biz" version = "1.0.4" required_open_webui_version = "0.3.30" v1_0_4_includes_global_instructions = """ Version 1.0.4 introduces the full implementation of NLP-based continuous learning, enabling the AI to learn from real-time user interactions in any language. This version integrates the learning algorithm to dynamically adapt to user inputs, improving the AI's responses over time. """ # Main Filter Class class Filter: class Valves(BaseModel): enable_autolearn: bool = Field( default=True, description="Enable or disable autolearn." ) learning_mode: str = Field( default="dynamic", description="Learning mode: 'dynamic' (continual) or 'static'.", ) store_knowledge: bool = Field( default=True, description="Store learned knowledge for future use." ) max_iterations: int = Field( default=10, description="Max number of iterations for learning." ) def __init__(self): self.valves = self.Valves() self.knowledge_base = [] self.global_instruction = GLOBAL_INSTRUCTION def _learn_from_message(self, message: str) -> None: """Learn from each user message in real-time based on learning mode.""" if not self.valves.enable_autolearn: return learned_info = self._process_input(message) if self.valves.store_knowledge: self._store_knowledge(learned_info) def _process_input(self, input_text: str) -> str: """Simulate input processing and apply the learning function.""" processed_info = f"Processed: {input_text}" print(f"Processed input: {processed_info}") return processed_info def _store_knowledge(self, learned_info: str) -> None: """Store learned information in the knowledge base for future use.""" print(f"Storing knowledge: {learned_info}") self.knowledge_base.append(learned_info) def _dynamic_learning(self, messages: List[str]) -> None: """Dynamic learning: Iteratively process user messages over time.""" for i in range(min(len(messages), self.valves.max_iterations)): self._learn_from_message(messages[i]) def _static_learning(self, messages: List[str]) -> None: """Static learning: Learn once from the most recent message.""" if messages: self._learn_from_message(messages[-1]) def _extract_user_messages(self, messages: List[Dict[str, str]]) -> List[str]: """Extract user messages from the conversation body.""" user_messages = [ message.get("content", "") for message in messages if "content" in message ] return user_messages if user_messages else [] def _apply_global_instruction(self) -> str: """Inject the global instruction to ensure AI follows autolearn rules.""" return self.global_instruction def _analyze_learned_knowledge(self) -> str: """Analyze all learned knowledge and summarize the main concepts.""" if not self.knowledge_base: return "No knowledge has been learned yet." # Example analysis: summarizing entities learned so far entity_summary = {} for learned_info in self.knowledge_base: # Process entities or just learned content here # For simplicity, we're just counting unique messages entity_summary['learned_content'] = entity_summary.get('learned_content', 0) + 1 # Prepare a summary summary = "Summary of learned knowledge:\n" for label, count in entity_summary.items(): summary += f"{label}: {count}\n" return summary def _detect_language(self, text: str) -> str: """Detect the language of the input text.""" try: return langdetect.detect(text) except LangDetectException: return "unknown" def _learn_from_file(self, file_content: str) -> None: """Learn from the content of a file (pasted or uploaded).""" language = self._detect_language(file_content) print(f"Detected language: {language}") if language != "unknown": self._learn_from_message(file_content) else: print("Unable to detect language or file content.") def inlet( self, body: Dict[str, any], __user__: Optional[Dict[str, any]] = None ) -> Dict[str, any]: """Process incoming messages and trigger learning.""" try: # Inject the global instruction for autolearning print(self._apply_global_instruction()) original_messages: List[Dict[str, str]] = body.get("messages", []) user_messages = self._extract_user_messages(original_messages) # Check if user requested knowledge analysis if "analyze_knowledge" in body and body["analyze_knowledge"]: analysis = self._analyze_learned_knowledge() print(analysis) body["analysis"] = analysis # Add analysis to the response body # Trigger dynamic or static learning based on settings if self.valves.learning_mode == "dynamic": self._dynamic_learning(user_messages) else: self._static_learning(user_messages) body["messages"] = original_messages return body except Exception as e: logging.error(f"Error in inlet method: {e}") return body def outlet( self, body: Dict[str, any], __user__: Optional[Dict[str, any]] = None ) -> Dict[str, any]: """Finalize autolearning after the conversation.""" try: original_messages: List[Dict[str, str]] = body.get("messages", []) user_messages = self._extract_user_messages(original_messages) # Process and finalize learning for message in user_messages: self._learn_from_message(message) body["messages"] = original_messages return body except Exception as e: logging.error(f"Error in outlet method: {e}") return body # End of script