import os import json import logging import threading import time from datetime import datetime from dotenv import load_dotenv from groq import Groq load_dotenv() logger = logging.getLogger(__name__) STATS_FILE = "usage_stats.json" class LLMSingleton: _instance = None _instance_lock = threading.Lock() @classmethod def get_instance(cls): if cls._instance is None: with cls._instance_lock: if cls._instance is None: cls._instance = cls() return cls._instance def __init__(self): if self._instance is not None: raise Exception("Singleton instance already exists!") self.api_key = os.getenv("GROQ_API_KEY") if not self.api_key: logger.warning("⚠️ GROQ_API_KEY not found in environment variables. AI features will fail.") self.client = None else: self.client = Groq(api_key=self.api_key) self.model_name = os.getenv("GROQ_MODEL", "llama-3.1-8b-instant") self._stats_lock = threading.Lock() self._rpm_lock = threading.Lock() self.stats = self._load_stats() self._check_daily_reset() self.rpm_limit = 30 self.minute_window_start = time.time() self.requests_this_minute = 0 def _load_stats(self): default_stats = { "total_requests": 0, "successful_requests": 0, "rate_limit_hits": 0, "input_tokens": 0, "output_tokens": 0, "errors": 0, "local_model_requests": 0, "date": datetime.now().strftime("%Y-%m-%d"), "daily_requests_count": 0, } if os.path.exists(STATS_FILE): try: with open(STATS_FILE, "r") as f: data = json.load(f) return {**default_stats, **data} except Exception as e: logger.error(f"Failed to load stats: {e}") return default_stats def _save_stats(self): try: with open(STATS_FILE, "w") as f: json.dump(self.stats, f, indent=2) except Exception as e: logger.error(f"Failed to save stats: {e}") def _check_daily_reset(self): with self._stats_lock: today = datetime.now().strftime("%Y-%m-%d") if self.stats.get("date") != today: logger.info("📅 New day detected. Resetting daily AI quotas.") self.stats["date"] = today self.stats["daily_requests_count"] = 0 self._save_stats() def _check_rpm_window(self): with self._rpm_lock: now = time.time() if now - self.minute_window_start >= 60: self.minute_window_start = now self.requests_this_minute = 0 def get_usage_stats(self): self._check_daily_reset() self._check_rpm_window() with self._stats_lock: stats = self.stats.copy() with self._rpm_lock: requests_this_minute = self.requests_this_minute daily_limit = 1000 stats["limits"] = { "requests_per_minute": self.rpm_limit, "requests_per_day": daily_limit, } stats["remaining_daily_requests"] = max(0, daily_limit - stats["daily_requests_count"]) stats["remaining_rpm"] = max(0, self.rpm_limit - requests_this_minute) return stats def track_local_usage(self, input_chars: int = 0): with self._stats_lock: self.stats["local_model_requests"] += 1 self.stats["input_tokens"] += input_chars // 4 self._save_stats() def _reserve_request_slot(self) -> bool: with self._stats_lock: if self.stats["daily_requests_count"] >= 1000: return False self.stats["total_requests"] += 1 self.stats["daily_requests_count"] += 1 self._save_stats() with self._rpm_lock: self.requests_this_minute += 1 return True def generate(self, prompt: str, max_tokens: int = 2048) -> str: import random self._check_daily_reset() self._check_rpm_window() if not self.api_key or not self.client: logger.error("Cannot generate: Missing GROQ_API_KEY") raise RuntimeError("MISSING_API_KEY") if not self._reserve_request_slot(): logger.error("❌ Daily Quota Exceeded. Request blocked.") raise RuntimeError("QUOTA_EXCEEDED") logger.info(f"🤖 Generating with Groq ({self.model_name}). Prompt start: {prompt[:50]}...") retries = 0 max_retries = 3 base_delay = 2 while retries <= max_retries: try: with self._stats_lock: self.stats["input_tokens"] += len(prompt) // 4 self._save_stats() chat_completion = self.client.chat.completions.create( model=self.model_name, messages=[ { "role": "system", "content": "You are a senior Android code reviewer. You MUST return a valid JSON object matching the requested schema strictly.", }, {"role": "user", "content": prompt}, ], response_format={"type": "json_object"}, max_tokens=max_tokens, temperature=0.2, ) response_text = chat_completion.choices[0].message.content or "" with self._stats_lock: self.stats["successful_requests"] += 1 if response_text: self.stats["output_tokens"] += len(response_text) // 4 self._save_stats() return response_text.strip() except Exception as e: error_str = str(e) if "429" in error_str or "rate_limit_exceeded" in error_str.lower(): with self._stats_lock: self.stats["rate_limit_hits"] += 1 self._save_stats() wait_time = (base_delay * (2**retries)) + random.uniform(0.5, 1.5) logger.warning( f"⚠️ Groq rate limit hit. Retrying in {wait_time:.2f}s... (Attempt {retries + 1}/{max_retries})" ) time.sleep(wait_time) retries += 1 else: with self._stats_lock: self.stats["errors"] += 1 self._save_stats() logger.error(f"Groq generation failed: {e}") raise RuntimeError(f"GENERATION_FAILED: {e}") with self._stats_lock: self.stats["errors"] += 1 self._save_stats() logger.error("❌ Max retries reached. Request failed.") raise RuntimeError("RATE_LIMIT_EXCEEDED") def generate_text(self, prompt: str) -> str: self._check_daily_reset() self._check_rpm_window() if not self.api_key or not self.client: return "Error: Missing API Key." if not self._reserve_request_slot(): return "Error: Daily Quota Exceeded." try: with self._stats_lock: self.stats["input_tokens"] += len(prompt) // 4 self._save_stats() chat_completion = self.client.chat.completions.create( model=self.model_name, messages=[ { "role": "system", "content": "You are GitGud AI, an expert software architect.", }, {"role": "user", "content": prompt}, ], max_tokens=2048, temperature=0.3, ) response_text = chat_completion.choices[0].message.content or "" with self._stats_lock: self.stats["successful_requests"] += 1 self.stats["output_tokens"] += len(response_text) // 4 self._save_stats() return response_text except Exception as e: logger.error(f"Groq text generation failed: {e}") return f"Error generating content: {str(e)}" llm_engine = LLMSingleton.get_instance()