Delete gemma_utils.py

gemma_utils.py

@@ -1,244 +0,0 @@
-# models/gemma/gemma_utils.py
-
-"""
-Gemma Model Utilities for PENNY Project
-Handles text generation using the Gemma-based core language model pipeline.
-Provides async generation with structured error handling and logging.
-"""
-
-import asyncio
-import time
-from typing import Dict, Any, Optional
-
-# --- Logging Imports ---
-from app.logging_utils import log_interaction, sanitize_for_logging
-
-# --- Model Loader Import ---
-try:
-    from app.model_loader import load_model_pipeline
-    MODEL_LOADER_AVAILABLE = True
-except ImportError:
-    MODEL_LOADER_AVAILABLE = False
-    import logging
-    logging.getLogger(__name__).warning("Could not import load_model_pipeline. Gemma service unavailable.")
-
-# Global variable to store the loaded pipeline for re-use
-GEMMA_PIPELINE: Optional[Any] = None
-AGENT_NAME = "penny-core-agent"
-INITIALIZATION_ATTEMPTED = False
-
-
-def _initialize_gemma_pipeline() -> bool:
-    """
-    Initializes the Gemma pipeline only once.
-    
-    Returns:
-        bool: True if initialization succeeded, False otherwise.
-    """
-    global GEMMA_PIPELINE, INITIALIZATION_ATTEMPTED
-    
-    if INITIALIZATION_ATTEMPTED:
-        return GEMMA_PIPELINE is not None
-    
-    INITIALIZATION_ATTEMPTED = True
-    
-    if not MODEL_LOADER_AVAILABLE:
-        log_interaction(
-            intent="gemma_initialization",
-            success=False,
-            error="model_loader unavailable"
-        )
-        return False
-    
-    try:
-        log_interaction(
-            intent="gemma_initialization",
-            success=None,
-            details=f"Loading {AGENT_NAME}"
-        )
-        
-        GEMMA_PIPELINE = load_model_pipeline(AGENT_NAME)
-        
-        if GEMMA_PIPELINE is None:
-            log_interaction(
-                intent="gemma_initialization",
-                success=False,
-                error="Pipeline returned None"
-            )
-            return False
-        
-        log_interaction(
-            intent="gemma_initialization",
-            success=True,
-            details=f"Model {AGENT_NAME} loaded successfully"
-        )
-        return True
-        
-    except Exception as e:
-        log_interaction(
-            intent="gemma_initialization",
-            success=False,
-            error=str(e)
-        )
-        return False
-
-
-# Attempt initialization at module load
-_initialize_gemma_pipeline()
-
-
-def is_gemma_available() -> bool:
-    """
-    Check if Gemma service is available.
-    
-    Returns:
-        bool: True if Gemma pipeline is loaded and ready.
-    """
-    return GEMMA_PIPELINE is not None
-
-
-async def generate_response(
-    prompt: str,
-    max_new_tokens: int = 256,
-    temperature: float = 0.7,
-    tenant_id: Optional[str] = None,
-) -> Dict[str, Any]:
-    """
-    Runs text generation using the loaded Gemma pipeline.
-
-    Args:
-        prompt: The conversational or instruction prompt.
-        max_new_tokens: The maximum number of tokens to generate (default: 256).
-        temperature: Controls randomness in generation (default: 0.7).
-        tenant_id: Optional tenant identifier for logging.
-
-    Returns:
-        A dictionary containing:
-            - response (str): The generated text
-            - available (bool): Whether the service was available
-            - error (str, optional): Error message if generation failed
-            - response_time_ms (int, optional): Generation time in milliseconds
-    """
-    start_time = time.time()
-    
-    global GEMMA_PIPELINE
-
-    # Check availability
-    if not is_gemma_available():
-        log_interaction(
-            intent="gemma_generate",
-            tenant_id=tenant_id,
-            success=False,
-            error="Gemma pipeline not available",
-            fallback_used=True
-        )
-        return {
-            "response": "I'm having trouble accessing my language model right now. Please try again in a moment!",
-            "available": False,
-            "error": "Pipeline not initialized"
-        }
-
-    # Validate inputs
-    if not prompt or not isinstance(prompt, str):
-        log_interaction(
-            intent="gemma_generate",
-            tenant_id=tenant_id,
-            success=False,
-            error="Invalid prompt provided"
-        )
-        return {
-            "response": "I didn't receive a valid prompt. Could you try again?",
-            "available": True,
-            "error": "Invalid input"
-        }
-
-    # Configure generation parameters
-    gen_kwargs = {
-        "max_new_tokens": max_new_tokens,
-        "temperature": temperature,
-        "do_sample": True if temperature > 0.0 else False,
-        "return_full_text": False
-    }
-
-    try:
-        loop = asyncio.get_event_loop()
-        
-        # Run model inference in thread executor
-        results = await loop.run_in_executor(
-            None,
-            lambda: GEMMA_PIPELINE(prompt, **gen_kwargs)
-        )
-        
-        response_time_ms = int((time.time() - start_time) * 1000)
-        
-        # Parse results
-        if results and isinstance(results, list) and len(results) > 0:
-            if isinstance(results[0], dict) and 'generated_text' in results[0]:
-                generated_text = results[0]['generated_text'].strip()
-                
-                # Log slow responses
-                if response_time_ms > 5000:
-                    log_interaction(
-                        intent="gemma_generate_slow",
-                        tenant_id=tenant_id,
-                        success=True,
-                        response_time_ms=response_time_ms,
-                        details="Slow generation detected"
-                    )
-                
-                log_interaction(
-                    intent="gemma_generate",
-                    tenant_id=tenant_id,
-                    success=True,
-                    response_time_ms=response_time_ms,
-                    prompt_preview=sanitize_for_logging(prompt[:100])
-                )
-                
-                return {
-                    "response": generated_text,
-                    "available": True,
-                    "response_time_ms": response_time_ms
-                }
-        
-        # Unexpected output format
-        log_interaction(
-            intent="gemma_generate",
-            tenant_id=tenant_id,
-            success=False,
-            error="Unexpected model output format",
-            response_time_ms=response_time_ms
-        )
-        
-        return {
-            "response": "I got an unexpected response from my language model. Let me try to help you another way!",
-            "available": True,
-            "error": "Unexpected output format"
-        }
-
-    except asyncio.CancelledError:
-        log_interaction(
-            intent="gemma_generate",
-            tenant_id=tenant_id,
-            success=False,
-            error="Generation cancelled"
-        )
-        raise
-        
-    except Exception as e:
-        response_time_ms = int((time.time() - start_time) * 1000)
-        
-        log_interaction(
-            intent="gemma_generate",
-            tenant_id=tenant_id,
-            success=False,
-            error=str(e),
-            response_time_ms=response_time_ms,
-            fallback_used=True
-        )
-        
-        return {
-            "response": "I'm having trouble generating a response right now. Please try again!",
-            "available": False,
-            "error": str(e),
-            "response_time_ms": response_time_ms
-        }