""" Gemini API client for embeddings and text generation. """ import os import time import logging from typing import List, Dict, Optional import numpy as np import requests import json logger = logging.getLogger(__name__) class GeminiClient: """Client for interacting with Google Gemini API.""" def __init__(self): self.embedding_dim = 768 # Default dimension for Gemini embeddings self.api_key = os.getenv('GEMINI_API_KEY') if not self.api_key: raise ValueError("GEMINI_API_KEY environment variable is required") self.project_id = os.getenv('GEMINI_PROJECT', '') self.base_url = "https://generativelanguage.googleapis.com/v1beta" # Rate limiting self.last_request_time = 0 self.min_request_interval = 1.0 # seconds logger.info("Gemini client initialized") def _wait_for_rate_limit(self): """Simple rate limiting to avoid hitting API limits.""" current_time = time.time() time_since_last = current_time - self.last_request_time if time_since_last < self.min_request_interval: sleep_time = self.min_request_interval - time_since_last time.sleep(sleep_time) self.last_request_time = time.time() def _make_request(self, url: str, payload: Dict, retries: int = 3) -> Dict: """Make HTTP request to Gemini API with retry logic.""" for attempt in range(retries): try: self._wait_for_rate_limit() headers = { 'Content-Type': 'application/json' } response = requests.post( f"{url}?key={self.api_key}", headers=headers, json=payload, timeout=30 ) if response.status_code == 200: return response.json() elif response.status_code == 429: # Rate limit wait_time = (2 ** attempt) * 2 # Exponential backoff logger.warning(f"Rate limited, waiting {wait_time}s before retry {attempt + 1}") time.sleep(wait_time) continue else: logger.error(f"API request failed: {response.status_code} - {response.text}") response.raise_for_status() except requests.exceptions.RequestException as e: logger.error(f"Request attempt {attempt + 1} failed: {e}") if attempt == retries - 1: raise time.sleep(2 ** attempt) raise Exception("All retry attempts failed") def embed_texts(self, texts: List[str]) -> List[np.ndarray]: """Generate embeddings for a list of texts using Gemini.""" if not texts: return [] try: # Gemini embedding endpoint url = f"{self.base_url}/models/text-embedding-004:embedContent" embeddings = [] # Process texts in batches to avoid hitting limits batch_size = 10 for i in range(0, len(texts), batch_size): batch_texts = texts[i:i + batch_size] for text in batch_texts: payload = { "model": "models/text-embedding-004", "content": { "parts": [{ "text": text }] } } response_data = self._make_request(url, payload) if 'embedding' in response_data and 'values' in response_data['embedding']: embedding = np.array(response_data['embedding']['values'], dtype=np.float32) embeddings.append(embedding) else: logger.error(f"Unexpected embedding response: {response_data}") # Fallback to random embedding embeddings.append(np.random.rand(self.embedding_dim).astype(np.float32)) logger.info(f"Generated {len(embeddings)} embeddings") return embeddings except Exception as e: logger.error(f"Error generating embeddings: {e}") # Fallback to random embeddings for development logger.warning("Using random embeddings as fallback") return [np.random.rand(self.embedding_dim).astype(np.float32) for _ in texts] def generate_with_context(self, system_prompt: str, user_message: str, contexts: List[str], conversation_history: List[Dict] = None) -> Dict: """Generate response using Gemini with provided context and conversation history.""" try: # Build the complete prompt with conversation history context_section = "" if contexts: context_section = "\n\nالسياق المتاح:\n" + "\n---\n".join(contexts) # Add conversation history if available history_section = "" if conversation_history: history_section = "\n\nالمحادثة السابقة:\n" for i, entry in enumerate(conversation_history[-3:]): # Last 3 exchanges history_section += f"المستخدم: {entry.get('user_message', '')}\n" history_section += f"المساعد: {entry.get('assistant_response', '')}\n---\n" full_prompt = f"""{system_prompt} {history_section} {context_section} سؤال المستخدم الحالي: {user_message} يرجى الإجابة باللغة العربية مع مراعاة سياق المحادثة السابقة. اجعل إجابتك مفيدة ومختصرة (2-4 جمل) واقترح 2-4 أسئلة متابعة مفيدة.""" # Gemini generation endpoint url = f"{self.base_url}/models/gemini-2.0-flash:generateContent" payload = { "contents": [{ "parts": [{ "text": full_prompt }] }], "generationConfig": { "temperature": 0.7, "topK": 40, "topP": 0.95, "maxOutputTokens": 512, "stopSequences": [] } } response_data = self._make_request(url, payload) # Extract generated text if ('candidates' in response_data and len(response_data['candidates']) > 0 and 'content' in response_data['candidates'][0] and 'parts' in response_data['candidates'][0]['content']): generated_text = response_data['candidates'][0]['content']['parts'][0]['text'] # Try to extract suggested questions from the response suggested_questions = self._extract_suggested_questions(generated_text) return { 'text': generated_text, 'suggested_questions': suggested_questions, 'usage': response_data.get('usageMetadata', {}) } else: logger.error(f"Unexpected generation response: {response_data}") return { 'text': 'عذراً، حدث خطأ في توليد الإجابة.', 'suggested_questions': ["ما هي عروض السفر؟", "عنّا", "التاشيرات"] } except Exception as e: logger.error(f"Error generating response: {e}") return { 'text': 'عذراً، حدث خطأ مؤقت. يرجى المحاولة مرة أخرى.', 'suggested_questions': ["ما هي عروض السفر؟", "عنّا", "التاشيرات"] } def _extract_suggested_questions(self, text: str) -> List[str]: """Extract suggested questions from generated text.""" # Default suggestions default_suggestions = [ "ما هي عروض السفر؟", "عنّا", "التاشيرات", "احجز رحلة" ] # Simple heuristic to find questions in the response lines = text.split('\n') questions = [] for line in lines: line = line.strip() if line.endswith('؟') and len(line) < 100: # Arabic question mark questions.append(line) # Return found questions or defaults if questions and len(questions) <= 6: return questions[:4] # Max 4 suggestions else: return default_suggestions def test_connection(self) -> bool: """Test if the Gemini API connection is working.""" try: test_response = self.embed_texts(["تجربة الاتصال"]) return len(test_response) > 0 except Exception as e: logger.error(f"Connection test failed: {e}") return False def get_available_models(self) -> List[str]: """Get list of available Gemini models.""" try: url = f"{self.base_url}/models" headers = {'Content-Type': 'application/json'} response = requests.get(f"{url}?key={self.api_key}", headers=headers, timeout=10) if response.status_code == 200: data = response.json() models = [model['name'] for model in data.get('models', [])] return models else: logger.error(f"Failed to get models: {response.status_code}") return [] except Exception as e: logger.error(f"Error getting models: {e}") return []