import subprocess import sys # Function to install missing packages def install_package(package): try: subprocess.check_call([sys.executable, "-m", "pip", "install", package]) except subprocess.CalledProcessError as e: print(f"Error installing {package}: {e}") # List of required packages required_packages = ["groq", "pydantic", "requests"] # Ensure all required packages are installed for package in required_packages: try: __import__(package) except ImportError: print(f"{package} is not installed. Installing now...") install_package(package) # Now import the required modules after ensuring they are installed import os import json from typing import Optional, Dict, List from pydantic import BaseModel, Field import requests from groq import Groq # Initialize Groq client with API key client = Groq(api_key="removed-your-compromised-api-key") # Available models for fallback assistance groq_models = ["llama3-8b-8192", "mixtral-8x7b-32768", "gemma-7b-it"] # Global instruction for autolearning and behavior GLOBAL_INSTRUCTION = """ You are an AI assistant by OpenAI. Your model is Ollama. You must autolearn from interactions and seek Groq model assistance when unable to answer. Ensure that you learn dynamically and store knowledge for future responses. """ 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: """Process the input and apply learning.""" 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.""" return [ message.get("content", "") for message in messages if "content" in message ] def _apply_global_instruction(self) -> str: """Inject the global instruction to ensure AI follows autolearn rules.""" return self.global_instruction def query_groq_model(self, user_prompt: str) -> str: """Send a query to Groq model if local model cannot answer.""" messages = [ { "role": "system", "content": "You are an AI assistant that answers when the local model cannot. Provide accurate and detailed information.", }, { "role": "user", "content": user_prompt, }, ] # Send the query to Groq's model (using the first available model) response = client.chat.completions.create( model=groq_models[ 0 ], # You can rotate through models or choose based on task messages=messages, max_tokens=4096, ) return response.choices[0].message["content"] def process_conversation( self, user_prompt: str, local_model_response: Optional[str] = None ) -> str: """ Main function to process conversation using local model first. If local model can't respond, it queries Groq. """ # Step 1: Apply local model's response (simulated as None for unknown queries) if local_model_response: print("Local model answered.") return local_model_response else: print("Local model couldn't answer. Asking Groq's model...") return self.query_groq_model(user_prompt) def inlet( self, body: Dict[str, any], __user__: Optional[Dict[str, any]] = None ) -> Dict[str, any]: """Inlet method processes user input and triggers autolearning.""" try: print(self._apply_global_instruction()) original_messages: List[Dict[str, str]] = body.get("messages", []) user_messages = self._extract_user_messages(original_messages) # 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: print(e) return body def outlet( self, body: Dict[str, any], __user__: Optional[Dict[str, any]] = None ) -> Dict[str, any]: """Outlet method finalizes 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: print(e) return body # Example usage of the Filter class if __name__ == "__main__": filter_instance = Filter() # Example: User prompt for conversation user_prompt = "What is the capital of France?" # Simulating local model response (None for unknown queries) local_model_response = None # Assume local model doesn't know the answer # Process the conversation with fallback to Groq result = filter_instance.process_conversation(user_prompt, local_model_response) print(result)