Pixel-Native RAG: A Practical Guide to Visual Document Indexing

A new tutorial from StarTrail-org introduces PixelRAG, a pixel-native retrieval-augmented generation pipeline that renders web pages and PDFs as images, divides them into overlapping tiles, and generates multimodal embeddings with SigLIP, CLIP, or Qwen3-VL backends, storing vectors in a FAISS index. The system integrates OCR-based BM25 scoring with reciprocal rank fusion, aggregates tile-level evidence into document-level results, and exposes a FastAPI search service, with evaluation using Recall@k and mean reciprocal rank.

In this tutorial, we build a complete pixel https://github.com/StarTrail-org/PixelRAG -native retrieval-augmented generation pipeline from scratch and examine how document retrieval works without relying on conventional HTML parsing, text extraction, or fixed chunking strategies. We render web pages and PDF documents as images, divide them into overlapping tiles, generate multimodal embeddings with SigLIP, CLIP, or an optional Qwen3-VL backend, and store the resulting vectors in a FAISS index for efficient similarity search. We also strengthen retrieval with OCR-based BM25 scoring and reciprocal rank fusion, aggregate tile-level evidence into document-level results, and expose the system through a FastAPI search service. Along the way, we evaluate retrieval quality using Recall@k and mean reciprocal rank, train a lightweight residual adapter with contrastive learning, visualize retrieved screenshots, and optionally pass the strongest evidence tiles to a vision-language model for grounded answer generation. python import os import sys import io import re import json import time import math import shutil import hashlib import asyncio import logging import argparse import threading import subprocess from pathlib import Path from dataclasses import dataclass, field, asdict from typing import List, Dict, Any, Optional, Tuple @dataclass class Config: urls: List str = field default factory=lambda: "https://en.wikipedia.org/wiki/Retrieval-augmented generation", "https://en.wikipedia.org/wiki/Vector database", "https://en.wikipedia.org/wiki/Transformer deep learning architecture ", "https://en.wikipedia.org/wiki/Photosynthesis", "https://en.wikipedia.org/wiki/Delhi", include synthetic pdf: bool = True tile width: int = 1024 tile height: int = 1024 tile overlap: int = 128 device scale: float = 1.0 max page height: int = 24000 max tiles per doc: int = 12 min tile height: int = 200 blank std threshold: float = 6.0 dedup hamming: int = 4 nav timeout ms: int = 60000 headless args: List str = field default factory=lambda: "--no-sandbox", "--disable-dev-shm-usage", "--hide-scrollbars", "--disable-gpu", "--force-color-profile=srgb", "--font-render-hinting=none", backend: str = "siglip" model id: str = "google/siglip-base-patch16-224" qwen model id: str = "Qwen/Qwen3-VL-Embedding-2B" embed batch size: int = 8 embed image size: Optional int = None index dir: str = "./pixel index" ivf threshold: int = 2000 ivf nprobe: int = 16 top k tiles: int = 20 n docs: int = 5 use ocr hybrid: bool = True rrf k: int = 60 dense weight: float = 1.0 sparse weight: float = 1.0 enable server: bool = True server port: int = 8000 enable eval: bool = True enable adapter train: bool = True enable vlm answer: bool = False vlm model id: str = "Qwen/Qwen2.5-VL-3B-Instruct" show plots: bool = True work dir: str = "./pixelrag work" seed: int = 0 CFG = Config EVAL QUERIES: List Tuple str, str = "how do plants convert sunlight into chemical energy", "Photosynthesis" , "chlorophyll light dependent reactions", "Photosynthesis" , "converting scanned images of text into machine readable characters", "Optical character" , "approximate nearest neighbour search over embeddings", "Vector database" , "self-attention multi-head architecture", "Transformer" , "grounding a language model with retrieved documents", "Retrieval-augmented" , "capital territory of india red fort", "Delhi" , logging.basicConfig level=logging.INFO, format="% asctime s | % levelname -7s | % message s", datefmt="%H:%M:%S" log = logging.getLogger "pixelrag" for noisy in "urllib3", "PIL", "matplotlib", "httpx", "asyncio", "uvicorn.error" : logging.getLogger noisy .setLevel logging.WARNING IN COLAB = "google.colab" in sys.modules def pip pkgs: str - None: """Install quietly; never explode the notebook on a single bad wheel.""" cmd = sys.executable, "-m", "pip", "install", "-q", "--disable-pip-version-check", pkgs subprocess.run cmd, check=False, stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT def have mod: str - bool: import importlib.util return importlib.util.find spec mod is not None def ensure deps cfg: Config - None: log.info "Installing dependencies first run only, ~2-4 min ..." wanted = for mod, pkg in "PIL", "pillow" , "numpy", "numpy" , "faiss", "faiss-cpu" , "fitz", "pymupdf" , "transformers", "transformers" , "fastapi", "fastapi" , "uvicorn", "uvicorn" , "requests", "requests" , "matplotlib", "matplotlib" , "tqdm", "tqdm" , "rank bm25", "rank-bm25" , "playwright", "playwright" , "sentencepiece", "sentencepiece" , : if not have mod : wanted.append pkg if cfg.use ocr hybrid and not have "pytesseract" : wanted.append "pytesseract" if wanted: pip wanted if not have "torch" : log.warning "torch not found — installing CPU wheel Colab normally ships torch ." pip "torch", "torchvision" if cfg.use ocr hybrid and shutil.which "tesseract" is None: log.info "Installing tesseract-ocr system package..." subprocess.run "apt-get -qq update && apt-get -qq install -y tesseract-ocr", shell=True, check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL if shutil.which "tesseract" is None: log.warning "tesseract unavailable - hybrid retrieval will run dense-only." cfg.use ocr hybrid = False marker = Path cfg.work dir / ".chromium ok" if not marker.exists : log.info "Downloading Playwright Chromium..." r = subprocess.run sys.executable, "-m", "playwright", "install", "--with-deps", "chromium" , capture output=True, text=True if r.returncode = 0: r = subprocess.run sys.executable, "-m", "playwright", "install", "chromium" , capture output=True, text=True if r.returncode == 0: marker.parent.mkdir parents=True, exist ok=True marker.write text "ok" else: log.warning "Chromium install failed - falling back to the text renderer.\n%s", r.stderr or "" -600: log.info "Dependencies ready." def run async coro : """ Run a coroutine from a Jupyter/Colab cell. Colab already owns a running event loop, which makes Playwright's sync API raise. Rather than monkey-patching with nest asyncio, we hand the coroutine to a private loop on a private thread — the most robust option. """ box: Dict str, Any = {} def runner : loop = asyncio.new event loop asyncio.set event loop loop try: box "value" = loop.run until complete coro except BaseException as exc: box "error" = exc finally: try: loop.run until complete loop.shutdown asyncgens finally: loop.close t = threading.Thread target= runner, daemon=True t.start t.join if "error" in box: raise box "error" return box.get "value" We define the global configuration, evaluation queries, logging behavior, and runtime settings for the PixelRAG pipeline. We install the required Python and system dependencies, including Playwright, Chromium, Tesseract, FAISS, and transformer libraries. We also create an asynchronous execution helper that allows browser-rendering coroutines to run reliably inside Google Colab and Jupyter environments. @dataclass class Tile: tile id: str doc id: str source: str kind: str page: int seq: int y0: int y1: int path: str ocr text: str = "" title: str = "" def doc id from source src: str - str: tail = src.rstrip "/" .split "/" -1 or src tail = re.sub r"\. html?|pdf|png|jpg $", "", tail, flags=re.I return re.sub r" ^A-Za-z0-9 .\- +", " ", tail :80 or hashlib.md5 src.encode .hexdigest :10 def ahash img, size: int = 8 - int: """64-bit average hash — cheap near-duplicate detection for repeated headers.""" import numpy as np g = img.convert "L" .resize size, size a = np.asarray g, dtype="float32" bits = a a.mean .flatten out = 0 for b in bits: out = out << 1 | int b return out def hamming a: int, b: int - int: return bin a ^ b .count "1" def is informative img, cfg: Config - bool: """Reject blank / solid-colour tiles before they ever reach the GPU.""" import numpy as np a = np.asarray img.convert "L" , dtype="float32" return float a.std = cfg.blank std threshold def save tile img, out dir: Path, name: str - str: out dir.mkdir parents=True, exist ok=True p = out dir / f"{name}.png" img.convert "RGB" .save p, format="PNG", optimize=True return str p def slice image to tiles img, cfg: Config, , doc id: str, source: str, kind: str, page: int, out dir: Path, start seq: int = 0, seen hashes: Optional List int = None, title: str = "" - List Tile : """Vertical sliding window with overlap. Used for PDFs and text fallback.""" from PIL import Image seen hashes = seen hashes if seen hashes is not None else W, H = img.size if W = cfg.tile width: new h = max 1, int H cfg.tile width / W img = img.resize cfg.tile width, new h W, H = img.size step = max 1, cfg.tile height - cfg.tile overlap tiles: List Tile = y, seq = 0, start seq while y < H and seq - start seq < cfg.max tiles per doc: h = min cfg.tile height, H - y if h < cfg.min tile height and seq start seq: break crop = img.crop 0, y, W, y + h if is informative crop, cfg : hsh = ahash crop if all hamming hsh, s cfg.dedup hamming for s in seen hashes : seen hashes.append hsh tid = f"{doc id} p{page} t{seq}" tiles.append Tile tile id=tid, doc id=doc id, source=source, kind=kind, page=page, seq=seq, y0=y, y1=y + h, title=title, path= save tile crop, out dir, tid , seq += 1 y += step return tiles JS AUTOSCROLL = """ async = { await new Promise resolve = { let y = 0; const timer = setInterval = { window.scrollBy 0, 800 ; y += 800; if y = document.body.scrollHeight || y 40000 { clearInterval timer ; window.scrollTo 0, 0 ; setTimeout resolve, 250 ; } }, 40 ; } ; } """ JS FLATTEN = """ = { document.querySelectorAll ' ' .forEach el = { const s = getComputedStyle el ; if s.position === 'fixed' || s.position === 'sticky' el.style.position = 'absolute'; } ; document.querySelectorAll ' role="dialog" , .cookie, cookie-banner, .cc-banner' .forEach el = el.remove ; } """ CSS CLEANUP = """ { animation: none important; transition: none important; scroll-behavior: auto important; } html { -webkit-font-smoothing: antialiased; } video, iframe src ="youtube" { visibility: hidden important; } """ UA = "Mozilla/5.0 X11; Linux x86 64 AppleWebKit/537.36 KHTML, like Gecko " "Chrome/124.0 Safari/537.36 PixelRAG-Tutorial/1.0" async def render urls async urls: List str , cfg: Config, out dir: Path - List Tile : from playwright.async api import async playwright from PIL import Image all tiles: List Tile = async with async playwright as pw: browser = await pw.chromium.launch headless=True, args=cfg.headless args ctx = await browser.new context viewport={"width": cfg.tile width, "height": cfg.tile height}, device scale factor=cfg.device scale, user agent= UA, java script enabled=True, for url in urls: doc id = doc id from source url page = await ctx.new page try: await page.goto url, wait until="domcontentloaded", timeout=cfg.nav timeout ms try: await page.wait for load state "networkidle", timeout=12000 except Exception: pass await page.evaluate JS AUTOSCROLL await page.add style tag content= CSS CLEANUP await page.evaluate JS FLATTEN title = await page.title or doc id height = await page.evaluate " = Math.max document.body.scrollHeight, " "document.documentElement.scrollHeight " height = int min height, cfg.max page height step = max 1, cfg.tile height - cfg.tile overlap seen: List int = y, seq = 0, 0 while y < height and seq < cfg.max tiles per doc: h = min cfg.tile height, height - y if h < cfg.min tile height and seq 0: break buf = await page.screenshot full page=True, type="png", clip={"x": 0, "y": y, "width": cfg.tile width, "height": h} img = Image.open io.BytesIO buf .convert "RGB" if img.size 0 = cfg.tile width: img = img.resize cfg.tile width, max 1, int img.size 1 cfg.tile width / img.size 0 if is informative img, cfg : hsh = ahash img if all hamming hsh, s cfg.dedup hamming for s in seen : seen.append hsh tid = f"{doc id} p0 t{seq}" all tiles.append Tile tile id=tid, doc id=doc id, source=url, kind="web", page=0, seq=seq, y0=y, y1=y + h, title=title, path= save tile img, out dir, tid seq += 1 y += step log.info " rendered %-34s - %2d tiles page %dpx ", doc id, seq, height except Exception as exc: log.warning " FAILED %s %s ", url, type exc . name finally: await page.close await ctx.close await browser.close return all tiles def render urls urls: List str , cfg: Config, out dir: Path - List Tile : """Screenshot every URL into tiles; degrade to the text renderer on failure.""" try: tiles = run async render urls async urls, cfg, out dir if tiles: return tiles log.warning "Browser produced no tiles — using text-render fallback." except Exception as exc: log.warning "Playwright unavailable %s: %s — using text-render fallback.", type exc . name , str exc :160 return t for u in urls for t in render url as text u, cfg, out dir def strip html html: str - str: html = re.sub r" ?is < script|style|nav|footer|header|noscript . ?</\1 ", " ", html html = re.sub r" ?s < --. ?-- ", " ", html html = re.sub r" ?i </ p|div|h 1-6 |li|tr|br ", "\n", html text = re.sub r" ?s < ^ + ", " ", html for a, b in " ", " " , "&", "&" , "<", "<" , " ", " " , """, '"' : text = text.replace a, b text = re.sub r"\ \d+\ ", "", text text = re.sub r" \t +", " ", text return re.sub r"\n{2,}", "\n", text .strip def mono font size: int = 20 : from PIL import ImageFont for cand in "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", "/usr/share/fonts/truetype/liberation/LiberationSans-Regular.ttf" : if os.path.exists cand : return ImageFont.truetype cand, size try: import matplotlib.font manager as fm return ImageFont.truetype fm.findfont "DejaVu Sans" , size except Exception: return ImageFont.load default def text to image text: str, cfg: Config, title: str = "" - Any: """Render plain text onto a tall white canvas — a browser-free stand-in.""" from PIL import Image, ImageDraw font, tfont = mono font 20 , mono font 30 pad, lh, wrap = 40, 30, max 20, cfg.tile width - 80 // 11 lines: List str = for para in text.split "\n" : para = para.strip if not para: continue while len para wrap: cut = para.rfind " ", 0, wrap cut = cut if cut 0 else wrap lines.append para :cut para = para cut: .lstrip lines.append para lines = lines :900 height = pad 2 + 60 + lh len lines img = Image.new "RGB", cfg.tile width, max cfg.tile height, height , "white" d = ImageDraw.Draw img d.text pad, pad , title :60 , font=tfont, fill= 15, 15, 15 for i, ln in enumerate lines : d.text pad, pad + 60 + i lh , ln, font=font, fill= 35, 35, 35 return img def render url as text url: str, cfg: Config, out dir: Path - List Tile : import requests doc id = doc id from source url try: r = requests.get url, timeout=30, headers={"User-Agent": UA} r.raise for status body = strip html r.text m = re.search r" ?is <title . ? </title ", r.text title = m.group 1 .strip if m else doc id except Exception as exc: log.warning " fetch failed for %s %s ", url, type exc . name return img = text to image body, cfg, title=title log.info " text-rendered %-30s - canvas %dpx", doc id, img.size 1 return slice image to tiles img, cfg, doc id=doc id, source=url, kind="text", page=0, out dir=out dir, title=title def render pdf pdf path: str, cfg: Config, out dir: Path, dpi: int = 150 - List Tile : import fitz from PIL import Image doc id = doc id from source pdf path tiles: List Tile = with fitz.open pdf path as doc: title = doc.metadata or {} .get "title" or doc id n pages = doc.page count for pno in range n pages : pix = doc pno .get pixmap dpi=dpi img = Image.frombytes "RGB", pix.width, pix.height , pix.samples tiles += slice image to tiles img, cfg, doc id=doc id, source=pdf path, kind="pdf", page=pno, out dir=out dir, title=title log.info " rendered %-34s - %2d tiles %d pages ", doc id, len tiles , n pages return tiles def make synthetic pdf path: Path - str: """A tiny PDF so the tutorial always exercises the PDF path, offline or not.""" import fitz body = "PixelRAG Internal Note", 22 , "", 12 , "Why pixel-native retrieval?", 16 , "Parsers are per-site glue code. A renderer is one code path for every", 11 , "document type: HTML, PDF, scanned fax, spreadsheet export, dashboard.", 11 , "", 11 , "Tiling policy", 16 , "Tiles are 1024x1024 with 128px of vertical overlap. Overlap keeps a", 11 , "sentence or table row from being split across two embeddings, which is", 11 , "the single biggest source of recall loss in naive screenshot pipelines.", 11 , "", 11 , "Serving", 16 , "FAISS inner-product over L2-normalised vectors equals cosine similarity.", 11 , "Tile scores are max-pooled per document so one strong tile can surface", 11 , "a long page, mirroring late-interaction retrieval behaviour.", 11 , "", 11 , "The mitochondria reference is a joke; the overlap advice is not.", 11 , doc = fitz.open page = doc.new page y = 72 for line, size in body: page.insert text 72, y , line, fontsize=size, fontname="helv" y += size + 8 doc.save str path doc.close return str path We create the document-rendering layer that converts web pages, text content, and PDF files into structured image tiles. We capture web pages with Playwright, clean distracting page elements, apply overlapping vertical slicing, and remove blank or duplicate tiles. We also provide text-rendering and synthetic-PDF fallbacks so the pipeline continues to operate when browser rendering or external content is unavailable. php def ocr tiles tiles: List Tile , cfg: Config - None: if not cfg.use ocr hybrid: return try: import pytesseract from PIL import Image except Exception: log.warning "pytesseract missing - dense-only retrieval." cfg.use ocr hybrid = False return from tqdm.auto import tqdm t0 = time.time for t in tqdm tiles, desc="OCR", unit="tile" : try: raw = pytesseract.image to string Image.open t.path , config="--psm 6" t.ocr text = re.sub r"\s+", " ", raw .strip :4000 except Exception: t.ocr text = "" log.info "OCR over %d tiles in %.1fs", len tiles , time.time - t0 def torch device - str: import torch if torch.cuda.is available : return "cuda" if getattr torch.backends, "mps", None and torch.backends.mps.is available : return "mps" return "cpu" class DualEncoderBackend: """ SigLIP / CLIP image-text dual encoder. Honest caveat: these encoders were trained on natural images with short captions 64-77 token text towers . They understand a screenshot's gist — layout, topic, figures — not its fine print. That is exactly why upstream PixelRAG uses Qwen3-VL-Embedding-2B plus a LoRA trained on screenshots. Sections §9 OCR hybrid and §10 adapter exist to close part of the gap on hardware that can't host a 2B VLM. """ def init self, cfg: Config : import torch from transformers import AutoModel, AutoProcessor self.cfg = cfg self.device = torch device self.dtype = torch.float16 if self.device == "cuda" else torch.float32 self.model id = cfg.model id if cfg.backend = "clip" else "openai/clip-vit-base-patch32" log.info "Loading embedding model %s on %s %s ", self.model id, self.device, str self.dtype .replace "torch.", "" self.processor = AutoProcessor.from pretrained self.model id self.model = AutoModel.from pretrained self.model id, torch dtype=self.dtype self.model.to self.device .eval self.is siglip = "siglip" in self.model id.lower self.dim = int getattr self.model.config, "projection dim", 0 or getattr self.model.config.text config, "hidden size", 512 self.name = f"{'siglip' if self.is siglip else 'clip'}:{self.model id}" @staticmethod def l2 x : import numpy as np n = np.linalg.norm x, axis=-1, keepdims=True return x / np.clip n, 1e-12, None .astype "float32" def embed images self, images: List Any , bs: Optional int = None : import torch, numpy as np from tqdm.auto import tqdm bs = bs or self.cfg.embed batch size out = for i in tqdm range 0, len images , bs , desc="embed:image", unit="batch" : batch = images i:i + bs inputs = self.processor images=batch, return tensors="pt" inputs = {k: v.to self.device, self.dtype if v.is floating point else v.dtype for k, v in inputs.items } with torch.no grad : feats = self.model.get image features inputs out.append feats.float .cpu .numpy return self. l2 np.concatenate out, 0 if out else np.zeros 0, self.dim , "float32" def embed texts self, texts: List str , bs: Optional int = None : import torch, numpy as np bs = bs or max 16, self.cfg.embed batch size out = for i in range 0, len texts , bs : batch = t if t.strip else " " for t in texts i:i + bs kw = dict text=batch, return tensors="pt", truncation=True kw.update padding="max length", max length=64 if self.is siglip else kw.update padding=True inputs = self.processor kw inputs = {k: v.to self.device for k, v in inputs.items } with torch.no grad : feats = self.model.get text features inputs out.append feats.float .cpu .numpy return self. l2 np.concatenate out, 0 if out else np.zeros 0, self.dim , "float32" class Qwen3VLEmbeddingBackend: """ Opt-in backend matching upstream Qwen/Qwen3-VL-Embedding-2B . Needs a recent transformers = 4.57 and ~8 GB of VRAM in fp16. It embeds text and images into one space by mean-pooling the last hidden state of a VLM prompt, which is why it handles dense screenshot text far better than a CLIP-style tower. """ def init self, cfg: Config : import torch from transformers import AutoModel, AutoProcessor self.cfg = cfg self.device = torch device self.dtype = torch.float16 if self.device == "cuda" else torch.float32 mid = cfg.qwen model id log.info "Loading %s this is a large download ...", mid self.processor = AutoProcessor.from pretrained mid, trust remote code=True self.model = AutoModel.from pretrained mid, torch dtype=self.dtype, trust remote code=True .to self.device .eval self.dim = int self.model.config.hidden size self.name = f"qwen3vl:{mid}" def pool self, hidden, mask : import torch m = mask.unsqueeze -1 .to hidden.dtype return hidden m .sum 1 / m.sum 1 .clamp min=1e-6 def encode self, proc kwargs : import torch, numpy as np inputs = self.processor return tensors="pt", padding=True, proc kwargs inputs = {k: v.to self.device if hasattr v, "to" else v for k, v in inputs.items } with torch.no grad : out = self.model inputs, output hidden states=True hidden = out.hidden states -1 if getattr out, "hidden states", None is not None \ else out.last hidden state vec = self. pool hidden, inputs "attention mask" .float .cpu .numpy return DualEncoderBackend. l2 vec def embed images self, images: List Any , bs: Optional int = None : import numpy as np from tqdm.auto import tqdm bs = bs or max 1, self.cfg.embed batch size // 4 chunks = for i in tqdm range 0, len images , bs , desc="embed:image", unit="batch" : batch = images i:i + bs prompt = "Describe this document screenshot for retrieval." len batch chunks.append self. encode text=prompt, images=batch return np.concatenate chunks, 0 def embed texts self, texts: List str , bs: Optional int = None : import numpy as np bs = bs or 8 chunks = self. encode text= f"Query: {t}" for t in texts i:i + bs for i in range 0, len texts , bs return np.concatenate chunks, 0 if chunks else np.zeros 0, self.dim , "float32" def build backend cfg: Config : if cfg.backend == "qwen3vl": try: return Qwen3VLEmbeddingBackend cfg except Exception as exc: log.warning "Qwen3-VL backend failed %s: %s - falling back to SigLIP.", type exc . name , str exc :200 cfg.backend = "siglip" return DualEncoderBackend cfg def embed tiles tiles: List Tile , backend, cfg: Config : from PIL import Image import numpy as np vecs = bs = cfg.embed batch size for i in range 0, len tiles , bs : imgs = Image.open t.path .convert "RGB" for t in tiles i:i + bs vecs.append backend.embed images imgs, bs=bs for im in imgs: im.close return np.concatenate vecs, 0 if vecs else np.zeros 0, backend.dim , "float32" We extract OCR text from each rendered tile to support sparse retrieval and automatic training-pair generation. We implement SigLIP, CLIP, and Qwen3-VL embedding backends that place text queries and document screenshots within a shared vector space. We then process the tile images in batches and generate normalized embeddings that are ready for similarity indexing. class PixelIndex: """ Inner-product FAISS index over L2-normalised vectors == cosine similarity . Flat below ivf threshold vectors exact, no training ; IVF above it sub-linear, needs training + nprobe tuning . Raw vectors are also kept in memory so §10 can re-project them after adapter training without re-running the encoder. """ def init self, dim: int, cfg: Config : self.dim, self.cfg = dim, cfg self.index = None self.metas: List Dict str, Any = self.vectors = None self. bm25 = None self. bm25 corpus: List List str = def build self, vectors, tiles: List Tile - "PixelIndex": import faiss, numpy as np vectors = np.ascontiguousarray vectors.astype "float32" n = vectors.shape 0 if n == 0: raise RuntimeError "No vectors to index — did rendering produce any tiles?" if n = self.cfg.ivf threshold: nlist = max 4, min 4096, int 4 math.sqrt n quant = faiss.IndexFlatIP self.dim base = faiss.IndexIVFFlat quant, self.dim, nlist, faiss.METRIC INNER PRODUCT base.train vectors base.nprobe = self.cfg.ivf nprobe log.info "FAISS IndexIVFFlat n=%d nlist=%d nprobe=%d", n, nlist, base.nprobe else: base = faiss.IndexFlatIP self.dim log.info "FAISS IndexFlatIP n=%d dim=%d exact search ", n, self.dim self.index = faiss.IndexIDMap2 base self.index.add with ids vectors, np.arange n .astype "int64" self.vectors = vectors self.metas = asdict t for t in tiles self. fit bm25 return self def fit bm25 self - None: if not self.cfg.use ocr hybrid: return try: from rank bm25 import BM25Okapi except Exception: return self. bm25 corpus = re.findall r" a-z0-9 +", m.get "ocr text", "" + " " + m.get "title", "" .lower for m in self.metas if any self. bm25 corpus : self. bm25 = BM25Okapi c or " " for c in self. bm25 corpus log.info "BM25 fitted over OCR sidecar %d docs ", len self. bm25 corpus def search dense self, qvecs, k: int : import numpy as np scores, ids = self.index.search np.ascontiguousarray qvecs.astype "float32" , k return scores, ids def search sparse self, query: str, k: int - List Tuple int, float : if self. bm25 is None: return import numpy as np toks = re.findall r" a-z0-9 +", query.lower if not toks: return s = np.asarray self. bm25.get scores toks top = np.argsort -s :k return int i , float s i for i in top if s i 0 def save self, out dir: str - None: import faiss, numpy as np p = Path out dir p.mkdir parents=True, exist ok=True faiss.write index self.index, str p / "tiles.faiss" np.save p / "vectors.npy", self.vectors p / "metas.jsonl" .write text "\n".join json.dumps m for m in self.metas p / "manifest.json" .write text json.dumps {"dim": self.dim, "n": len self.metas , "created": time.time , "config": asdict self.cfg }, indent=2 log.info "Index saved to %s %d tiles ", p.resolve , len self.metas @classmethod def load cls, out dir: str, cfg: Config - "PixelIndex": import faiss, numpy as np p = Path out dir man = json.loads p / "manifest.json" .read text obj = cls man "dim" , cfg obj.index = faiss.read index str p / "tiles.faiss" obj.vectors = np.load p / "vectors.npy" obj.metas = json.loads l for l in p / "metas.jsonl" .read text .splitlines if l obj. fit bm25 return obj def reproject self, new vectors - None: """Swap in re-embedded vectors used after adapter training in §10 .""" tiles = Tile m for m in self.metas self.build new vectors, tiles def build index cfg: Config - Tuple PixelIndex, Any, List Tile : work = Path cfg.work dir tiles dir = work / "tiles" tiles dir.mkdir parents=True, exist ok=True log.info "=" 74 log.info "STAGE 1/4 RENDER documents - image tiles " log.info "=" 74 tiles: List Tile = render urls cfg.urls, cfg, tiles dir if cfg.include synthetic pdf: pdf path = make synthetic pdf work / "pixelrag note.pdf" tiles += render pdf pdf path, cfg, tiles dir if not tiles: raise RuntimeError "Rendering produced zero tiles. Check network access." log.info "Total tiles: %d across %d documents", len tiles , len {t.doc id for t in tiles} log.info "=" 74 log.info "STAGE 2/4 OCR SIDECAR for hybrid retrieval + pair mining " log.info "=" 74 ocr tiles tiles, cfg log.info "=" 74 log.info "STAGE 3/4 EMBED tiles - vectors " log.info "=" 74 backend = build backend cfg t0 = time.time vecs = embed tiles tiles, backend, cfg log.info "Embedded %d tiles - %s in %.1fs %.2f tiles/s ", vecs.shape 0 , vecs.shape, time.time - t0, vecs.shape 0 / max 1e-6, time.time - t0 log.info "=" 74 log.info "STAGE 4/4 INDEX vectors - FAISS " log.info "=" 74 index = PixelIndex vecs.shape 1 , cfg .build vecs, tiles index.save cfg.index dir return index, backend, tiles We construct the PixelIndex class and store the normalized tile embeddings inside a FAISS inner-product index. We support exact flat search for smaller datasets, IVF-based search for larger collections, BM25 indexing over OCR text, and persistent storage of vectors and metadata. We also orchestrate the complete indexing pipeline by rendering documents, running OCR, generating embeddings, building the index, and saving all outputs to disk. python def search query: str, index: PixelIndex, backend, cfg: Config, n docs: Optional int = None - List Dict str, Any : import numpy as np n docs = n docs or cfg.n docs k = min cfg.top k tiles, len index.metas qv = backend.embed texts query dscores, dids = index.search dense qv, k dense = int i , float s for i, s in zip dids 0 , dscores 0 if i = 0 fused: Dict int, float = {} for rank, tid, in enumerate dense : fused tid = fused.get tid, 0.0 + cfg.dense weight / cfg.rrf k + rank + 1 sparse = index.search sparse query, k if cfg.use ocr hybrid else for rank, tid, in enumerate sparse : fused tid = fused.get tid, 0.0 + cfg.sparse weight / cfg.rrf k + rank + 1 dense lookup = dict dense tile hits = sorted fused.items , key=lambda kv: -kv 1 per doc: Dict str, Dict str, Any = {} for tid, fscore in tile hits: m = index.metas tid d = per doc.setdefault m "doc id" , { "doc id": m "doc id" , "title": m.get "title" or m "doc id" , "source": m "source" , "kind": m "kind" , "score": 0.0, "dense score": 0.0, "tiles": , } d "score" = max d "score" , fscore d "dense score" = max d "dense score" , dense lookup.get tid, 0.0 if len d "tiles" < 3: d "tiles" .append { "tile id": m "tile id" , "path": m "path" , "seq": m "seq" , "page": m "page" , "y0": m "y0" , "y1": m "y1" , "rrf": round fscore, 6 , "cosine": round dense lookup.get tid, 0.0 , 4 , "snippet": m.get "ocr text", "" or "" :220 , } return sorted per doc.values , key=lambda d: -d "score" :n docs def pretty print query: str, results: List Dict str, Any - None: print f"\n\033 1mQ: {query}\033 0m" if not results: print " no hits " return for i, r in enumerate results, 1 : print f" {i}. {r 'score' :.4f} rrf | {r 'dense score' :.3f} cos " f"{r 'title' :64 } {r 'kind' } " top = r "tiles" 0 print f" tile {top 'tile id' } y={top 'y0' }-{top 'y1' }" if top "snippet" : print f" \033 2m{top 'snippet' :150 }...\033 0m" class SearchServer: """FastAPI + uvicorn on a background thread, mirroring upstream's POST /search.""" def init self, index: PixelIndex, backend, cfg: Config : from fastapi import FastAPI from pydantic import BaseModel class Query BaseModel : text: str class SearchRequest BaseModel : queries: List Query n docs: int = cfg.n docs app = FastAPI title="PixelRAG tutorial ", version="1.0" @app.get "/health" def health : return {"status": "ok", "tiles": len index.metas , "docs": len {m "doc id" for m in index.metas} , "backend": getattr backend, "name", "unknown" } @app.post "/search" def do search req: SearchRequest : return {"results": {"query": q.text, "docs": search q.text, index, backend, cfg, req.n docs } for q in req.queries } self.app, self.cfg = app, cfg self.thread: Optional threading.Thread = None self.server = None def start self - bool: import uvicorn, requests config = uvicorn.Config self.app, host="127.0.0.1", port=self.cfg.server port, log level="error" self.server = uvicorn.Server config self.thread = threading.Thread target=self.server.run, daemon=True self.thread.start for in range 40 : time.sleep 0.25 try: if requests.get f"http://127.0.0.1:{self.cfg.server port}/health", timeout=2 .ok: log.info "Search API live on http://127.0.0.1:%d", self.cfg.server port return True except Exception: continue log.warning "Server did not come up in time." return False def stop self - None: if self.server: self.server.should exit = True if self.thread: self.thread.join timeout=5 We implement hybrid retrieval by combining dense vector rankings and OCR-based BM25 rankings through reciprocal rank fusion. We aggregate matching tiles into document-level results while retaining the strongest evidence tiles, similarity scores, and OCR snippets for inspection. We also expose the retrieval system through a FastAPI server with health and search endpoints that run on a background Uvicorn thread. python def evaluate index: PixelIndex, backend, cfg: Config, queries: List Tuple str, str = EVAL QUERIES, label: str = "eval", quiet: bool = False - Dict str, float : ranks: List Optional int = for q, want in queries: docs = search q, index, backend, cfg, n docs=10 hit = next i for i, d in enumerate docs if want.lower in d "doc id" .lower , None ranks.append hit if not quiet: got = docs 0 "doc id" if docs else "-" mark = "OK " if hit == 0 else f"@{hit + 1}" if hit is not None else "MISS" print f" {mark: 4} {q :56 :<58} - {got :32 }" n = len ranks m = { "recall@1": sum r == 0 for r in ranks / n, "recall@3": sum r is not None and r < 3 for r in ranks / n, "recall@5": sum r is not None and r < 5 for r in ranks / n, "mrr": sum 1.0 / r + 1 for r in ranks if r is not None / n, } print f" \033 1m{label}\033 0m R@1={m 'recall@1' :.2f} R@3={m 'recall@3' :.2f} " f"R@5={m 'recall@5' :.2f} MRR={m 'mrr' :.3f}" return m def mine training pairs tiles: List Tile , max per tile: int = 2 - List Tuple str, int : """Weak supervision: pseudo-queries from a tile's own OCR text / title.""" import random rng = random.Random 0 pairs: List Tuple str, int = for idx, t in enumerate tiles : text = t.ocr text or "" .strip cands: List str = if len text 80: words = text.split for in range max per tile : if len words <= 14: break s = rng.randint 0, len words - 14 span = " ".join words s:s + rng.randint 8, 14 if len span 30: cands.append span if t.title: cands.append t.title for c in cands :max per tile : pairs.append c, idx return pairs class ResidualAdapter: """Shared two-layer residual MLP applied to both query and tile vectors.""" def init self, dim: int, hidden: int = 512, device: str = "cpu" : import torch import torch.nn as nn self.device = device self.net = nn.Sequential nn.Linear dim, hidden , nn.GELU , nn.Linear hidden, dim .to device for p in self.net -1 .parameters : torch.nn.init.zeros p self.logit scale = torch.nn.Parameter torch.tensor 2.996, device=device self.dim = dim def forward t self, x : import torch y = x + self.net x return torch.nn.functional.normalize y, dim=-1 def apply np self, arr : import torch, numpy as np with torch.no grad : t = torch.from numpy np.ascontiguousarray arr.astype "float32" .to self.device return self.forward t t .cpu .numpy .astype "float32" def train adapter index: PixelIndex, backend, tiles: List Tile , cfg: Config, epochs: int = 12, batch: int = 24, lr: float = 1e-4 : import torch, numpy as np pairs = mine training pairs tiles if len pairs < 32: log.warning "Only %d mined pairs — skipping adapter training " " enable OCR or add documents .", len pairs return None log.info "Mined %d pseudo-query, tile pairs from %d tiles", len pairs , len tiles q texts = p 0 for p in pairs t idx = np.array p 1 for p in pairs , dtype="int64" log.info "Pre-embedding pseudo-queries frozen encoder, done once ..." Q = torch.from numpy backend.embed texts q texts V = torch.from numpy index.vectors device = "cuda" if torch.cuda.is available else "cpu" ad = ResidualAdapter index.dim, device=device Q, V = Q.to device , V.to device opt = torch.optim.AdamW list ad.net.parameters + ad.logit scale , lr=lr, weight decay=1e-2 n = len pairs doc ids = torch.from numpy t idx .to device for ep in range epochs : perm = torch.randperm n, device=device total, steps = 0.0, 0 for i in range 0, n, batch : sel = perm i:i + batch if sel.numel < 4: continue qb = ad.forward t Q sel docs = doc ids sel vb = ad.forward t V docs logits = ad.logit scale.exp .clamp max=100 qb @ vb.T same = docs :, None == docs None, : eye = torch.eye len sel , dtype=torch.bool, device=device logits = logits.masked fill same & ~eye, float "-inf" labels = torch.arange len sel , device=device loss = 0.5 torch.nn.functional.cross entropy logits, labels + torch.nn.functional.cross entropy logits.T, labels opt.zero grad loss.backward torch.nn.utils.clip grad norm ad.net.parameters , 1.0 opt.step total += loss.detach .item steps += 1 if ep % 3 == 0 or ep == epochs - 1: log.info " epoch %2d/%d InfoNCE loss %.4f", ep + 1, epochs, total / max steps, 1 return ad class AdaptedBackend: """Wraps a frozen backend so queries pass through the trained adapter.""" def init self, backend, adapter: ResidualAdapter : self.backend, self.adapter = backend, adapter self.dim = backend.dim self.name = f"{getattr backend, 'name', 'backend' }+adapter" def embed texts self, texts, bs=None : return self.adapter.apply np self.backend.embed texts texts, bs=bs def embed images self, images, bs=None : return self.adapter.apply np self.backend.embed images images, bs=bs def answer with vlm query: str, results: List Dict str, Any , cfg: Config, max tiles: int = 3 - str: """ Retrieval returns pixels, so generation must accept pixels. Any VLM works; Qwen2.5-VL-3B is a reasonable Colab-sized default ~7 GB download . """ try: import torch from PIL import Image from transformers import AutoProcessor, AutoModelForImageTextToText except Exception as exc: return f" VLM unavailable: {exc} " paths = t "path" for r in results for t in r "tiles" :max tiles if not paths: return " no retrieved tiles " log.info "Loading VLM %s ...", cfg.vlm model id proc = AutoProcessor.from pretrained cfg.vlm model id model = AutoModelForImageTextToText.from pretrained cfg.vlm model id, torch dtype=torch.float16 if torch.cuda.is available else torch.float32, device map="auto" images = Image.open p .convert "RGB" for p in paths content = {"type": "image"} for in images + {"type": "text", "text": f"These are screenshots retrieved for the question. Answer using only what " f"is visible, and say so if the answer is not shown.\n\nQuestion: {query}"} prompt = proc.apply chat template {"role": "user", "content": content} , add generation prompt=True, tokenize=False inputs = proc text= prompt , images=images, return tensors="pt" .to model.device with torch.no grad : out = model.generate inputs, max new tokens=256, do sample=False text = proc.batch decode out :, inputs "input ids" .shape 1 : , skip special tokens=True 0 return text.strip def show results query: str, results: List Dict str, Any , max tiles: int = 3 - None: try: import matplotlib.pyplot as plt from PIL import Image except Exception: return tiles = r, t for r in results for t in r "tiles" :1 :max tiles if not tiles: return fig, axes = plt.subplots 1, len tiles , figsize= 5 len tiles , 6 axes = axes if len tiles == 1 else list axes for ax, r, t in zip axes, tiles : ax.imshow Image.open t "path" ax.set title f"{r 'title' :34 }\nrrf={r 'score' :.4f} cos={t 'cosine' :.3f}", fontsize=9 ax.axis "off" fig.suptitle f"Q: {query}", fontsize=12 plt.tight layout plt.show We evaluate retrieval quality using Recall@1, Recall@3, Recall@5, and mean reciprocal rank across a small benchmark. We mine pseudo-query and tile pairs from OCR content, train a residual contrastive adapter, and apply the learned transformation to both query and image embeddings. We also support grounded answer generation with a vision-language model and visualize the highest-ranked screenshot tiles with their retrieval scores. php def main cfg: Config = CFG - Dict str, Any : Path cfg.work dir .mkdir parents=True, exist ok=True ensure deps cfg import numpy as np np.random.seed cfg.seed banner = """ ██████╗ ██╗██╗ ██╗███████╗██╗ ██████╗ █████╗ ██████╗ ██╔══██╗██║╚██╗██╔╝██╔════╝██║ ██╔══██╗██╔══██╗██╔════╝ ██████╔╝██║ ╚███╔╝ █████╗ ██║ ██████╔╝███████║██║ ███╗ ██╔═══╝ ██║ ██╔██╗ ██╔══╝ ██║ ██╔══██╗██╔══██║██║ ██║ ██║ ██║██╔╝ ██╗███████╗███████╗██║ ██║██║ ██║╚██████╔╝ ╚═╝ ╚═╝╚═╝ ╚═╝╚══════╝╚══════╝╚═╝ ╚═╝╚═╝ ╚═╝ ╚═════╝ pixel-native retrieval: render - tile - embed - FAISS - serve """ print banner index, backend, tiles = build index cfg print "\n" + "=" 74 print "SEARCH DEMO — text query against a pixel index" print "=" 74 demo queries = "how do plants turn light into sugar", "what does a vector database store", "why use overlapping tiles when screenshotting a page", for q in demo queries: res = search q, index, backend, cfg pretty print q, res if cfg.show plots: show results q, res metrics before = None if cfg.enable eval: print "\n" + "=" 74 print "EVALUATION — baseline" print "=" 74 metrics before = evaluate index, backend, cfg, label="baseline" if cfg.use ocr hybrid: cfg.use ocr hybrid = False print "\n -- ablation: dense only OCR/BM25 disabled --" evaluate index, backend, cfg, label="dense-only", quiet=True cfg.use ocr hybrid = True active backend = backend if cfg.enable adapter train: print "\n" + "=" 74 print "ADAPTER TRAINING — contrastive head over frozen embeddings" print "=" 74 adapter = train adapter index, backend, tiles, cfg if adapter is not None: index.reproject adapter.apply np index.vectors active backend = AdaptedBackend backend, adapter if cfg.enable eval: print "\n -- after adapter --" after = evaluate index, active backend, cfg, label="adapted" if metrics before: d = after "mrr" - metrics before "mrr" print f" MRR delta: {d:+.3f} " f" {'improved' if d 0 else 'no gain — expected on a corpus this small'} " server = None if cfg.enable server: print "\n" + "=" 74 print "SERVE — FastAPI, upstream-compatible POST /search" print "=" 74 server = SearchServer index, active backend, cfg if server.start : import requests r = requests.post f"http://127.0.0.1:{cfg.server port}/search", json={"queries": {"text": "what is retrieval augmented generation"} , "n docs": 3}, timeout=120 payload = r.json for res in payload "results" : print f"\n POST /search query={res 'query' r}" for d in res "docs" : print f" - {d 'score' :.4f} {d 'title' :56 } <{d 'source' :48 } " print "\n Equivalent curl:" print f" curl -X POST http://127.0.0.1:{cfg.server port}/search \\" print " -H 'Content-Type: application/json' \\" print " -d '{\"queries\": {\"text\":\"capital of india\"} ,\"n docs\":3}'" if cfg.enable vlm answer: print "\n" + "=" 74 print "GENERATION — answering from retrieved pixels" print "=" 74 q = "According to the retrieved screenshots, what is photosynthesis?" res = search q, index, active backend, cfg, n docs=2 print answer with vlm q, res, cfg else: print "\n i Set CFG.enable vlm answer = True GPU to generate answers " "directly from the retrieved tiles." n docs = len {m 'doc id' for m in index.metas} print "\n" + "=" 74 print "DONE" print "=" 74 print f" tiles indexed : {len index.metas } across {n docs} documents" print f" embedding dim : {index.dim} backend: {getattr active backend, 'name', '?' }" print f" index on disk : {Path cfg.index dir .resolve }" print f" tiles on disk : {Path cfg.work dir .resolve / 'tiles'}" print """ Try next: CFG.urls - point at your own pages, then re-run main CFG.backend = "qwen3vl" - upstream's Qwen3-VL-Embedding-2B needs a big GPU CFG.device scale = 2.0 - sharper tiles, better small-text retrieval CFG.tile overlap = 256 - higher recall on prose, more vectors to store render pdf "/content/your.pdf", CFG, Path CFG.work dir /"tiles" The real deal: git clone https://github.com/StarTrail-org/PixelRAG uv sync --package pixelrag-index && pixelrag-index build """ return {"index": index, "backend": active backend, "tiles": tiles, "server": server, "search": lambda q, k=5: pretty print q, search q, index, active backend, cfg, k } if name == " main ": parser = argparse.ArgumentParser add help=False parser.add argument "--no-server", action="store true" parser.add argument "--no-train", action="store true" parser.add argument "--backend", default=None args, = parser.parse known args if args.no server: CFG.enable server = False if args.no train: CFG.enable adapter train = False if args.backend: CFG.backend = args.backend STATE = main CFG We connect every component through the main execution workflow and run the complete PixelRAG tutorial from end to end. We demonstrate search, benchmark the baseline system, compare dense-only retrieval, train the adapter, launch the API, and optionally generate answers from retrieved images. We finally display index statistics, saved output locations, extension options, and command-line controls for disabling the server, training stage, or changing the embedding backend. In conclusion, we implemented the complete PixelRAG workflow, from rendering documents into screenshot tiles to retrieving and serving relevant visual evidence through a searchable API. We combined dense vision-language embeddings, OCR-derived sparse retrieval, reciprocal rank fusion, FAISS indexing, document-level score aggregation, and contrastive adapter training within a single runnable pipeline. We also measured the system with retrieval benchmarks and inspected results visually, which allows us to compare configurations instead of relying only on qualitative outputs. By working directly with rendered pixels, we preserved document structure, tables, images, mathematical notation, code blocks, and visual layout that traditional text-only pipelines frequently discard, while creating a flexible foundation that we can extend to private documents, larger corpora, stronger multimodal embedding models, and fully grounded vision-language generation. Check out the FULL CODES here. Also, feel free to follow us on and don’t forget to join our Twitter https://x.com/intent/follow?screen name=marktechpost and Subscribe to 150k+ML SubReddit https://www.reddit.com/r/machinelearningnews/ . Wait are you on telegram? our Newsletter https://www.aidevsignals.com/ now you can join us on telegram as well. https://t.me/machinelearningresearchnews Need to partner with us for promoting your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar etc.? Connect with us https://forms.gle/wbash1wF6efRj8G58 Sana Hassan, a consulting intern at Marktechpost and dual-degree student at IIT Madras, is passionate about applying technology and AI to address real-world challenges. With a keen interest in solving practical problems, he brings a fresh perspective to the intersection of AI and real-life solutions.