Image Features Pipeline

Overview

The Image Features Pipeline is a multi-modal document enrichment system that extracts visual intelligence from images and stores the results as searchable metadata and embeddings. When a document contains images, the pipeline downsamples large frames, converts them to data URIs, and then runs multiple analysis passes: OCR for text extraction, captioning for natural-language descriptions, and CLIP similarity scoring for semantic embedding. Each of these capabilities is gated by runtime feature flags, allowing operators to enable or disable expensive processing paths based on cost and latency budgets.

A key design constraint is that the vision API (the underlying multi-modal model service) has strict concurrency limits. The pipeline serializes all vision API calls through a semaphore, ensuring that concurrent document processing never overwhelms the API. This is critical in batch ingestion scenarios where thousands of documents with images arrive in bursts.

The extracted features—caption text, OCR text, CLIP embedding vectors, and image URIs—are attached to each document before it enters the vector index. This enables rich multi-modal search: users can find documents by the content of their images, not just their text, without needing to reprocess images at query time.

Core Concepts

How It Works

The pipeline processes documents in three stages: image preprocessing, feature extraction (gated by flags), and document enrichment. All vision API calls are funneled through a shared semaphore regardless of which feature flag triggered them.

Image Preprocessing

import base64
from io import BytesIO
from PIL import Image
from typing import Optional, Tuple

MAX_DIMENSION = 1536

class ImageProcessor:
    """Downsamples and encodes images for the vision API."""

    @staticmethod
    def process(image_bytes: bytes, max_dim: int = MAX_DIMENSION
                ) -> Tuple[str, Tuple[int, int]]:
        """
        Downsample if needed and return a data URI plus final dimensions.

        Args:
            image_bytes: Raw image bytes (JPEG, PNG, etc.)
            max_dim: Maximum allowed dimension on either axis.

        Returns:
            (data_uri, (width, height)) after downsampling.
        """
        img = Image.open(BytesIO(image_bytes))
        original_size = img.size  # (width, height)

        # Downsample if either dimension exceeds max_dim
        if max(original_size) > max_dim:
            ratio = max_dim / max(original_size)
            new_size = (
                int(original_size[0] * ratio),
                int(original_size[1] * ratio),
            )
            img = img.resize(new_size, Image.LANCZOS)

        # Encode to JPEG data URI
        buffer = BytesIO()
        img.save(buffer, format="JPEG", quality=85)
        b64 = base64.b64encode(buffer.getvalue()).decode("ascii")
        data_uri = f"data:image/jpeg;base64,{b64}"

        return data_uri, img.size

Semaphore-Gated Feature Extraction

import asyncio
from dataclasses import dataclass, field
from typing import List, Optional

@dataclass
class ImageFeatures:
    caption: Optional[str] = None
    ocr_text: Optional[str] = None
    clip_embedding: Optional[List[float]] = None
    image_uri: Optional[str] = None

@dataclass
class FeatureFlags:
    enable_ocr: bool = True
    enable_captioning: bool = True
    enable_clip: bool = True

class VisionClient:
    """Wraps the vision API with semaphore-based concurrency control."""

    def __init__(self, api_key: str, max_concurrent: int = 3):
        self.api_key = api_key
        self._semaphore = asyncio.Semaphore(max_concurrent)

    async def caption(self, data_uri: str) -> str:
        async with self._semaphore:
            # POST to the vision API /v1/caption with the data URI
            # Returns a natural-language description of the image
            return await self._call_api("caption", data_uri)

    async def ocr(self, data_uri: str) -> str:
        async with self._semaphore:
            # POST to the vision API /v1/ocr with the data URI
            # Returns extracted text from the image
            return await self._call_api("ocr", data_uri)

    async def clip_embed(self, data_uri: str) -> List[float]:
        async with self._semaphore:
            # POST to the vision API /v1/clip/embed with the data URI
            # Returns CLIP embedding vector
            return await self._call_api("clip_embed", data_uri)

    async def _call_api(self, endpoint: str, data_uri: str):
        """Make the actual HTTP call to the vision API. Implementation omitted."""
        ...


async def extract_image_features(
    client: VisionClient,
    image_bytes: bytes,
    flags: FeatureFlags,
) -> ImageFeatures:
    """
    Extract features from an image according to the enabled flags.
    All API calls go through the client's semaphore.
    """
    processor = ImageProcessor()
    data_uri, dimensions = processor.process(image_bytes)

    features = ImageFeatures(image_uri=data_uri)

    # Run enabled extractions concurrently (semaphore handles serialization)
    tasks = []

    if flags.enable_captioning:
        tasks.append(("caption", client.caption(data_uri)))
    if flags.enable_ocr:
        tasks.append(("ocr", client.ocr(data_uri)))
    if flags.enable_clip:
        tasks.append(("clip", client.clip_embed(data_uri)))

    # Gather results
    if tasks:
        results = await asyncio.gather(
            *[t[1] for t in tasks], return_exceptions=True
        )
        for (name, _), result in zip(tasks, results):
            if isinstance(result, Exception):
                print(f"  Warning: {name} extraction failed: {result}")
                continue
            if name == "caption":
                features.caption = result
            elif name == "ocr":
                features.ocr_text = result
            elif name == "clip":
                features.clip_embedding = result

    return features

Document Enrichment Pipeline

from typing import Dict, Any

async def enrich_document(
    doc: Dict[str, Any],
    client: VisionClient,
    flags: FeatureFlags,
) -> Dict[str, Any]:
    """
    Enrich a document with image features if it contains images.

    The document's 'images' field should be a list of raw image bytes.
    Features are merged into the document's metadata fields.
    """
    images = doc.get("images", [])
    if not images:
        return doc

    # Process the primary image (first in the list)
    primary_image = images[0]
    features = await extract_image_features(client, primary_image, flags)

    # Attach features to document
    if features.caption:
        doc["image_caption"] = features.caption
    if features.ocr_text:
        doc["image_ocr_text"] = features.ocr_text
    if features.clip_embedding:
        doc["clip_embedding"] = features.clip_embedding
    if features.image_uri:
        doc["image_uri"] = features.image_uri

    # If CLIP embedding is available, use it as the document's _embedding
    # for multi-modal vector search
    if features.clip_embedding and "_embedding" not in doc:
        doc["_embedding"] = features.clip_embedding

    return doc


async def enrich_batch(
    documents: List[Dict[str, Any]],
    client: VisionClient,
    flags: FeatureFlags,
) -> List[Dict[str, Any]]:
    """Enrich a batch of documents concurrently."""
    tasks = [
        enrich_document(doc, client, flags)
        for doc in documents
    ]
    return await asyncio.gather(*tasks)

Practical Applications

The search platform uses this pipeline to make image-heavy documents searchable. When users upload PDFs with screenshots, slide decks with diagrams, or chat logs with inline images, the pipeline extracts visual features before indexing. A search for “architecture diagram showing microservices” can match a document whose image caption mentions microservices, even if the document’s text never uses that term.

OCR extraction is particularly valuable for scanned documents and screenshots of terminal output or code. Rather than storing these as opaque binary blobs, the pipeline extracts their text content and indexes it alongside the document. Users searching for a specific error message will find the screenshot that contains it.

The CLIP embedding path enables visual similarity search. Given a query image, the system can find documents containing visually similar images by comparing CLIP vectors. This is used in the search platform UI for “find similar” functionality, where clicking on an image in search results surfaces other documents with related visual content.

Common Pitfalls

• Semaphore starvation under load: If the semaphore count is set too low relative to the batch size, most coroutines spend their time waiting for the semaphore. This creates a bottleneck where the pipeline’s throughput is limited by the API concurrency, not by processing speed. Monitor semaphore wait times and increase the limit if the API can handle more concurrent requests.
• Data URI size explosion: A 1536×1536 JPEG at quality 85 produces a data URI of roughly 200–400KB. For documents with many images, the enriched document can exceed storage limits. Consider storing only the primary image’s data URI, or switching to object storage references for images above a size threshold.
• OCR on non-text images: Running OCR on photographs, illustrations, or abstract diagrams produces garbage text that pollutes the search index. Use the feature flag to disable OCR for document types unlikely to contain text-heavy images, or add a confidence threshold to discard low-quality OCR results.
• CLIP dimension mismatch: Different CLIP model versions produce embeddings of different dimensions (512 for ViT-B/32, 768 for ViT-L/14). If the vision API changes its default CLIP model, existing embeddings become incomparable with new ones. Pin the model version explicitly in API calls and reindex if the model changes.
• Ignoring extraction failures: The pipeline logs warnings for failed extractions but continues processing. If the vision API is down, all documents pass through without image features, and the index silently becomes text-only. Add alerting on extraction failure rates and consider blocking batch completion if the failure rate exceeds a threshold.
• Memory pressure from concurrent image processing: Each image is loaded into memory as a PIL Image, processed, and then base64-encoded. Processing many large images concurrently can spike memory usage. The semaphore limits API calls but not local processing. Add a second semaphore or worker pool to limit concurrent image loads.