Upload 3 files

model_Custm.py

@@ -708,27 +708,54 @@ class Wildnerve_tlm01(nn.Module, AbstractModel):
             # Handle prompt if provided (convert to input_ids)
             if prompt is not None and input_ids is None:
                 if self.tokenizer is not None:
-                    inputs = self.tokenizer(prompt, return_tensors="pt", truncation=True, padding=True)
-                    input_ids = inputs.input_ids
+                    # Check if tokenizer is directly callable
+                    if callable(self.tokenizer):
+                        inputs = self.tokenizer(prompt, return_tensors="pt", truncation=True, padding=True)
+                    # Check for encode method (common in TokenizerWrapper implementations)
+                    elif hasattr(self.tokenizer, "encode"):
+                        tokens = self.tokenizer.encode(prompt)
+                        # Convert to tensor if needed
+                        if isinstance(tokens, list):
+                            input_ids = torch.tensor([tokens], dtype=torch.long)
+                        else:
+                            input_ids = tokens.unsqueeze(0) if tokens.dim() == 1 else tokens
+                    # Check for tokenize method
+                    elif hasattr(self.tokenizer, "tokenize"):
+                        tokens = self.tokenizer.tokenize(prompt)
+                        if hasattr(self.tokenizer, "convert_tokens_to_ids"):
+                            token_ids = self.tokenizer.convert_tokens_to_ids(tokens)
+                            input_ids = torch.tensor([token_ids], dtype=torch.long)
+                    else:
+                        raise ValueError(f"Tokenizer type {type(self.tokenizer)} doesn't support required methods")
                 else:
                     # Try to get tokenizer from registry
                     from service_registry import registry, TOKENIZER
+                    from transformers import AutoTokenizer
+                    
+                    # Try to get from registry first
                     tokenizer = registry.get(TOKENIZER)
-                    if tokenizer:
+                    
+                    # If not available, create a new one
+                    if not tokenizer:
+                        tokenizer = AutoTokenizer.from_pretrained("gpt2")
+                        
+                    # Now use the tokenizer safely
+                    if callable(tokenizer):
                         inputs = tokenizer(prompt, return_tensors="pt", truncation=True, padding=True)
                         input_ids = inputs.input_ids
-                    else:
-                        raise ValueError("No tokenizer available to process prompt")
+                    elif hasattr(tokenizer, "encode"):
+                        tokens = tokenizer.encode(prompt)
+                        input_ids = torch.tensor([tokens], dtype=torch.long) if isinstance(tokens, list) else tokens
             
             # Check if we have valid input_ids at this point
             if input_ids is None:
                 raise ValueError("Either prompt or input_ids must be provided")
                 
             # Now continue with original generate implementation that uses input_ids
-            # ...existing implementation...
+            # ...existing code...
             
             # Simple fallback if no implementation exists
-            return f"I processed your request about '{prompt[:30]}...' successfully."
+            return f"I processed your request about '{prompt[:30] if prompt else 'your input'}...' successfully."
             
         except Exception as e:
             logger.error(f"Error in generate: {e}")

model_PrTr.py

@@ -25,8 +25,15 @@ logger = logging.getLogger(__name__)
 # Positional Encoding Module (for decoder)
 # ----------------------------
 class PositionalEncoding(nn.Module):
-    def __init__(self, d_model: int, max_len: int = app_config.MAX_SEQ_LENGTH):
+    def __init__(self, d_model: int, max_len: Optional[int] = None):
         super().__init__()
+        # Get MAX_SEQ_LENGTH safely from config
+        if max_len is None:
+            if hasattr(app_config, "TRANSFORMER_CONFIG") and isinstance(app_config.TRANSFORMER_CONFIG, dict):
+                max_len = app_config.TRANSFORMER_CONFIG.get("MAX_SEQ_LENGTH", 1024)
+            else:
+                max_len = 1024  # Safe default
+                
         pe = torch.zeros(max_len, d_model)
         position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
         div_term = torch.exp(torch.arange(0, d_model, 2, dtype=torch.float) * (-math.log(10000.0) / d_model))

tokenizer.py

@@ -22,115 +22,55 @@ from service_registry import registry, TOKENIZER
 logger = logging.getLogger(__name__)
 
 class TokenizerWrapper:
-    """A simple wrapper around AutoTokenizer to standardize tokenizer usage."""
-    def __init__(self, 
-                 primary_model: str = "Wildnerve-tlm01-0.05Bx12", 
-                 fallback_model: str = "bert-base-uncased", 
-                 fallback2_model: str = "gpt2", 
-                 sp_model_path: str = None):
-        # Use a robust, multi-fallback initialization
-        self.primary_model = primary_model
-        self.fallback_model = fallback_model
-        self.fallback2_model = fallback2_model
-        self.sp_model_path = sp_model_path
-        self.sp = None
-        self.tokenizer = None
-        # Advanced feature flags
-        self.features = {
-            "normalize_text": True,
-            "custom_preprocessing": True,
-            "multi_fallback": True
-        }
-        self.initialize_tokenizer()
-
-    def initialize_tokenizer(self):
-        """Initialize the tokenizer with proper error handling"""
-        # First, try to load SentencePiece model if provided
-        if self.sp_model_path and SP_AVAILABLE:
-            try:
-                self.sp = spm.SentencePieceProcessor()
-                self.sp.Load(self.sp_model_path)
-                logger.info(f"Loaded SentencePiece model from {self.sp_model_path}")
-            except Exception as e:
-                logger.warning(f"Failed to load SentencePiece model: {e}")
-                self.sp = None
-        else:
-            if not SP_AVAILABLE and self.sp_model_path:
-                logger.warning("SentencePiece is not installed; skipping SP model loading")
-            self.sp = None
-
-        # Next, attempt to load the primary tokenizer
+    """A wrapper for transformer tokenizers with fallbacks"""
+    
+    def __init__(self, model_name="gpt2"):
+        self.model_name = model_name
         try:
-            self.tokenizer = AutoTokenizer.from_pretrained(self.primary_model)
-            logger.info(f"Loaded primary tokenizer: {self.primary_model}")
+            self.tokenizer = AutoTokenizer.from_pretrained(model_name)
+            # Add pad token if it doesn't exist (important for GPT-2)
+            if self.tokenizer.pad_token is None:
+                self.tokenizer.pad_token = self.tokenizer.eos_token
+            logger.info(f"Initialized tokenizer from {model_name}")
         except Exception as e:
-            logger.warning(f"Primary tokenizer '{self.primary_model}' load failed: {e}")
-            try:
-                self.tokenizer = BertTokenizer.from_pretrained(self.fallback_model)
-                logger.info(f"Loaded fallback tokenizer: {self.fallback_model}")
-            except Exception as e2:
-                logger.warning(f"Fallback tokenizer '{self.fallback_model}' load failed: {e2}")
-                try:
-                    self.tokenizer = AutoTokenizer.from_pretrained(self.fallback2_model)
-                    logger.info(f"Loaded second fallback tokenizer: {self.fallback2_model}")
-                except Exception as e3:
-                    logger.error(f"All tokenizer loads failed: {e3}")
-                    self.tokenizer = None
-
-        if self.tokenizer:
-            registry.register(TOKENIZER, self.tokenizer)
-
-    def advanced_normalize(self, text: str) -> str:
-        # Advanced normalization: lowercasing and removing extra spaces
-        normalized = text.strip().lower()
-        normalized = " ".join(normalized.split())
-        return normalized
-
-    def tokenize(self, text: str, use_sentencepiece: bool = False) -> list:
-        """
-        Tokenize text robustly using SentencePiece if requested and available;
-        Otherwise use the transformer tokenizer; fallback to simple split if needed.
-        """
+            logger.error(f"Error loading tokenizer: {e}")
+            self.tokenizer = None
+            
+    def __call__(self, text, **kwargs):
+        """Make the wrapper callable like a standard HF tokenizer"""
+        if self.tokenizer is None:
+            raise ValueError("Tokenizer not initialized")
+        return self.tokenizer(text, **kwargs)
+    
+    def encode(self, text, **kwargs):
+        """Encode text to token IDs"""
+        if self.tokenizer is None:
+            raise ValueError("Tokenizer not initialized")
+        return self.tokenizer.encode(text, **kwargs)
+        
+    def decode(self, token_ids, **kwargs):
+        """Decode token IDs to text"""
+        if self.tokenizer is None:
+            raise ValueError("Tokenizer not initialized")
+        return self.tokenizer.decode(token_ids, **kwargs)
+        
+    def tokenize(self, text, **kwargs):
+        """Tokenize text to tokens"""
+        if self.tokenizer is None:
+            raise ValueError("Tokenizer not initialized")
+        return self.tokenizer.tokenize(text, **kwargs)
+    
+def get_tokenizer(model_name="gpt2"):
+    """Get a tokenizer instance with proper fallback handling"""
+    try:
+        return TokenizerWrapper(model_name)
+    except Exception as e:
+        logger.error(f"Error creating TokenizerWrapper: {e}")
         try:
-            # Apply text normalization if enabled
-            if self.features["normalize_text"]:
-                text = self.advanced_normalize(text)
-            if use_sentencepiece and self.sp:
-                tokens = self.sp.EncodeAsPieces(text)
-                logger.debug("Tokenized text using SentencePiece")
-                return tokens
-            elif self.tokenizer:
-                tokens = self.tokenizer.tokenize(text)
-                logger.debug("Tokenized text using transformer tokenizer")
-                # Optional custom preprocessing: filter out empty tokens
-                if self.features["custom_preprocessing"]:
-                    tokens = [tok for tok in tokens if tok.strip()]
-                return tokens
-            else:
-                raise ValueError("No tokenizer available")
-        except Exception as ex:
-            logger.error(f"Tokenization failed: {ex}")
-            return text.split()  # fallback
-
-    def encode(self, text: str, **kwargs) -> list:
-        try:
-            if self.tokenizer:
-                return self.tokenizer.encode(text, **kwargs)
-            else:
-                raise ValueError("Tokenizer not initialized")
-        except Exception as e:
-            logger.error(f"Encoding error: {e}")
-            return []
-
-    def decode(self, token_ids: list) -> str:
-        try:
-            return self.tokenizer.decode(token_ids, skip_special_tokens=True, clean_up_tokenization_spaces=True)
-        except Exception as ex:
-            logger.error(f"Decoding failed: {ex}")
-            return "Decoding error"
-
-def get_tokenizer(model_name: str = "bert-base-uncased") -> TokenizerWrapper:
-    return TokenizerWrapper(model_name)
+            return AutoTokenizer.from_pretrained(model_name)
+        except Exception as e2:
+            logger.error(f"Error loading AutoTokenizer: {e2}")
+            return None
 
 if __name__ == "__main__":
     # Example usage showcasing advanced features