""" API Handler - Obsługa komunikacji z OpenAI API """ from openai import OpenAI import time class APIHandler: """Obsługuje komunikację z OpenAI API""" def __init__(self, api_key): self.api_key = api_key self.client = None if api_key: self.client = OpenAI(api_key=api_key) def set_api_key(self, api_key): """Ustawia nowy klucz API""" self.api_key = api_key self.client = OpenAI(api_key=api_key) def validate_api_key(self): """ Waliduje klucz API poprzez próbę pobrania listy modeli Returns: tuple: (success: bool, message: str) """ if not self.api_key: return False, "Klucz API jest pusty" try: self.client = OpenAI(api_key=self.api_key) # Próba pobrania modeli jako test połączenia models = self.client.models.list() return True, "Klucz API jest poprawny" except Exception as e: error_msg = str(e) if "401" in error_msg or "Incorrect API key" in error_msg: return False, "Nieprawidłowy klucz API" elif "429" in error_msg: return False, "Przekroczono limit zapytań" else: return False, f"Błąd połączenia: {error_msg[:100]}" def get_available_models(self): """ Pobiera listę dostępnych modeli (GPT, o1, fine-tuned) Returns: list: Lista nazw modeli """ if not self.client: return ["gpt-4o", "gpt-4", "gpt-3.5-turbo"] # Domyślne modele try: models = self.client.models.list() # Filtruj modele OpenAI (GPT, o1, fine-tuned) # Wykluczamy tylko modele innych firm (np. whisper, dall-e, tts, embeddings) excluded_prefixes = ('whisper-', 'dall-e', 'tts-', 'text-embedding', 'babbage', 'davinci') openai_models = [ model.id for model in models.data if not model.id.startswith(excluded_prefixes) ] # Sortuj z priorytetem dla popularnych modeli priority_models = ["gpt-4o", "gpt-4-turbo", "gpt-4", "gpt-3.5-turbo", "o1", "o1-mini", "o1-preview"] sorted_models = [] # Dodaj modele priorytetowe, jeśli istnieją for pm in priority_models: if pm in openai_models: sorted_models.append(pm) openai_models.remove(pm) # Posortuj pozostałe modele # Fine-tuned modele (ft:...) na końcu, alfabetycznie standard_models = [m for m in openai_models if not m.startswith('ft:')] finetuned_models = [m for m in openai_models if m.startswith('ft:')] sorted_models.extend(sorted(standard_models)) sorted_models.extend(sorted(finetuned_models)) return sorted_models if sorted_models else ["gpt-4o", "gpt-4", "gpt-3.5-turbo"] except Exception as e: print(f"Błąd pobierania modeli: {e}") return ["gpt-4o", "gpt-4", "gpt-3.5-turbo"] def generate_response(self, prompt, model="gpt-4o", temperature=0.1, max_tokens=2000, top_p=1.0): """ Generuje odpowiedź z OpenAI API z automatycznym fallback dla nowych modeli Args: prompt: Tekst promptu systemowego model: Model OpenAI temperature: Temperatura (0.0-2.0) max_tokens: Maksymalna długość odpowiedzi top_p: Nucleus sampling parameter Returns: str: Wygenerowana odpowiedź lub komunikat błędu """ if not self.client: return "ERROR: Brak połączenia z API (nieprawidłowy klucz)" # Przygotuj parametry API - najpierw spróbuj ze starym API (max_tokens) api_params = { "model": model, "messages": [ {"role": "system", "content": prompt}, {"role": "user", "content": "Please provide your response based on the system prompt."} ], "temperature": temperature, "max_tokens": max_tokens, # Starsze modele (gpt-4, gpt-3.5, fine-tuned) "top_p": top_p } try: # Pierwsza próba: użyj max_tokens (kompatybilność ze starszymi modelami) response = self.client.chat.completions.create(**api_params) return response.choices[0].message.content except Exception as e: error_msg = str(e) # Automatyczny fallback: jeśli błąd dotyczy max_tokens, przełącz na max_completion_tokens if "max_tokens" in error_msg and "max_completion_tokens" in error_msg: try: # Usuń stary parametr i dodaj nowy api_params.pop("max_tokens") api_params["max_completion_tokens"] = max_tokens # Usuń również temperature i top_p (nowe modele ich nie akceptują) api_params.pop("temperature", None) api_params.pop("top_p", None) # Ponów zapytanie z nowymi parametrami response = self.client.chat.completions.create(**api_params) return response.choices[0].message.content except Exception as retry_error: return f"ERROR: Retry failed: {str(retry_error)[:200]}" # Standardowa obsługa błędów if "429" in error_msg: return f"ERROR: Rate limit exceeded - poczekaj chwilę" elif "insufficient_quota" in error_msg: return f"ERROR: Brak środków na koncie OpenAI" elif "invalid_request_error" in error_msg: return f"ERROR: Nieprawidłowe parametry: {error_msg[:200]}" else: return f"ERROR: {error_msg[:200]}" def estimate_cost(self, model, num_responses, avg_prompt_tokens=500, avg_completion_tokens=1000): """ Szacuje koszt testu Args: model: Nazwa modelu num_responses: Liczba odpowiedzi (dla obu promptów łącznie) avg_prompt_tokens: Średnia liczba tokenów w prompcie avg_completion_tokens: Średnia liczba tokenów w odpowiedzi Returns: float: Szacunkowy koszt w USD """ # Ceny za 1M tokenów (stan na grudzień 2024) pricing = { "gpt-4o": {"input": 2.50, "output": 10.00}, "gpt-4-turbo": {"input": 10.00, "output": 30.00}, "gpt-4": {"input": 30.00, "output": 60.00}, "gpt-3.5-turbo": {"input": 0.50, "output": 1.50}, "o1-preview": {"input": 15.00, "output": 60.00}, "o1-mini": {"input": 3.00, "output": 12.00}, "o1": {"input": 15.00, "output": 60.00} } # Znajdź odpowiedni cennik model_pricing = None for key in pricing: if key in model: model_pricing = pricing[key] break if not model_pricing: model_pricing = pricing["gpt-4o"] # Domyślnie gpt-4o # Oblicz koszt input_cost = (avg_prompt_tokens * num_responses * model_pricing["input"]) / 1_000_000 output_cost = (avg_completion_tokens * num_responses * model_pricing["output"]) / 1_000_000 total_cost = input_cost + output_cost return round(total_cost, 4)