""" authors: Yuchen Xie (https://github.com/yuchenxie4645) title: External Vision Layer Features: 1. This filter implements an asynchronous image-to-text transcriber system using Google's Gemini API (v1beta). - Users are permitted to modify code to utilize different models. 2. Supports both single and batch image processing. - Meaning one or multiple images per query will be batched as one request 3. Includes a retry mechanism, per-image caching to avoid redundant processing. - Cached images are entirely skipped from further analysis to Gemini. 4. Images are fetched via `aiohttp`, encoded in base64, and submitted to Gemini’s `generate_content` endpoint using `inline_data`. 5. Generated content from VLM (in this case Gemini) will replace the image URL as context for non-vlm base model. - VLM base model also works because the base model will not even see the images, completely stripped from chat. - API's such as OpenRouter, Groq, and Sambanova API models are tested to function. 6. The base model knows the order the images were sent, and will receive the images in this format: <image 1>[detailed transcription of first image]</image> <image 2>[detailed transcription of second image]</image> <image 3>[detailed transcription of third image]</image> 7. Currently hardcoded to limit max 3 images sent per query. Increase as you see fit. Demo: --- website tested: https://arlowgpt.com version: 1.0 license: MIT License """ import aiohttp import base64 import google.generativeai as genai import google.generativeai.types as types from google.generativeai.types import GenerateContentResponse from google.api_core.exceptions import TooManyRequests from typing import Callable, Awaitable, Any, Optional, Dict from pydantic import BaseModel, Field class Filter: class Valves(BaseModel): priority: int = Field(default=0) # REQUIRED: Set your Google API key here for the vision functionality to work API_Key: str = Field( default="", description="API Key for Gemini API.", ) retries: int = Field(default=3) # use a supported model for v1beta generateContent model: str = Field( default="gemini-2.5-flash", description="Vision-capable model.", ) prompt: str = Field( default=( """You act solely as a vision-to-text assistant transcribing every word you see, every small detail needs to be covered. You respond only with detail—no affirmations, no queues. If the picture only depicts words, repeat them word for word. If you see words + pictures, describe both. If something is blurry, unclear, or ambiguous, say so. If there is a specific character/item that has a name, name it, don't just describe it. You are allowed to use search. You are to use strictly in this format: Image 1: <your description for image number 1> Image 2: <your description for image number 2 if available> Image x: <your description for image number x if available>""" ), description="System prompt for image processing.", ) def __init__(self): # load settings self.valves = self.Valves() # deterministic + cap output length self._gen_config = types.GenerationConfig( temperature=0.0, top_p=1.0, max_output_tokens=2048, ) # cache to avoid reprocessing the same image multiple times in a single session # maps image_url -> transcription text self._ocr_cache: Dict[str, str] = {} def _ensure_api_configured(self): """Ensure the Gemini API is properly configured with a valid key.""" if not self.valves.API_Key or self.valves.API_Key.strip() == "": raise RuntimeError( "Google API key is required. Please set the API_Key in the valves configuration." ) # Configure the API with the current key genai.configure(api_key=self.valves.API_Key) # Create/recreate the model with current settings return genai.GenerativeModel( model_name=self.valves.model, generation_config=self._gen_config, ) async def _perform_ocr( self, image_url: str, __event_emitter__: Callable[[Any], Awaitable[None]], ) -> str: # short-circuit if we've already processed this image before if image_url in self._ocr_cache: return self._ocr_cache[image_url] # emit "Analyzing Image..." await __event_emitter__( { "type": "status", "data": {"description": "Analyzing image(s)...", "done": False}, } ) # prepare base64 data if image_url.startswith("data:"): _, b64data = image_url.split(",", 1) mime = image_url.split(";")[0].split(":")[1] b64_image = b64data else: async with aiohttp.ClientSession() as session: resp = await session.get(image_url) resp.raise_for_status() raw = await resp.read() mime = "image/jpeg" b64_image = base64.b64encode(raw).decode("utf-8") payload = [ {"inline_data": {"mime_type": mime, "data": b64_image}}, {"text": self.valves.prompt}, ] # retry loop for attempt in range(self.valves.retries): try: # Ensure API is configured with current key model = self._ensure_api_configured() resp: GenerateContentResponse = model.generate_content( payload, safety_settings={ # block anything flagged as sexually explicit types.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: types.HarmBlockThreshold.BLOCK_LOW_AND_ABOVE, }, ) text = resp.text.strip() # cache transcription for future reuse self._ocr_cache[image_url] = text # emit "Analyzation Complete" await __event_emitter__( { "type": "status", "data": {"description": "Analyzation Complete", "done": True}, } ) return text except TooManyRequests: # Rate-limit encountered – surface a user-friendly message and abort further processing raise RuntimeError( "Something went wrong, please try again in one minute." ) except Exception as e: if attempt == self.valves.retries - 1: raise RuntimeError( f"OCR failed after {self.valves.retries} tries: {e}" ) # otherwise, retry silently async def _perform_ocr_batch( self, image_urls: list[str], __event_emitter__: Callable[[Any], Awaitable[None]], ) -> None: """Batch OCR processing for multiple images in a single Gemini request.""" # remove URLs that have already been processed image_urls = [u for u in image_urls if u not in self._ocr_cache] if not image_urls: return # emit analyzing status await __event_emitter__( { "type": "status", "data": { "description": f"Analyzing {len(image_urls)} image(s)...", "done": False, }, } ) # build payload with numbered markers before each inline image payload_parts = [] async with aiohttp.ClientSession() as session: for idx, url in enumerate(image_urls): if url.startswith("data:"): _, b64data = url.split(",", 1) mime = url.split(";")[0].split(":")[1] b64_image = b64data else: resp = await session.get(url) resp.raise_for_status() raw = await resp.read() mime = "image/jpeg" b64_image = base64.b64encode(raw).decode("utf-8") # add marker so the model knows which image is which payload_parts.append({"text": f"Image {idx + 1}"}) payload_parts.append( {"inline_data": {"mime_type": mime, "data": b64_image}} ) # use the original OCR system prompt as-is (no additional formatting instructions) prompt_text = self.valves.prompt payload = payload_parts + [{"text": prompt_text}] for attempt in range(self.valves.retries): try: # Ensure API is configured with current key model = self._ensure_api_configured() resp: GenerateContentResponse = model.generate_content( payload, safety_settings={ types.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: types.HarmBlockThreshold.BLOCK_LOW_AND_ABOVE, }, ) raw_text = resp.text.strip() # Expecting numbered lines in the form "Image N: <description>" transcriptions = [] for line in raw_text.splitlines(): if line.lower().startswith("image"): parts = line.split(":", 1) if len(parts) == 2: transcriptions.append(parts[1].strip()) if len(transcriptions) != len(image_urls): raise RuntimeError( "OCR's response did not contain the expected number of image descriptions. Please try again." ) # cache results for idx, url in enumerate(image_urls): self._ocr_cache[url] = str(transcriptions[idx]).strip() # emit completion status await __event_emitter__( { "type": "status", "data": {"description": "Analyzation Complete", "done": True}, } ) return except TooManyRequests: raise RuntimeError( "Something went wrong, please try again in a second." ) except Exception as e: if attempt == self.valves.retries - 1: raise RuntimeError( f"OCR batch failed after {self.valves.retries} tries: {e}" ) # else retry silently async def inlet( self, body: dict, __event_emitter__: Callable[[Any], Awaitable[None]], __user__: Optional[dict] = None, __model__: Optional[dict] = None, ) -> dict: messages = body.get("messages", []) for idx, msg in enumerate(messages): if msg.get("role") == "user" and isinstance(msg.get("content"), list): # extract image URLs (preserve order, allow duplicates but they will be deduped for processing) image_urls = [ part["image_url"]["url"] for part in msg["content"] if part.get("type") == "image_url" ] # enforce attachment limit if len(image_urls) > 3: raise RuntimeError( "Too many image attachments. Please limit to 3 images." ) # perform batch OCR for any new images (cached images will be skipped internally) await self._perform_ocr_batch(image_urls, __event_emitter__) # rebuild content replacing image_url parts with cached transcriptions new_parts = [] image_counter = 1 for part in msg["content"]: if part.get("type") == "image_url": url = part["image_url"]["url"] transcription = self._ocr_cache[url] new_parts.append( { "type": "text", "text": f"<image {image_counter}>{transcription}</image>", } ) image_counter += 1 else: new_parts.append(part) messages[idx]["content"] = new_parts body["messages"] = messages return body async def outlet( self, body: dict, __event_emitter__: Callable[[Any], Awaitable[None]], __user__: Optional[dict] = None, __model__: Optional[dict] = None, ) -> dict: return body