""" 🤖 LLM Sender Unified Module ماژول یکپارچه برای ارسال به ChatGPT و Grok ✨ با پشتیبانی از GPT-5 models و رفع مشکل temperature """ import requests import os import logging from typing import Optional import time from abc import ABC, abstractmethod logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class LLMSender(ABC): """کلاس پایه برای ارسال به مدل‌های مختلف LLM""" def __init__(self, api_key: Optional[str] = None, model: str = None): self.api_key = api_key self.model = model self.base_url = "" @abstractmethod def get_default_model(self) -> str: """مدل پیش‌فرض""" pass @abstractmethod def get_base_url(self) -> str: """URL پایه API""" pass def set_api_key(self, api_key: str): """تنظیم کلید API""" self.api_key = api_key logger.info("✅ کلید API تنظیم شد") def set_model(self, model: str): """تغییر مدل""" self.model = model logger.info(f"✅ مدل تغییر یافت به: {model}") def _uses_max_completion_tokens(self) -> bool: """بررسی اینکه آیا مدل از max_completion_tokens استفاده می‌کند""" models_with_completion_tokens = [ 'gpt-5', # تمام مدل‌های GPT-5 'gpt-5.1' # GPT-5.1 ] return any(self.model.startswith(prefix) for prefix in models_with_completion_tokens) def _requires_default_temperature(self) -> bool: """بررسی اینکه آیا مدل فقط temperature=1 را قبول می‌کند""" models_requiring_default_temp = [ 'gpt-5', # تمام مدل‌های GPT-5 'o1' # تمام مدل‌های O1 ] return any(self.model.startswith(prefix) for prefix in models_requiring_default_temp) def send_simple(self, text: str, lang: str = 'fa') -> str: """ارسال ساده بدون system message سفارشی""" system_msg = ( "شما یک تحلیلگر متخصص هستید. متن حاوی کدهای ناشناس است. " "به درخواست‌ها با دقت و حرفه‌ای پاسخ دهید." if lang == 'fa' else "You are a professional analyst. The text contains anonymous codes. " "Answer requests accurately and professionally." ) return self.send(text, system_msg=system_msg, lang=lang) def send( self, text: str, system_msg: Optional[str] = None, max_tokens: int = 2000, temperature: float = 0.2, # ✅ کاهش از 0.7 به 0.2 برای دقت بیشتر timeout: int = 60, lang: str = 'fa', retry_count: int = 3 ) -> str: """ارسال متن به LLM با کنترل کامل""" try: # بررسی اولیه if not text or not text.strip(): error_msg = "متن خالی است!" if lang == 'fa' else "Text is empty!" logger.error(f"❌ {error_msg}") return f"❌ {error_msg}" if not self.api_key: error_msg = "کلید API تنظیم نشده است!" if lang == 'fa' else "API Key not configured!" logger.error(f"❌ {error_msg}") return f"❌ {error_msg}" # تنظیم system message پیش‌فرض if system_msg is None: system_msg = ( "شما یک تحلیلگر مالی حرفه‌ای هستید. متن حاوی کدهای ناشناس است. " "به سوالات با دقت پاسخ دهید." if lang == 'fa' else "You are a professional financial analyst. The text contains anonymous codes. " "Answer questions accurately." ) # تهیه headers headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } # ✨ تنظیم temperature مناسب if self._requires_default_temperature(): actual_temperature = 1.0 if temperature != 1.0: logger.info(f"⚠️ مدل {self.model} فقط temperature=1 را قبول می‌کند") else: actual_temperature = temperature # ساخت request body data = { "model": self.model, "messages": [ {"role": "system", "content": system_msg}, {"role": "user", "content": text} ], "temperature": actual_temperature } # ✨ انتخاب پارامتر مناسب برای max tokens if self._uses_max_completion_tokens(): data["max_completion_tokens"] = max_tokens else: data["max_tokens"] = max_tokens # ارسال با retry mechanism for attempt in range(retry_count): try: logger.info(f"📤 ارسال درخواست به {self.__class__.__name__} (تلاش {attempt + 1}/{retry_count})...") response = requests.post( self.base_url, headers=headers, json=data, timeout=timeout ) # پردازش پاسخ موفق if response.status_code == 200: result = response.json() llm_response = result['choices'][0]['message']['content'] logger.info("✅ پاسخ دریافت شد") return llm_response # پردازش خطاهای مختلف elif response.status_code == 429: # Rate limiting wait_time = 5 * (attempt + 1) logger.warning(f"⚠️ Rate limit | صبر: {wait_time} ثانیه") if attempt < retry_count - 1: time.sleep(wait_time) continue else: return ( "❌ سهمیه API تمام شده است. لطفاً بعداً تلاش کنید." if lang == 'fa' else "❌ API quota exceeded. Please try later." ) elif response.status_code == 401: return ( "❌ کلید API نامعتبر است!" if lang == 'fa' else "❌ Invalid API key!" ) elif response.status_code in [502, 503, 504]: # Server errors wait_time = 2 * (attempt + 1) logger.warning(f"⚠️ Server error {response.status_code} | صبر: {wait_time} ثانیه") if attempt < retry_count - 1: time.sleep(wait_time) continue else: return ( f"❌ خطای سرور: {response.status_code}" if lang == 'fa' else f"❌ Server error: {response.status_code}" ) else: # خطای دیگر try: error_data = response.json() if response.content else {} if isinstance(error_data, dict): error_msg = error_data.get('error', {}).get('message', response.text) else: error_msg = str(error_data) except: error_msg = response.text[:200] logger.error(f"❌ API Error: {error_msg}") return f"❌ API Error: {error_msg}" except requests.exceptions.Timeout: logger.warning("⚠️ Timeout | صبر: 3 ثانیه و تلاش مجدد") if attempt < retry_count - 1: time.sleep(3) continue else: return ( "❌ خطای اتصال: timeout" if lang == 'fa' else "❌ Connection error: timeout" ) except requests.exceptions.ConnectionError as e: logger.warning("⚠️ Connection error | صبر: 2 ثانیه و تلاش مجدد") if attempt < retry_count - 1: time.sleep(2) continue else: return ( f"❌ خطای اتصال: {str(e)}" if lang == 'fa' else f"❌ Connection error: {str(e)}" ) except Exception as e: logger.error(f"❌ خطای غیرمنتظره: {str(e)}") return ( f"❌ خطا در ارتباط با LLM: {str(e)}" if lang == 'fa' else f"❌ Error connecting to LLM: {str(e)}" ) class ChatGPTSender(LLMSender): """کلاس برای ارسال به ChatGPT""" def __init__(self, api_key: Optional[str] = None, model: str = "gpt-4o-mini"): raw_key = api_key or os.getenv("OPENAI_API_KEY", "") cleaned_key = raw_key.strip() if raw_key else "" super().__init__(cleaned_key, model) self.base_url = self.get_base_url() if not self.api_key: logger.warning("⚠️ کلید OpenAI API تنظیم نشده است!") def get_default_model(self) -> str: return "gpt-4o-mini" def get_base_url(self) -> str: return "https://api.openai.com/v1/chat/completions" class GrokSender(LLMSender): """کلاس برای ارسال به Grok (xAI)""" def __init__(self, api_key: Optional[str] = None, model: str = "grok-beta"): raw_key = api_key or os.getenv("XAI_API_KEY", "") cleaned_key = raw_key.strip() if raw_key else "" super().__init__(cleaned_key, model) self.base_url = self.get_base_url() if not self.api_key: logger.warning("⚠️ کلید xAI API تنظیم نشده است!") def get_default_model(self) -> str: return "grok-beta" def get_base_url(self) -> str: return "https://api.x.ai/v1/chat/completions" def create_llm_sender( provider: str = "chatgpt", api_key: Optional[str] = None, model: Optional[str] = None ) -> LLMSender: """ایجاد LLM sender بر اساس provider""" provider = provider.lower() if provider == "chatgpt": if model is None: model = "gpt-4o-mini" return ChatGPTSender(api_key=api_key, model=model) elif provider == "grok": if model is None: model = "grok-beta" return GrokSender(api_key=api_key, model=model) else: raise ValueError(f"Provider نامعتبر: {provider}") # ✅ مدل‌های موجود (اصلاح شده) AVAILABLE_MODELS = { "chatgpt": [ "gpt-5.1", # ✅ بهترین GPT-5 "gpt-5", # ✅ GPT-5 پایه "gpt-4.1", "gpt-4o-mini", "gpt-4o", "gpt-4-turbo", "gpt-3.5-turbo" ], "grok": [ "grok-3-mini", "grok-3", "grok-2-1212" ] } if __name__ == "__main__": print("=" * 60) print("🤖 LLM Sender - نسخه اصلاح شده") print("✨ رفع مشکل temperature برای GPT-5") print("=" * 60) # تست print("\n🧪 تست مدل‌ها:") test_models = ['gpt-5', 'gpt-5.1', 'gpt-4o'] for model in test_models: sender = create_llm_sender("chatgpt", model=model) uses_completion = sender._uses_max_completion_tokens() requires_default_temp = sender._requires_default_temperature() print(f"\n مدل: {model}") print(f" • max_tokens: {'max_completion_tokens' if uses_completion else 'max_tokens'}") print(f" • temperature: {'1.0 (default only)' if requires_default_temp else '0.7 (custom)'}")