File size: 2,979 Bytes
6b62834 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 | """Image modal processor — VLM captioning and visual analysis."""
import base64
from pathlib import Path
from typing import Callable, Optional
from agentic_rag.services.knowledge.content_list import ContentItem, ContentType
from agentic_rag.services.knowledge.processors.base import BaseModalProcessor
class ImageModalProcessor(BaseModalProcessor):
"""Generate captions and extract information from images using a Vision LLM.
Supports two calling conventions (inspired by RAG-Anything):
1. messages format — full VLM with image data + prompt
2. Callable function — any (image_data, prompt) -> str
"""
content_type = ContentType.IMAGE
description = "Generate captions for images using Vision LLM"
def __init__(
self,
vision_model_func: Optional[Callable] = None,
prompt: str = "",
):
"""
Args:
vision_model_func: Async callable (image_b64, prompt) -> str.
If None, uses the LLM provider's vision capability.
prompt: Custom prompt for image description.
"""
from agentic_rag.config.prompts import Prompts
self.vision_model_func = vision_model_func
self.prompt = prompt or Prompts.IMAGE_CAPTION
async def process(self, item: ContentItem) -> ContentItem:
"""Generate a caption for an image content item."""
if item.type != ContentType.IMAGE:
return item
image_path = item.img_path
if not image_path:
return item
# Load image as base64
try:
path = Path(image_path)
if not path.exists():
item.metadata["image_error"] = f"File not found: {image_path}"
return item
ext = path.suffix.lower()
mime_map = {".jpg": "image/jpeg", ".jpeg": "image/jpeg",
".png": "image/png", ".gif": "image/gif",
".webp": "image/webp", ".bmp": "image/bmp"}
mime = mime_map.get(ext, "image/png")
image_b64 = base64.b64encode(path.read_bytes()).decode()
data_uri = f"data:{mime};base64,{image_b64}"
except Exception as e:
item.metadata["image_error"] = str(e)
return item
# Generate caption (only if a vision function is configured)
if self.vision_model_func:
try:
caption = await self.vision_model_func(data_uri, self.prompt)
item.img_caption = caption
item.text = caption # Make searchable
except Exception as e:
item.metadata["image_error"] = f"Vision model error: {e}"
# When no vision function is available, we leave text/caption empty.
# The multimodal embedding model (e.g. jina-embeddings-v5-omni) will
# encode visual features directly from the image pixels — no text
# placeholder needed.
return item
|