# llm_providers/gemini_provider.py from google import genai from google.genai import types import logging import os import mimetypes from urllib.parse import urlparse import requests import base64 logger = logging.getLogger(__name__) def convert_to_gemini_parts(content): """Converts message content to a format compatible with Gemini.""" def _image_part_from_data_url(data_url: str): try: if isinstance(data_url, str) and data_url.startswith('data:') and ';base64,' in data_url: header, b64 = data_url.split(';base64,', 1) mime = 'image/jpeg' if header.startswith('data:') and '/' in header: mime = header[len('data:'):] return {"inline_data": {"mime_type": mime, "data": b64}} except Exception as e: logger.error(f"Error processing data URL: {e}", exc_info=True) pass return None def _image_part_from_url(image_url: str): """ Converts an image URL to Gemini's inline_data format. First, it attempts to read the image from the local filesystem if the URL points to a local resource. If the local file is not found, it falls back to fetching the image via HTTP. """ try: if not isinstance(image_url, str) or not image_url.startswith('http'): return None parsed = urlparse(image_url) # Attempt to read from local filesystem first for user images. if parsed.path.startswith('/user_images/'): fs_path = '' try: # Construct the local filesystem path from the web path. base_dir = os.path.join(os.path.dirname(__file__), '..', 'user_images') relative_path = parsed.path[len('/user_images/'):].lstrip('/') fs_path = os.path.join(base_dir, *relative_path.split('/')) if os.path.exists(fs_path): logger.info(f"Found image in local filesystem: {fs_path}") with open(fs_path, 'rb') as f: data_bytes = f.read() mime = mimetypes.guess_type(fs_path)[0] or 'image/jpeg' b64 = base64.b64encode(data_bytes).decode('utf-8') return {"inline_data": {"mime_type": mime, "data": b64}} except Exception as e: logger.warning(f"Failed to read local image {fs_path}: {e}. Falling back to HTTP.", exc_info=True) # Fallback to HTTP fetch if local read is not applicable or fails. logger.info(f"Local file not found or not applicable. Fetching from URL: {image_url}") try: resp = requests.get(image_url, timeout=15) resp.raise_for_status() mime = resp.headers.get('content-type', 'image/jpeg') if 'image' not in mime: guessed_mime = mimetypes.guess_type(parsed.path)[0] if guessed_mime and 'image' in guessed_mime: mime = guessed_mime b64 = base64.b64encode(resp.content).decode('utf-8') return {"inline_data": {"mime_type": mime, "data": b64}} except Exception as e: logger.error(f"Failed to fetch image from URL {image_url}: {e}", exc_info=True) return None except Exception as e: logger.error(f"Unexpected error in _image_part_from_url for {image_url}: {e}", exc_info=True) return None if isinstance(content, str): return [{"text": content}] if isinstance(content, list): parts = [] for item in content: if isinstance(item, dict): if item.get('type') == 'text' and 'text' in item: parts.append({"text": item['text']}) continue if item.get('type') == 'image_url': url = None if isinstance(item.get('image_url'), dict): url = item['image_url'].get('url') elif isinstance(item.get('image_url'), str): url = item.get('image_url') img_part = _image_part_from_data_url(url or '') or _image_part_from_url(url or '') parts.append(img_part if img_part else {"text": str(url or item)}) continue if item.get('type') == 'image': source = item.get('source') or {} if isinstance(source, dict) and source.get('type') == 'base64': mime = source.get('media_type', 'image/jpeg') b64 = source.get('data', '') if b64: parts.append({"inline_data": {"mime_type": mime, "data": b64}}) continue parts.append({"text": str(item)}) return parts if isinstance(content, dict): if content.get('type') == 'text' and 'text' in content: return [{"text": content['text']}] if content.get('type') == 'image_url': url = None if isinstance(content.get('image_url'), dict): url = content['image_url'].get('url') elif isinstance(content.get('image_url'), str): url = content.get('image_url') img_part = _image_part_from_data_url(url or '') or _image_part_from_url(url or '') return [img_part] if img_part else [{"text": str(url or content)}] return [{"text": str(content)}] def call_gemini_stream(chat_history, model_name, api_key, system_instruction, subject, reasoner_mode): """ Sends messages to Gemini API and returns the response as a stream. Handles message formatting and error handling. """ # Removing http_options for now as it causes 400 INVALID_ARGUMENT in some environments client = genai.Client(api_key=api_key) logger.info("Starting Gemini API processing...") try: instr_len = len(system_instruction) if system_instruction is not None else 0 except Exception: instr_len = 0 logger.info(f"System instruction length: {instr_len} chars") messages_for_gemini = chat_history gemini_history = [] for msg in messages_for_gemini: role = 'model' if msg['role'] == 'assistant' else 'user' parts = convert_to_gemini_parts(msg['content']) gemini_history.append({'role': role, 'parts': parts}) logger.info(f"Gemini history contains {len(gemini_history)} messages.") try: config_kwargs = {} if reasoner_mode: # If in reasoner mode, set temperature to 0 config_kwargs['temperature'] = 0 else: a = 1 # If not in reasoner mode (chat model), set thinking_budget to 0 # config_kwargs['thinking_config'] = types.ThinkingConfig(thinking_budget=0) # Temperature can be left at default or set explicitly if needed for chat models config = types.GenerateContentConfig( system_instruction=system_instruction if system_instruction else None, **config_kwargs ) logger.info("Sending message to Gemini via generate_content_stream...") response = client.models.generate_content_stream( model=model_name, contents=gemini_history, config=config ) for chunk in response: try: if not chunk.candidates: continue # Yield text part if no function call if hasattr(chunk, 'text') and chunk.text: yield chunk.text except Exception as e: logger.error(f"Error processing Gemini stream chunk: {e}", exc_info=True) pass except Exception as e: logger.error(f"Failed to stream from Gemini: {e}", exc_info=True) yield f"Error streaming from Gemini: {str(e)}"