import fitz # PyMuPDFimport torchfrom PIL import Imagefrom transformers import AutoProcessorfrom 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", # or "sdpa" if flash-attn is unavailable 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", ) # Force the first predicted token to be a letter by appending "[" 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)