import fitz
import torch
from PIL import Image
from transformers import AutoProcessor
from transformers.models.qwen3_vl import Qwen3VLForConditionalGeneration
MODEL_ID = "mtri-admin/ZipRerank"
processor = AutoProcessor.from_pretrained(MODEL_ID, trust_remote_code=True)
model = Qwen3VLForConditionalGeneration.from_pretrained(
MODEL_ID,
torch_dtype=torch.bfloat16,
device_map="auto",
attn_implementation="flash_attention_2",
trust_remote_code=True,
).eval()
tokenizer = processor.tokenizer
def pdf_to_images(pdf_path: str, max_size: int = 1024):
"""Render every page so the longest edge is at most ``max_size`` pixels."""
doc = fitz.open(pdf_path)
images = []
for page in doc:
scale = max_size / max(page.rect.width, page.rect.height)
pix = page.get_pixmap(matrix=fitz.Matrix(scale, scale))
images.append(Image.frombytes("RGB", (pix.width, pix.height), pix.samples))
doc.close()
return images
def create_ranking_prompt(query: str, num_passages: int) -> str:
lines = [
"You are RankGPT, an intelligent assistant that can rank passages "
"based on their relevancy to the query.",
"",
f"I will provide you with {num_passages} passages as images.",
"Rank the passages based on their relevance to the search query.",
"",
"The images are provided in order: "
+ ", ".join(
f"Picture {i + 1} is passage [{chr(ord('A') + i)}]"
for i in range(num_passages)
)
+ ".",
"",
f"Search Query: {query}",
"",
"Rank the passages above based on their relevance to the search query.",
"The passages should be listed in descending order using identifiers.",
"The most relevant passages should be listed first.",
"The output format should be [A] > [B], etc.",
"Only output the ranking results, do not say anything else.",
]
return "
".join(lines)
@torch.no_grad()
def rerank_window(query: str, images):
"""Rank up to 20 page images in a single forward pass.
Returns a list of 0-based indices into ``images``, ordered best-first.
"""
assert 1 <= len(images) <= 20, "Window size must be between 1 and 20."
messages = [{
"role": "user",
"content": [{"type": "text", "text": create_ranking_prompt(query, len(images))}]
+ [{"type": "image", "image": img} for img in images],
}]
inputs = processor.apply_chat_template(
[messages],
tokenize=True,
add_generation_prompt=True,
return_dict=True,
return_tensors="pt",
)
prompt_ids = inputs["input_ids"][0].tolist()
prompt_ids.append(tokenizer.encode("[", add_special_tokens=False)[0])
input_ids = torch.tensor([prompt_ids], dtype=torch.long, device=model.device)
logits = model(
input_ids=input_ids,
attention_mask=torch.ones_like(input_ids),
pixel_values=inputs["pixel_values"].to(model.device),
image_grid_thw=inputs["image_grid_thw"].to(model.device),
).logits[0, -1, :]
letter_ids = [
tokenizer.encode(chr(ord("A") + i), add_special_tokens=False)[0]
for i in range(len(images))
]
scores = [logits[tid].item() for tid in letter_ids]
return sorted(range(len(images)), key=lambda i: scores[i], reverse=True)
pages = pdf_to_images("report.pdf", max_size=1024)
ranking = rerank_window("What is the company revenue?", pages[:20])
print("Best-first page indices:", ranking)