backend/colpali.py

#!/usr/bin/env python3

import torch
from PIL import Image
import numpy as np
from typing import cast
from pathlib import Path
import base64
from io import BytesIO
from typing import Union, Tuple, List, Dict, Any
import matplotlib
import matplotlib.cm as cm
import re
import io

import json
import time

from colpali_engine.models import ColPali, ColPaliProcessor
from colpali_engine.utils.torch_utils import get_torch_device
from einops import rearrange
from vidore_benchmark.interpretability.torch_utils import (
    normalize_similarity_map_per_query_token,
)
from vidore_benchmark.interpretability.vit_configs import VIT_CONFIG
from vespa.application import Vespa
from vespa.io import VespaQueryResponse

matplotlib.use("Agg")

MAX_QUERY_TERMS = 64

COLPALI_GEMMA_MODEL_NAME = "vidore/colpaligemma-3b-pt-448-base"


def load_model() -> Tuple[ColPali, ColPaliProcessor]:
    model_name = "vidore/colpali-v1.2"

    device = get_torch_device("auto")
    print(f"Using device: {device}")

    # Load the model
    model = cast(
        ColPali,
        ColPali.from_pretrained(
            model_name,
            torch_dtype=torch.bfloat16 if torch.cuda.is_available() else torch.float32,
            device_map=device,
        ),
    ).eval()

    # Load the processor
    processor = cast(ColPaliProcessor, ColPaliProcessor.from_pretrained(model_name))
    return model, processor


def load_vit_config(model):
    # Load the ViT config
    print(f"VIT config: {VIT_CONFIG}")
    vit_config = VIT_CONFIG[COLPALI_GEMMA_MODEL_NAME]
    return vit_config


def save_figure(fig, filename: str = "similarity_map.png"):
    try:
        OUTPUT_DIR = Path(__file__).parent.parent / "output" / "sim_maps"
        OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
        fig.savefig(
            OUTPUT_DIR / filename,
            bbox_inches="tight",
            pad_inches=0,
        )
    except Exception as e:
        print(f"Failed to save figure: {e}")


def annotate_plot(ax, query, selected_token):
    # Add the query text as a title over the image with opacity
    ax.text(
        0.5,
        0.95,  # Adjust the position to be on the image (y=0.1 is 10% from the bottom)
        query,
        fontsize=18,
        color="white",
        ha="center",
        va="center",
        alpha=0.8,  # Set opacity (1 is fully opaque, 0 is fully transparent)
        bbox=dict(
            boxstyle="round,pad=0.5", fc="black", ec="none", lw=0, alpha=0.5
        ),  # Add a semi-transparent background
        transform=ax.transAxes,  # Ensure the coordinates are relative to the axes
    )

    # Add annotation with the selected token over the image with opacity
    ax.text(
        0.5,
        0.05,  # Position towards the top of the image
        f"Selected token: `{selected_token}`",
        fontsize=18,
        color="white",
        ha="center",
        va="center",
        alpha=0.8,  # Set opacity for the text
        bbox=dict(
            boxstyle="round,pad=0.3", fc="black", ec="none", lw=0, alpha=0.5
        ),  # Semi-transparent background
        transform=ax.transAxes,  # Keep the coordinates relative to the axes
    )
    return ax


def gen_similarity_maps(
    model: ColPali,
    processor: ColPaliProcessor,
    device,
    vit_config,
    query: str,
    query_embs: torch.Tensor,
    token_idx_map: dict,
    images: List[Union[Path, str]],
    vespa_sim_maps: List[str],
) -> List[Dict[str, str]]:
    """
    Generate similarity maps for the given images and query, and return base64-encoded blended images.

    Args:
        model (ColPali): The model used for generating embeddings.
        processor (ColPaliProcessor): Processor for images and text.
        device: Device to run the computations on.
        vit_config: Configuration for the Vision Transformer.
        query (str): The query string.
        query_embs (torch.Tensor): Query embeddings.
        token_idx_map (dict): Mapping from tokens to their indices.
        images (List[Union[Path, str]]): List of image paths or base64-encoded strings.
        vespa_sim_maps (List[str]): List of Vespa similarity maps.

    Returns:
        List[Dict[str, str]]: A list where each item is a dictionary mapping tokens to base64-encoded blended images.
    """

    start = time.perf_counter()

    # Prepare the colormap once to avoid recomputation
    colormap = cm.get_cmap("viridis")

    # Process images and store original images and sizes
    processed_images = []
    original_images = []
    original_sizes = []
    for img in images:
        if isinstance(img, Path):
            try:
                img_pil = Image.open(img).convert("RGB")
            except Exception as e:
                raise ValueError(f"Failed to open image from path: {e}")
        elif isinstance(img, str):
            try:
                img_pil = Image.open(BytesIO(base64.b64decode(img))).convert("RGB")
            except Exception as e:
                raise ValueError(f"Failed to open image from base64 string: {e}")
        else:
            raise ValueError(f"Unsupported image type: {type(img)}")
        original_images.append(img_pil.copy())
        original_sizes.append(img_pil.size)  # (width, height)
        processed_images.append(img_pil)

    # If similarity maps are provided, use them instead of computing them
    if vespa_sim_maps:
        print("Using provided similarity maps")
        # A sim map looks like this:
        # "similarities": [
        #      {
        #        "address": {
        #          "patch": "0",
        #          "querytoken": "0"
        #        },
        #        "value": 1.2599412202835083
        #      },
        # ... and so on.
        # Now turn these into a tensor of same shape as previous similarity map
        vespa_sim_map_tensor = torch.zeros(
            (
                len(vespa_sim_maps),
                query_embs.size(dim=1),
                vit_config.n_patch_per_dim,
                vit_config.n_patch_per_dim,
            )
        )
        for idx, vespa_sim_map in enumerate(vespa_sim_maps):
            for cell in vespa_sim_map["similarities"]["cells"]:
                patch = int(cell["address"]["patch"])
                if patch >= processor.image_seq_length:
                    continue
                query_token = int(cell["address"]["querytoken"])
                value = cell["value"]
                vespa_sim_map_tensor[
                    idx,
                    int(query_token),
                    int(patch) // vit_config.n_patch_per_dim,
                    int(patch) % vit_config.n_patch_per_dim,
                ] = value

        # Normalize the similarity map per query token
        similarity_map_normalized = normalize_similarity_map_per_query_token(
            vespa_sim_map_tensor
        )
    else:
        # Preprocess inputs
        print("Computing similarity maps")
        start2 = time.perf_counter()
        input_image_processed = processor.process_images(processed_images).to(device)

        # Forward passes
        with torch.no_grad():
            output_image = model.forward(**input_image_processed)

        # Remove the special tokens from the output
        output_image = output_image[:, : processor.image_seq_length, :]

        # Rearrange the output image tensor to represent the 2D grid of patches
        output_image = rearrange(
            output_image,
            "b (h w) c -> b h w c",
            h=vit_config.n_patch_per_dim,
            w=vit_config.n_patch_per_dim,
        )

        # Ensure query_embs has batch dimension
        if query_embs.dim() == 2:
            query_embs = query_embs.unsqueeze(0).to(device)
        else:
            query_embs = query_embs.to(device)

        # Compute the similarity map
        similarity_map = torch.einsum(
            "bnk,bhwk->bnhw", query_embs, output_image
        )  # Shape: (batch_size, query_tokens, h, w)

        end2 = time.perf_counter()
        print(f"Similarity map computation took: {end2 - start2} s")

        # Normalize the similarity map per query token
        similarity_map_normalized = normalize_similarity_map_per_query_token(
            similarity_map
        )

    # Collect the blended images
    start3 = time.perf_counter()
    results = []
    for idx, img in enumerate(original_images):
        original_size = original_sizes[idx]  # (width, height)
        result_per_image = {}
        for token, token_idx in token_idx_map.items():
            if is_special_token(token):
                continue

            # Get the similarity map for this image and the selected token
            sim_map = similarity_map_normalized[idx, token_idx, :, :]  # Shape: (h, w)

            # Move the similarity map to CPU, convert to float (as BFloat16 not supported by Numpy) and convert to NumPy array
            sim_map_np = sim_map.cpu().float().numpy()

            # Resize the similarity map to the original image size
            sim_map_img = Image.fromarray(sim_map_np)
            sim_map_resized = sim_map_img.resize(original_size, resample=Image.BICUBIC)

            # Convert the resized similarity map to a NumPy array
            sim_map_resized_np = np.array(sim_map_resized, dtype=np.float32)

            # Normalize the similarity map to range [0, 1]
            sim_map_min = sim_map_resized_np.min()
            sim_map_max = sim_map_resized_np.max()
            if sim_map_max - sim_map_min > 1e-6:
                sim_map_normalized = (sim_map_resized_np - sim_map_min) / (
                    sim_map_max - sim_map_min
                )
            else:
                sim_map_normalized = np.zeros_like(sim_map_resized_np)

            # Apply a colormap to the normalized similarity map
            heatmap = colormap(sim_map_normalized)  # Returns an RGBA array

            # Convert the heatmap to a PIL Image
            heatmap_uint8 = (heatmap * 255).astype(np.uint8)
            heatmap_img = Image.fromarray(heatmap_uint8)

            # Ensure both images are in RGBA mode
            original_img_rgba = img.convert("RGBA")
            heatmap_img_rgba = heatmap_img.convert("RGBA")

            # Overlay the heatmap onto the original image
            blended_img = Image.blend(
                original_img_rgba, heatmap_img_rgba, alpha=0.4
            )  # Adjust alpha as needed
            # Save the blended image to a BytesIO buffer
            buffer = io.BytesIO()
            blended_img.save(buffer, format="PNG")
            buffer.seek(0)

            # Encode the image to base64
            blended_img_base64 = base64.b64encode(buffer.read()).decode("utf-8")

            # Store the base64-encoded image
            result_per_image[token] = blended_img_base64
        results.append(result_per_image)
    end3 = time.perf_counter()
    print(f"Collecting blended images took: {end3 - start3} s")
    print(f"Total heatmap generation took: {end3 - start} s")
    return results


def get_query_embeddings_and_token_map(
    processor, model, query
) -> Tuple[torch.Tensor, dict]:
    inputs = processor.process_queries([query]).to(model.device)
    with torch.no_grad():
        embeddings_query = model(**inputs)
        q_emb = embeddings_query.to("cpu")[0]  # Extract the single embedding
    # Use this cell output to choose a token using its index
    query_tokens = processor.tokenizer.tokenize(processor.decode(inputs.input_ids[0]))
    # reverse key, values in dictionary
    print(query_tokens)
    token_to_idx = {val: idx for idx, val in enumerate(query_tokens)}
    return q_emb, token_to_idx


def format_query_results(query, response, hits=5) -> dict:
    query_time = response.json.get("timing", {}).get("searchtime", -1)
    query_time = round(query_time, 2)
    count = response.json.get("root", {}).get("fields", {}).get("totalCount", 0)
    result_text = f"Query text: '{query}', query time {query_time}s, count={count}, top results:\n"
    print(result_text)
    return response.json


async def query_vespa_default(
    app: Vespa,
    query: str,
    q_emb: torch.Tensor,
    hits: int = 3,
    timeout: str = "10s",
    **kwargs,
) -> dict:
    async with app.asyncio(connections=1, total_timeout=120) as session:
        query_embedding = format_q_embs(q_emb)

        start = time.perf_counter()
        response: VespaQueryResponse = await session.query(
            body={
                "yql": "select id,title,url,full_image,page_number,snippet,text,summaryfeatures from pdf_page where userQuery();",
                "ranking": "default",
                "query": query,
                "timeout": timeout,
                "hits": hits,
                "input.query(qt)": query_embedding,
                "presentation.timing": True,
                **kwargs,
            },
        )
        assert response.is_successful(), response.json
        stop = time.perf_counter()
        print(
            f"Query time + data transfer took: {stop - start} s, vespa said searchtime was {response.json.get('timing', {}).get('searchtime', -1)} s"
        )
        open("response.json", "w").write(json.dumps(response.json))
    return format_query_results(query, response)


async def query_vespa_bm25(
    app: Vespa,
    query: str,
    hits: int = 3,
    timeout: str = "10s",
    **kwargs,
) -> dict:
    async with app.asyncio(connections=1, total_timeout=120) as session:
        response: VespaQueryResponse = await session.query(
            body={
                "yql": "select id,title,url,full_image,page_number,snippet,text from pdf_page where userQuery();",
                "ranking": "bm25",
                "query": query,
                "timeout": timeout,
                "hits": hits,
                "presentation.timing": True,
                **kwargs,
            },
        )
        assert response.is_successful(), response.json
    return format_query_results(query, response)


def float_to_binary_embedding(float_query_embedding: dict) -> dict:
    binary_query_embeddings = {}
    for k, v in float_query_embedding.items():
        binary_vector = (
            np.packbits(np.where(np.array(v) > 0, 1, 0)).astype(np.int8).tolist()
        )
        binary_query_embeddings[k] = binary_vector
        if len(binary_query_embeddings) >= MAX_QUERY_TERMS:
            print(f"Warning: Query has more than {MAX_QUERY_TERMS} terms. Truncating.")
            break
    return binary_query_embeddings


def create_nn_query_strings(
    binary_query_embeddings: dict, target_hits_per_query_tensor: int = 20
) -> Tuple[str, dict]:
    # Query tensors for nearest neighbor calculations
    nn_query_dict = {}
    for i in range(len(binary_query_embeddings)):
        nn_query_dict[f"input.query(rq{i})"] = binary_query_embeddings[i]
    nn = " OR ".join(
        [
            f"({{targetHits:{target_hits_per_query_tensor}}}nearestNeighbor(embedding,rq{i}))"
            for i in range(len(binary_query_embeddings))
        ]
    )
    return nn, nn_query_dict


def format_q_embs(q_embs: torch.Tensor) -> dict:
    float_query_embedding = {k: v.tolist() for k, v in enumerate(q_embs)}
    return float_query_embedding


async def query_vespa_nearest_neighbor(
    app: Vespa,
    query: str,
    q_emb: torch.Tensor,
    target_hits_per_query_tensor: int = 20,
    hits: int = 3,
    timeout: str = "10s",
    **kwargs,
) -> dict:
    # Hyperparameter for speed vs. accuracy
    async with app.asyncio(connections=1, total_timeout=180) as session:
        float_query_embedding = format_q_embs(q_emb)
        binary_query_embeddings = float_to_binary_embedding(float_query_embedding)

        # Mixed tensors for MaxSim calculations
        query_tensors = {
            "input.query(qtb)": binary_query_embeddings,
            "input.query(qt)": float_query_embedding,
        }
        nn_string, nn_query_dict = create_nn_query_strings(
            binary_query_embeddings, target_hits_per_query_tensor
        )
        query_tensors.update(nn_query_dict)
        response: VespaQueryResponse = await session.query(
            body={
                **query_tensors,
                "presentation.timing": True,
                # if we use rank({nn_string}, userQuery()), dynamic summary doesn't work, see https://github.com/vespa-engine/vespa/issues/28704
                "yql": f"select id,title,snippet,text,url,full_image,page_number from pdf_page where {nn_string} or userQuery()",
                "ranking.profile": "retrieval-and-rerank",
                "timeout": timeout,
                "hits": hits,
                "query": query,
                **kwargs,
            },
        )
        assert response.is_successful(), response.json
    return format_query_results(query, response)


def is_special_token(token: str) -> bool:
    # Pattern for tokens that start with '<', numbers, whitespace, or single characters, or the string 'Question'
    pattern = re.compile(r"^<.*$|^\d+$|^\s+$|^\w$|^Question$")
    if pattern.match(token):
        return True
    return False


async def get_result_from_query(
    app: Vespa,
    processor: ColPaliProcessor,
    model: ColPali,
    query: str,
    q_embs: torch.Tensor,
    token_to_idx: Dict[str, int],
    ranking: str,
) -> Dict[str, Any]:
    # Get the query embeddings and token map
    print(query)

    print(token_to_idx)
    if ranking == "nn+colpali":
        result = await query_vespa_nearest_neighbor(app, query, q_embs)
    elif ranking == "bm25+colpali":
        result = await query_vespa_default(app, query, q_embs)
    elif ranking == "bm25":
        result = await query_vespa_bm25(app, query)
    else:
        raise ValueError(f"Unsupported ranking: {ranking}")
    # Print score, title id, and text of the results
    for idx, child in enumerate(result["root"]["children"]):
        print(
            f"Result {idx+1}: {child['relevance']}, {child['fields']['title']}, {child['fields']['id']}"
        )
    for single_result in result["root"]["children"]:
        print(single_result["fields"].keys())
    return result


def add_sim_maps_to_result(
    result: Dict[str, Any],
    model: ColPali,
    processor: ColPaliProcessor,
    query: str,
    q_embs: Any,
    token_to_idx: Dict[str, int],
) -> Dict[str, Any]:
    vit_config = load_vit_config(model)
    imgs: List[str] = []
    vespa_sim_maps: List[str] = []
    for single_result in result["root"]["children"]:
        img = single_result["fields"]["full_image"]
        if img:
            imgs.append(img)
        vespa_sim_map = single_result["fields"].get("summaryfeatures", None)
        if vespa_sim_map:
            vespa_sim_maps.append(vespa_sim_map)
    sim_map_imgs = gen_similarity_maps(
        model=model,
        processor=processor,
        device=model.device,
        vit_config=vit_config,
        query=query,
        query_embs=q_embs,
        token_idx_map=token_to_idx,
        images=imgs,
        vespa_sim_maps=vespa_sim_maps,
    )
    for single_result, sim_map_dict in zip(result["root"]["children"], sim_map_imgs):
        for token, sim_mapb64 in sim_map_dict.items():
            single_result["fields"][f"sim_map_{token}"] = sim_mapb64
    return result


if __name__ == "__main__":
    model, processor = load_model()
    vit_config = load_vit_config(model)
    query = "How many percent of source water is fresh water?"
    image_filepath = (
        Path(__file__).parent.parent
        / "static"
        / "assets"
        / "ConocoPhillips Sustainability Highlights - Nature (24-0976).png"
    )
    q_embs, token_to_idx = get_query_embeddings_and_token_map(
        processor,
        model,
        query,
    )
    figs_images = gen_similarity_maps(
        model,
        processor,
        model.device,
        vit_config,
        query=query,
        query_embs=q_embs,
        token_idx_map=token_to_idx,
        images=[image_filepath],
        vespa_sim_maps=None,
    )
    for fig_token in figs_images:
        for token, (fig, ax) in fig_token.items():
            print(f"Token: {token}")
            save_figure(fig, f"similarity_map_{token}.png")
    print("Done")