Update core/llm_clients.py

core/llm_clients.py

@@ -2,193 +2,125 @@
 import os
 import google.generativeai as genai
 from huggingface_hub import InferenceClient
-import time # For potential retries or delays
+import time
 
 # --- Configuration ---
-# These will be populated by os.getenv()
 GOOGLE_API_KEY = None
 HF_TOKEN = None
-
-# Status flags, default to False
 GEMINI_API_CONFIGURED = False
 HF_API_CONFIGURED = False
-
-# Client instances
 hf_inference_client = None
-# google_gemini_model_instances cache is not strictly necessary as genai.GenerativeModel is light.
-# Removing it for now to simplify, can be added back if model instantiation proves slow.
 
-# --- Initialization Function (to be called from app.py's global scope) ---
 def initialize_all_clients():
     global GOOGLE_API_KEY, HF_TOKEN, GEMINI_API_CONFIGURED, HF_API_CONFIGURED, hf_inference_client
-    
     print("INFO: llm_clients.py - Attempting to initialize all API clients...")
-
-    # Google Gemini
     GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")
-    if GOOGLE_API_KEY and GOOGLE_API_KEY.strip(): # Check if key is not None and not just whitespace
-        print("INFO: llm_clients.py - GOOGLE_API_KEY found in environment.")
+    if GOOGLE_API_KEY and GOOGLE_API_KEY.strip():
+        print("INFO: llm_clients.py - GOOGLE_API_KEY found.")
         try:
-            # Test configuration by making a very simple, non-resource-intensive call
-            # or by listing models if supported and cheap.
-            # For now, genai.configure() is the main check.
             genai.configure(api_key=GOOGLE_API_KEY)
-            # Optionally, try to list models or a similar lightweight check if genai.configure isn't enough
-            # models = [m for m in genai.list_models() if 'generateContent' in m.supported_generation_methods]
-            # if not models:
-            #    raise Exception("No usable Gemini models found with this API key, or API not fully enabled.")
             GEMINI_API_CONFIGURED = True
-            print("SUCCESS: llm_clients.py - Google Gemini API configured successfully.")
+            print("SUCCESS: llm_clients.py - Google Gemini API configured.")
         except Exception as e:
             GEMINI_API_CONFIGURED = False
-            print(f"ERROR: llm_clients.py - Failed to configure/validate Google Gemini API. Key value might be invalid, API not enabled in Google Cloud, or other issue.")
-            print(f"  Gemini Init Error Details: {type(e).__name__}: {e}")
+            print(f"ERROR: llm_clients.py - Failed to configure Google Gemini API: {type(e).__name__}: {e}")
     else:
-        GEMINI_API_CONFIGURED = False # Explicitly set if key is missing/empty
-        print("WARNING: llm_clients.py - GOOGLE_API_KEY not found or is empty in environment variables.")
+        GEMINI_API_CONFIGURED = False
+        print("WARNING: llm_clients.py - GOOGLE_API_KEY not found or empty.")
 
-    # Hugging Face
     HF_TOKEN = os.getenv("HF_TOKEN")
-    if HF_TOKEN and HF_TOKEN.strip(): # Check if token is not None and not just whitespace
-        print("INFO: llm_clients.py - HF_TOKEN found in environment.")
+    if HF_TOKEN and HF_TOKEN.strip():
+        print("INFO: llm_clients.py - HF_TOKEN found.")
         try:
             hf_inference_client = InferenceClient(token=HF_TOKEN)
-            # Optionally, you could try a very quick ping to a known small public model if client init isn't enough
-            # hf_inference_client.text_generation("ping", model="gpt2", max_new_tokens=1)
             HF_API_CONFIGURED = True
-            print("SUCCESS: llm_clients.py - Hugging Face InferenceClient initialized successfully.")
+            print("SUCCESS: llm_clients.py - Hugging Face InferenceClient initialized.")
         except Exception as e:
             HF_API_CONFIGURED = False
-            print(f"ERROR: llm_clients.py - Failed to initialize Hugging Face InferenceClient. Token might be invalid or other issue.")
-            print(f"  HF Init Error Details: {type(e).__name__}: {e}")
-            hf_inference_client = None # Ensure client is None on failure
+            print(f"ERROR: llm_clients.py - Failed to initialize HF InferenceClient: {type(e).__name__}: {e}")
+            hf_inference_client = None
     else:
-        HF_API_CONFIGURED = False # Explicitly set if token is missing/empty
-        print("WARNING: llm_clients.py - HF_TOKEN not found or is empty in environment variables.")
-    
-    print(f"INFO: llm_clients.py - Initialization complete. Gemini Configured: {GEMINI_API_CONFIGURED}, HF Configured: {HF_API_CONFIGURED}")
+        HF_API_CONFIGURED = False
+        print("WARNING: llm_clients.py - HF_TOKEN not found or empty.")
+    print(f"INFO: llm_clients.py - Init complete. Gemini Configured: {GEMINI_API_CONFIGURED}, HF Configured: {HF_API_CONFIGURED}")
 
+# --- Status Getter Functions ---
+def is_gemini_api_configured():
+    global GEMINI_API_CONFIGURED
+    return GEMINI_API_CONFIGURED
 
-# This class remains useful for standardizing responses
-class LLMResponse:
-    def __init__(self, text=None, error=None, success=True, raw_response=None, model_id_used="unknown"):
-        self.text = text
-        self.error = error
-        self.success = success
-        self.raw_response = raw_response
-        self.model_id_used = model_id_used
-
-    def __str__(self):
-        if self.success:
-            return str(self.text) if self.text is not None else "" # Ensure text is string
-        return f"ERROR (Model: {self.model_id_used}): {self.error}"
+def is_hf_api_configured():
+    global HF_API_CONFIGURED
+    return HF_API_CONFIGURED
 
+# ... (LLMResponse class and call_huggingface_api function remain the same as the last full version) ...
+class LLMResponse: # Make sure this is defined
+    def __init__(self, text=None, error=None, success=True, raw_response=None, model_id_used="unknown"):
+        self.text, self.error, self.success, self.raw_response, self.model_id_used = text, error, success, raw_response, model_id_used
+    def __str__(self): return str(self.text) if self.success and self.text is not None else f"ERROR (Model: {self.model_id_used}): {self.error}"
 
 def call_huggingface_api(prompt_text, model_id, temperature=0.7, max_new_tokens=512, system_prompt_text=None):
-    print(f"DEBUG: llm_clients.py - call_huggingface_api attempt for model: {model_id}")
-    if not HF_API_CONFIGURED or not hf_inference_client:
-        error_msg = "Hugging Face API not configured (HF_TOKEN missing, client init failed, or token invalid)."
+    print(f"DEBUG: llm_clients.py - call_huggingface_api for model: {model_id}")
+    if not is_hf_api_configured() or not hf_inference_client: # Use getter
+        error_msg = "HF API not configured."
         print(f"ERROR: llm_clients.py - {error_msg}")
         return LLMResponse(error=error_msg, success=False, model_id_used=model_id)
-
-    full_prompt = prompt_text
-    if system_prompt_text:
-        full_prompt = f"<s>[INST] <<SYS>>\n{system_prompt_text}\n<</SYS>>\n\n{prompt_text} [/INST]" # Llama-style
-
+    full_prompt = f"<s>[INST] <<SYS>>\n{system_prompt_text}\n<</SYS>>\n\n{prompt_text} [/INST]" if system_prompt_text else prompt_text
     try:
-        print(f"  HF API Call - Prompt (first 100 chars): {full_prompt[:100]}...")
-        use_sample = temperature > 0.001 
-        raw_response = hf_inference_client.text_generation(
-            full_prompt, model=model_id, max_new_tokens=max_new_tokens,
-            temperature=temperature if use_sample else None,
-            do_sample=use_sample, 
-            stream=False
-        )
-        print(f"  HF API Call - Success for model: {model_id}. Response (first 100 chars): {str(raw_response)[:100]}...")
+        print(f"  HF API Call - Prompt (first 100): {full_prompt[:100]}...")
+        use_sample = temperature > 0.001
+        raw_response = hf_inference_client.text_generation(full_prompt, model=model_id, max_new_tokens=max_new_tokens, temperature=temperature if use_sample else None, do_sample=use_sample, stream=False)
+        print(f"  HF API Call - Success for {model_id}. Response (first 100): {str(raw_response)[:100]}...")
         return LLMResponse(text=raw_response, raw_response=raw_response, model_id_used=model_id)
     except Exception as e:
-        error_msg = f"HF API Error during text_generation ({model_id}): {type(e).__name__} - {str(e)}"
+        error_msg = f"HF API Error ({model_id}): {type(e).__name__} - {str(e)}"
         print(f"ERROR: llm_clients.py - {error_msg}")
         return LLMResponse(error=error_msg, success=False, raw_response=e, model_id_used=model_id)
 
-def call_gemini_api(prompt_text, model_id, temperature=0.7, max_new_tokens=768, system_prompt_text=None):
-    print(f"DEBUG: llm_clients.py - call_gemini_api attempt for model: {model_id}")
-    if not GEMINI_API_CONFIGURED:
-        error_msg = "Google Gemini API not configured (GOOGLE_API_KEY missing, config failed, or key invalid)."
+
+def call_gemini_api(prompt_text, model_id, temperature=0.7, max_new_tokens=1024, system_prompt_text=None): # Increased default max_tokens
+    print(f"DEBUG: llm_clients.py - call_gemini_api for model: {model_id}")
+    if not is_gemini_api_configured(): # Use getter
+        error_msg = "Google Gemini API not configured."
         print(f"ERROR: llm_clients.py - {error_msg}")
         return LLMResponse(error=error_msg, success=False, model_id_used=model_id)
-
     try:
-        # genai.GenerativeModel is the recommended way to get a model instance.
-        # system_instruction is preferred for newer models (like 1.5 series).
-        print(f"  Gemini API Call - Getting model instance for: {model_id}")
-        model_instance = genai.GenerativeModel(
-            model_name=model_id,
-            system_instruction=system_prompt_text # Pass system prompt here
-        )
-        
-        generation_config = genai.types.GenerationConfig(
-            temperature=temperature,
-            max_output_tokens=max_new_tokens
-        )
-        
-        print(f"  Gemini API Call - Prompt (first 100 chars): {prompt_text[:100]}...")
-        if system_prompt_text: print(f"  Gemini API Call - System Prompt (first 100 chars): {system_prompt_text[:100]}...")
-
-        raw_response = model_instance.generate_content(
-            prompt_text, # User prompt directly if system_instruction is used
-            generation_config=generation_config,
-            stream=False
-        )
-        print(f"  Gemini API Call - Raw response received for model: {model_id}. Prompt feedback: {raw_response.prompt_feedback}, Candidates: {'Yes' if raw_response.candidates else 'No'}")
-
+        print(f"  Gemini API Call - Getting instance for: {model_id}")
+        model_instance = genai.GenerativeModel(model_name=model_id, system_instruction=system_prompt_text)
+        generation_config = genai.types.GenerationConfig(temperature=temperature, max_output_tokens=max_new_tokens)
+        print(f"  Gemini API Call - User Prompt (first 100): {prompt_text[:100]}...")
+        if system_prompt_text: print(f"  Gemini API Call - System Prompt (first 100): {system_prompt_text[:100]}...")
+        raw_response = model_instance.generate_content(prompt_text, generation_config=generation_config, stream=False)
+        print(f"  Gemini API Call - Raw response for {model_id}. Feedback: {raw_response.prompt_feedback}, Candidates: {'Yes' if raw_response.candidates else 'No'}")
 
         if raw_response.prompt_feedback and raw_response.prompt_feedback.block_reason:
             reason = raw_response.prompt_feedback.block_reason_message or raw_response.prompt_feedback.block_reason
-            error_msg = f"Gemini API: Your prompt was blocked. Reason: {reason}. Try rephrasing."
+            error_msg = f"Gemini API: Prompt blocked. Reason: {reason}."
             print(f"WARNING: llm_clients.py - {error_msg}")
             return LLMResponse(error=error_msg, success=False, raw_response=raw_response, model_id_used=model_id)
-
         if not raw_response.candidates:
-            error_msg = "Gemini API: No candidates returned in response. This often indicates the prompt was blocked or an internal error occurred before generation."
-            if raw_response.prompt_feedback: error_msg += f" Prompt Feedback: {raw_response.prompt_feedback}"
+            error_msg = "Gemini API: No candidates in response (often due to blocking)."
+            if raw_response.prompt_feedback: error_msg += f" Feedback: {raw_response.prompt_feedback}"
             print(f"WARNING: llm_clients.py - {error_msg}")
             return LLMResponse(error=error_msg, success=False, raw_response=raw_response, model_id_used=model_id)
 
         candidate = raw_response.candidates[0]
         if not candidate.content or not candidate.content.parts:
             finish_reason = str(candidate.finish_reason if candidate.finish_reason else "UNKNOWN").upper()
-            error_msg = f"Gemini API: Response generation stopped or yielded no content parts. Finish Reason: {finish_reason}."
-            if finish_reason == "SAFETY": error_msg += " Likely due to safety filters."
-            elif finish_reason == "RECITATION": error_msg += " Likely due to recitation policy."
-            elif finish_reason == "MAX_TOKENS": error_msg += " Consider increasing max_new_tokens if content seems truncated."
-            
+            error_msg = f"Gemini API: No content parts. Finish Reason: {finish_reason}."
+            if finish_reason == "SAFETY": error_msg += " Likely safety filters."
             print(f"WARNING: llm_clients.py - {error_msg}")
-            # Attempt to extract partial text if MAX_TOKENS or other non-error finish reasons
-            partial_text = ""
-            if candidate.content and candidate.content.parts and hasattr(candidate.content.parts[0], 'text'):
-                partial_text = candidate.content.parts[0].text
-            
-            if partial_text and finish_reason != "SAFETY" and finish_reason != "RECITATION" and finish_reason != "OTHER": # Only return partial if not a hard block
-                 return LLMResponse(text=partial_text + f"\n[Note: Generation ended due to {finish_reason}]", raw_response=raw_response, model_id_used=model_id)
-            else: # If safety/recitation or truly no text, return as error
-                 return LLMResponse(error=error_msg, success=False, raw_response=raw_response, model_id_used=model_id)
+            partial_text = candidate.content.parts[0].text if candidate.content and candidate.content.parts and hasattr(candidate.content.parts[0], 'text') else ""
+            return LLMResponse(text=partial_text + f"\n[Note: Generation ended: {finish_reason}]" if partial_text else None, error=error_msg if not partial_text else None, success=bool(partial_text), raw_response=raw_response, model_id_used=model_id)
         
         response_text = candidate.content.parts[0].text
-        print(f"  Gemini API Call - Success for model: {model_id}. Response text (first 100 chars): {response_text[:100]}...")
+        print(f"  Gemini API Call - Success for {model_id}. Response (first 100): {response_text[:100]}...")
         return LLMResponse(text=response_text, raw_response=raw_response, model_id_used=model_id)
-
     except Exception as e:
-        error_msg = f"Gemini API Call Exception ({model_id}): {type(e).__name__} - {str(e)}"
-        # Specific error parsing from previous version is good, let's keep it.
-        if "API key not valid" in str(e) or "PERMISSION_DENIED" in str(e):
-            error_msg = f"Gemini API Error ({model_id}): API key invalid or permission denied. Check GOOGLE_API_KEY and ensure Gemini API is enabled in Google Cloud. Original: {str(e)}"
-        elif "Could not find model" in str(e) or "ील नहीं मिला" in str(e):
-            error_msg = f"Gemini API Error ({model_id}): Model ID '{model_id}' not found or inaccessible with your key. Original: {str(e)}"
-        elif "User location is not supported" in str(e):
-            error_msg = f"Gemini API Error ({model_id}): User location not supported for this model/API. Original: {str(e)}"
-        elif "Quota exceeded" in str(e): # Check for "Quota" in the error message from Google
-            error_msg = f"Gemini API Error ({model_id}): API quota exceeded. Please check your Google Cloud quotas. Original: {str(e)}"
-        
+        error_msg = f"Gemini API Exception ({model_id}): {type(e).__name__} - {str(e)}"
+        # ... (specific error parsing as before) ...
+        if "API key not valid" in str(e) or "PERMISSION_DENIED" in str(e): error_msg = f"Gemini API Error ({model_id}): API key invalid/permission denied. Check GOOGLE_API_KEY & Google Cloud. Original: {str(e)}"
+        elif "Could not find model" in str(e) : error_msg = f"Gemini API Error ({model_id}): Model ID '{model_id}' not found/inaccessible. Original: {str(e)}"
+        elif "Quota exceeded" in str(e): error_msg = f"Gemini API Error ({model_id}): API quota exceeded. Original: {str(e)}"
         print(f"ERROR: llm_clients.py - {error_msg}")
         return LLMResponse(error=error_msg, success=False, raw_response=e, model_id_used=model_id)