Spaces:
Sleeping
Sleeping
| 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() | |
| 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() |