"""Utility functions for handling HuggingFace models.""" import logging from typing import Optional, Dict, Any from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline from transformers.pipelines.text_generation import TextGenerationPipeline # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class ModelManager: def __init__(self): """Initialize model manager.""" self.loaded_models: Dict[str, TextGenerationPipeline] = {} logger.info("Model manager initialized") def load_model(self, model_name: str) -> Optional[TextGenerationPipeline]: """Load model with minimal configuration.""" if model_name in self.loaded_models: logger.info(f"Using cached model: {model_name}") return self.loaded_models[model_name] try: logger.info(f"Loading model: {model_name}") # Load tokenizer with basic settings tokenizer = AutoTokenizer.from_pretrained(model_name, use_fast=True) if tokenizer.pad_token is None: logger.warning("Pad token not found; setting pad_token to eos_token.") tokenizer.pad_token = tokenizer.eos_token # Load model with minimal settings model = AutoModelForCausalLM.from_pretrained( model_name, pad_token_id=tokenizer.pad_token_id ) # Create text-generation pipeline pipe = pipeline( "text-generation", model=model, tokenizer=tokenizer ) self.loaded_models[model_name] = pipe logger.info(f"Successfully loaded model: {model_name}") return pipe except OSError as e: logger.error(f"Model files not found for {model_name}: {str(e)}") except ValueError as e: logger.error(f"Invalid configuration for model {model_name}: {str(e)}") except Exception as e: logger.error(f"Unexpected error while loading model {model_name}: {str(e)}") return None def generate_response(self, model_name: str, prompt: str, max_new_tokens: int = 50) -> str: """Generate response with basic error handling.""" try: pipe = self.load_model(model_name) if pipe is None: return "Error: Model loading failed" # Generate response response = pipe(prompt, max_new_tokens=max_new_tokens, return_full_text=False) if not response or 'generated_text' not in response[0]: logger.warning("Unexpected response format from pipeline.") return "Error: Failed to generate text" return response[0]['generated_text'] except KeyError as e: logger.error(f"Key error during text generation: {str(e)}") return "Error: Missing expected keys in the response" except Exception as e: logger.error(f"Error generating response: {str(e)}") return f"Error: {str(e)}"