Pixel-Native RAG: A Practical Guide to Visual Document Indexing
In this tutorial, we construct an entire pixel-native retrieval-augmented era pipeline from scratch and study how doc retrieval works with out counting on standard HTML parsing, textual content extraction, or mounted chunking methods. We render net pages and PDF paperwork as photographs, divide them into overlapping tiles, generate multimodal embeddings with SigLIP, CLIP, or an elective Qwen3-VL backend, and retailer the ensuing vectors in a FAISS index for environment friendly similarity search. We additionally strengthen retrieval with OCR-based BM25 scoring and reciprocal rank fusion, combination tile-level proof into document-level outcomes, and expose the system by means of a FastAPI search service. Along the best way, we consider retrieval high quality utilizing Recall@ok and imply reciprocal rank, practice a light-weight residual adapter with contrastive studying, visualize retrieved screenshots, and optionally cross the strongest proof tiles to a vision-language mannequin for grounded reply era.
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, discipline, asdict
from typing import List, Dict, Any, Optional, Tuple
@dataclass
class Config:
urls: List[str] = discipline(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] = discipline(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.primaryConfig(stage=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; by no means explode the pocket book on a single unhealthy wheel."""
cmd = [sys.executable, "-m", "pip", "install", "-q", "--disable-pip-version-check", *pkgs]
subprocess.run(cmd, examine=False, stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
def _have(mod: str) -> bool:
import importlib.util
return importlib.util.find_spec(mod) isn't None
def ensure_deps(cfg: Config) -> None:
log.information("Installing dependencies (first run solely, ~2-4 min)...")
needed = []
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):
needed.append(pkg)
if cfg.use_ocr_hybrid and never _have("pytesseract"):
needed.append("pytesseract")
if needed:
_pip(*needed)
if not _have("torch"):
log.warning("torch not discovered — putting in CPU wheel (Colab usually ships torch).")
_pip("torch", "torchvision")
if cfg.use_ocr_hybrid and shutil.which("tesseract") is None:
log.information("Installing tesseract-ocr system bundle...")
subprocess.run("apt-get -qq replace && apt-get -qq set up -y tesseract-ocr",
shell=True, examine=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.information("Downloading Playwright Chromium...")
r = subprocess.run([sys.executable, "-m", "playwright", "install", "--with-deps", "chromium"],
capture_output=True, textual content=True)
if r.returncode != 0:
r = subprocess.run([sys.executable, "-m", "playwright", "install", "chromium"],
capture_output=True, textual content=True)
if r.returncode == 0:
marker.guardian.mkdir(mother and father=True, exist_ok=True)
marker.write_text("okay")
else:
log.warning("Chromium set up failed -> falling again to the textual content renderer.npercents",
(r.stderr or "")[-600:])
log.information("Dependencies prepared.")
def run_async(coro):
"""
Run a coroutine from a Jupyter/Colab cell.
Colab already owns a operating occasion loop, which makes Playwright's *sync*
API increase. Rather than monkey-patching with nest_asyncio, we hand the
coroutine to a non-public loop on a non-public thread — probably the most strong choice.
"""
field: Dict[str, Any] = {}
def _runner():
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
strive:
field["value"] = loop.run_until_complete(coro)
besides BaseException as exc:
field["error"] = exc
lastly:
strive:
loop.run_until_complete(loop.shutdown_asyncgens())
lastly:
loop.shut()
t = threading.Thread(goal=_runner, daemon=True)
t.begin()
t.be part of()
if "error" in field:
increase field["error"]
return field.get("worth")
We outline the worldwide configuration, analysis queries, logging habits, and runtime settings for the PixelRAG pipeline. We set up the required Python and system dependencies, together with Playwright, Chromium, Tesseract, FAISS, and transformer libraries. We additionally create an asynchronous execution helper that enables browser-rendering coroutines to run reliably inside Google Colab and Jupyter environments.
@dataclass
class Tile:
tile_id: str
doc_id: str
supply: str
form: str
web 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("/").cut up("/")[-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, measurement: int = 8) -> int:
"""64-bit common hash — low cost near-duplicate detection for repeated headers."""
import numpy as np
g = img.convert("L").resize((measurement, measurement))
a = np.asarray(g, dtype="float32")
bits = (a > a.imply()).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).rely("1")
def _is_informative(img, cfg: Config) -> bool:
"""Reject clean / solid-colour tiles earlier than they ever attain 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, title: str) -> str:
out_dir.mkdir(mother and father=True, exist_ok=True)
p = out_dir / f"{title}.png"
img.convert("RGB").save(p, format="PNG", optimize=True)
return str(p)
def slice_image_to_tiles(img, cfg: Config, *, doc_id: str, supply: str, form: str,
web 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 textual content fallback."""
from PIL import Image
seen_hashes = seen_hashes if seen_hashes isn't None else []
W, H = img.measurement
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.measurement
step = max(1, cfg.tile_height - cfg.tile_overlap)
tiles: List[Tile] = []
y, seq = 0, start_seq
whereas 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{web page}__t{seq}"
tiles.append(Tile(
tile_id=tid, doc_id=doc_id, supply=supply, form=form, web page=web 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 >= doc.physique.scrollHeight || y > 40000) {
clearInterval(timer);
window.scrollTo(0, 0);
setTimeout(resolve, 250);
}
}, 40);
});
}
"""
_JS_FLATTEN = """
() => {
doc.querySelectorAll('*').forEach((el) => s.place === 'sticky') el.fashion.place = 'absolute';
);
doc.querySelectorAll('[role="dialog"], .cookie, #cookie-banner, .cc-banner')
.forEach((el) => el.take away());
}
"""
_CSS_CLEANUP = """
* { animation: none !necessary; transition: none !necessary;
scroll-behavior: auto !necessary; }
html { -webkit-font-smoothing: antialiased; }
video, iframe[src*="youtube"] { visibility: hidden !necessary; }
"""
_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, "peak": 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)
web page = await ctx.new_page()
strive:
await web page.goto(url, wait_until="domcontentloaded", timeout=cfg.nav_timeout_ms)
strive:
await web page.wait_for_load_state("networkidle", timeout=12000)
besides Exception:
cross
await web page.consider(_JS_AUTOSCROLL)
await web page.add_style_tag(content material=_CSS_CLEANUP)
await web page.consider(_JS_FLATTEN)
title = (await web page.title()) or doc_id
peak = await web page.consider(
"() => Math.max(doc.physique.scrollHeight, "
"doc.documentElement.scrollHeight)")
peak = int(min(peak, cfg.max_page_height))
step = max(1, cfg.tile_height - cfg.tile_overlap)
seen: List[int] = []
y, seq = 0, 0
whereas y < peak and seq < cfg.max_tiles_per_doc:
h = min(cfg.tile_height, peak - y)
if h < cfg.min_tile_height and seq > 0:
break
buf = await web page.screenshot(
full_page=True, sort="png",
clip={"x": 0, "y": y, "width": cfg.tile_width, "peak": h})
img = Image.open(io.BytesIO(buf)).convert("RGB")
if img.measurement[0] != cfg.tile_width:
img = img.resize((cfg.tile_width,
max(1, int(img.measurement[1] * cfg.tile_width / img.measurement[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, supply=url, form="net",
web page=0, seq=seq, y0=y, y1=y + h, title=title,
path=_save_tile(img, out_dir, tid)))
seq += 1
y += step
log.information(" rendered %-34s -> %second tiles (web page %dpx)", doc_id, seq, peak)
besides Exception as exc:
log.warning(" FAILED %s (%s)", url, sort(exc).__name__)
lastly:
await web page.shut()
await ctx.shut()
await browser.shut()
return all_tiles
def render_urls(urls: List[str], cfg: Config, out_dir: Path) -> List[Tile]:
"""Screenshot each URL into tiles; degrade to the textual content renderer on failure."""
strive:
tiles = run_async(_render_urls_async(urls, cfg, out_dir))
if tiles:
return tiles
log.warning("Browser produced no tiles — utilizing text-render fallback.")
besides Exception as exc:
log.warning("Playwright unavailable (%s: %s) — utilizing text-render fallback.",
sort(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|fashion|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)
textual content = re.sub(r"(?s)<[^>]+>", " ", html)
for a, b in [(" ", " "), ("&", "&"), ("<", "<"), (">", ">"), (""", '"')]:
textual content = textual content.exchange(a, b)
textual content = re.sub(r"[d+]", "", textual content)
textual content = re.sub(r"[ t]+", " ", textual content)
return re.sub(r"n{2,}", "n", textual content).strip()
def _mono_font(measurement: 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, measurement)
strive:
import matplotlib.font_manager as fm
return ImageFont.truetype(fm.findfont("DejaVu Sans"), measurement)
besides Exception:
return ImageFont.load_default()
def text_to_image(textual content: str, cfg: Config, title: str = "") -> Any:
"""Render plain textual content 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)
strains: List[str] = []
for para in textual content.cut up("n"):
para = para.strip()
if not para:
proceed
whereas len(para) > wrap:
lower = para.rfind(" ", 0, wrap)
lower = lower if lower > 0 else wrap
strains.append(para[:cut])
para = para[cut:].lstrip()
strains.append(para)
strains = strains[:900]
peak = pad * 2 + 60 + lh * len(strains)
img = Image.new("RGB", (cfg.tile_width, max(cfg.tile_height, peak)), "white")
d = ImageDraw.Draw(img)
d.textual content((pad, pad), title[:60], font=tfont, fill=(15, 15, 15))
for i, ln in enumerate(strains):
d.textual content((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)
strive:
r = requests.get(url, timeout=30, headers={"User-Agent": _UA})
r.raise_for_status()
physique = _strip_html(r.textual content)
m = re.search(r"(?is)<title>(.*?)</title>", r.textual content)
title = m.group(1).strip() if m else doc_id
besides Exception as exc:
log.warning(" fetch failed for %s (%s)", url, sort(exc).__name__)
return []
img = text_to_image(physique, cfg, title=title)
log.information(" text-rendered %-30s -> canvas %dpx", doc_id, img.measurement[1])
return slice_image_to_tiles(img, cfg, doc_id=doc_id, supply=url, form="textual content",
web 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 vary(n_pages):
pix = doc[pno].get_pixmap(dpi=dpi)
img = Image.frombytes("RGB", (pix.width, pix.peak), pix.samples)
tiles += slice_image_to_tiles(img, cfg, doc_id=doc_id, supply=pdf_path,
form="pdf", web page=pno, out_dir=out_dir,
title=title)
log.information(" rendered %-34s -> %second tiles (%d pages)", doc_id, len(tiles), n_pages)
return tiles
def make_synthetic_pdf(path: Path) -> str:
"""A tiny PDF so the tutorial at all times workout routines the PDF path, offline or not."""
import fitz
physique = [
("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()
web page = doc.new_page()
y = 72
for line, measurement in physique:
web page.insert_text((72, y), line, fontsize=measurement, fontname="helv")
y += measurement + 8
doc.save(str(path))
doc.shut()
return str(path)
We create the document-rendering layer that converts net pages, textual content content material, and PDF recordsdata into structured picture tiles. We seize net pages with Playwright, clear distracting web page parts, apply overlapping vertical slicing, and take away clean or duplicate tiles. We additionally present text-rendering and synthetic-PDF fallbacks so the pipeline continues to function when browser rendering or exterior content material is unavailable.
def ocr_tiles(tiles: List[Tile], cfg: Config) -> None:
if not cfg.use_ocr_hybrid:
return
strive:
import pytesseract
from PIL import Image
besides Exception:
log.warning("pytesseract lacking -> 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"):
strive:
uncooked = pytesseract.image_to_string(Image.open(t.path), config="--psm 6")
t.ocr_text = re.sub(r"s+", " ", uncooked).strip()[:4000]
besides Exception:
t.ocr_text = ""
log.information("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 twin encoder.
Honest caveat: these encoders have been educated on pure photographs with brief
captions (64-77 token textual content towers). They perceive a screenshot's *gist*
— structure, matter, figures — not its fantastic print. That is precisely why upstream
PixelRAG makes use of Qwen3-VL-Embedding-2B plus a LoRA educated on screenshots.
Sections §9 (OCR hybrid) and §10 (adapter) exist to shut a part of the hole
on {hardware} that may't host a 2B VLM.
"""
def __init__(self, cfg: Config):
import torch
from transformers import AutoModel, AutoProcessor
self.cfg = cfg
self.system = torch_device()
self.dtype = torch.float16 if self.system == "cuda" else torch.float32
self.model_id = cfg.model_id if cfg.backend != "clip" else "openai/clip-vit-base-patch32"
log.information("Loading embedding mannequin %s on %s (%s)", self.model_id, self.system,
str(self.dtype).exchange("torch.", ""))
self.processor = AutoProcessor.from_pretrained(self.model_id)
self.mannequin = AutoModel.from_pretrained(self.model_id, torch_dtype=self.dtype)
self.mannequin.to(self.system).eval()
self.is_siglip = "siglip" in self.model_id.decrease()
self.dim = int(getattr(self.mannequin.config, "projection_dim", 0) or
getattr(self.mannequin.config.text_config, "hidden_size", 512))
self.title = 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, photographs: 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(vary(0, len(photographs), bs), desc="embed:picture", unit="batch"):
batch = photographs[i:i + bs]
inputs = self.processor(photographs=batch, return_tensors="pt")
inputs = {ok: v.to(self.system, self.dtype if v.is_floating_point() else v.dtype)
for ok, v in inputs.objects()}
with torch.no_grad():
feats = self.mannequin.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 vary(0, len(texts), bs):
batch = [t if t.strip() else " " for t in texts[i:i + bs]]
kw = dict(textual content=batch, return_tensors="pt", truncation=True)
kw.replace(padding="max_length", max_length=64) if self.is_siglip else kw.replace(padding=True)
inputs = self.processor(**kw)
inputs = {ok: v.to(self.system) for ok, v in inputs.objects()}
with torch.no_grad():
feats = self.mannequin.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 current transformers (>= 4.57) and ~8 GB of VRAM in fp16. It embeds
textual content and pictures into one house by mean-pooling the final hidden state of a
VLM immediate, which is why it handles dense screenshot textual content much better than
a CLIP-style tower.
"""
def __init__(self, cfg: Config):
import torch
from transformers import AutoModel, AutoProcessor
self.cfg = cfg
self.system = torch_device()
self.dtype = torch.float16 if self.system == "cuda" else torch.float32
mid = cfg.qwen_model_id
log.information("Loading %s (this can be a giant obtain)...", mid)
self.processor = AutoProcessor.from_pretrained(mid, trust_remote_code=True)
self.mannequin = AutoModel.from_pretrained(mid, torch_dtype=self.dtype,
trust_remote_code=True).to(self.system).eval()
self.dim = int(self.mannequin.config.hidden_size)
self.title = f"qwen3vl:{mid}"
def _pool(self, hidden, masks):
import torch
m = masks.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 = {ok: (v.to(self.system) if hasattr(v, "to") else v) for ok, v in inputs.objects()}
with torch.no_grad():
out = self.mannequin(**inputs, output_hidden_states=True)
hidden = out.hidden_states[-1] if getattr(out, "hidden_states", None) isn't None
else out.last_hidden_state
vec = self._pool(hidden, inputs["attention_mask"]).float().cpu().numpy()
return DualEncoderBackend._l2(vec)
def embed_images(self, photographs: 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(vary(0, len(photographs), bs), desc="embed:picture", unit="batch"):
batch = photographs[i:i + bs]
immediate = ["Describe this document screenshot for retrieval."] * len(batch)
chunks.append(self._encode(textual content=immediate, photographs=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 vary(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":
strive:
return Qwen3VLEmbeddingBackend(cfg)
besides Exception as exc:
log.warning("Qwen3-VL backend failed (%s: %s) -> falling again to SigLIP.",
sort(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 vary(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.shut()
return np.concatenate(vecs, 0) if vecs else np.zeros((0, backend.dim), "float32")
We extract OCR textual content from every rendered tile to help sparse retrieval and automated training-pair era. We implement SigLIP, CLIP, and Qwen3-VL embedding backends that place textual content queries and doc screenshots inside a shared vector house. We then course of the tile photographs in batches and generate normalized embeddings which are prepared for similarity indexing.
class PixelIndex:
"""
Inner-product FAISS index over L2-normalised vectors (== cosine similarity).
Flat beneath `ivf_threshold` vectors (actual, no coaching); IVF above it
(sub-linear, wants coaching + nprobe tuning). Raw vectors are additionally saved in
reminiscence so §10 can re-project them after adapter coaching with out 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 construct(self, vectors, tiles: List[Tile]) -> "PixelIndex":
import faiss, numpy as np
vectors = np.ascontiguousarray(vectors.astype("float32"))
n = vectors.form[0]
if n == 0:
increase 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.practice(vectors)
base.nprobe = self.cfg.ivf_nprobe
log.information("FAISS IndexIVFFlat n=%d nlist=%d nprobe=%d", n, nlist, base.nprobe)
else:
base = faiss.IndexFlatIP(self.dim)
log.information("FAISS IndexFlatIP n=%d dim=%d (actual 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
strive:
from rank_bm25 import BM25Okapi
besides Exception:
return
self._bm25_corpus = [re.findall(r"[a-z0-9]+", (m.get("ocr_text", "") + " " +
m.get("title", "")).decrease())
for m in self.metas]
if any(self._bm25_corpus):
self._bm25 = BM25Okapi([c or ["_"] for c in self._bm25_corpus])
log.information("BM25 fitted over OCR sidecar (%d docs)", len(self._bm25_corpus))
def search_dense(self, qvecs, ok: int):
import numpy as np
scores, ids = self.index.search(np.ascontiguousarray(qvecs.astype("float32")), ok)
return scores, ids
def search_sparse(self, question: str, ok: int) -> List[Tuple[int, float]]:
if self._bm25 is None:
return []
import numpy as np
toks = re.findall(r"[a-z0-9]+", question.decrease())
if not toks:
return []
s = np.asarray(self._bm25.get_scores(toks))
prime = np.argsort(-s)[:k]
return [(int(i), float(s[i])) for i in prime if s[i] > 0]
def save(self, out_dir: str) -> None:
import faiss, numpy as np
p = Path(out_dir)
p.mkdir(mother and father=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".be part of(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.information("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.masses((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 coaching in §10)."""
tiles = [Tile(**m) for m in self.metas]
self.construct(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(mother and father=True, exist_ok=True)
log.information("=" * 74)
log.information("STAGE 1/4 RENDER (paperwork -> picture tiles)")
log.information("=" * 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:
increase RuntimeError("Rendering produced zero tiles. Check community entry.")
log.information("Total tiles: %d throughout %d paperwork",
len(tiles), len({t.doc_id for t in tiles}))
log.information("=" * 74)
log.information("STAGE 2/4 OCR SIDECAR (for hybrid retrieval + pair mining)")
log.information("=" * 74)
ocr_tiles(tiles, cfg)
log.information("=" * 74)
log.information("STAGE 3/4 EMBED (tiles -> vectors)")
log.information("=" * 74)
backend = build_backend(cfg)
t0 = time.time()
vecs = embed_tiles(tiles, backend, cfg)
log.information("Embedded %d tiles -> %s in %.1fs (%.2f tiles/s)",
vecs.form[0], vecs.form, time.time() - t0,
vecs.form[0] / max(1e-6, time.time() - t0))
log.information("=" * 74)
log.information("STAGE 4/4 INDEX (vectors -> FAISS)")
log.information("=" * 74)
index = PixelIndex(vecs.form[1], cfg).construct(vecs, tiles)
index.save(cfg.index_dir)
return index, backend, tiles
We assemble the PixelIndex class and retailer the normalized tile embeddings inside a FAISS inner-product index. We help actual flat seek for smaller datasets, IVF-based seek for bigger collections, BM25 indexing over OCR textual content, and chronic storage of vectors and metadata. We additionally orchestrate the whole indexing pipeline by rendering paperwork, operating OCR, producing embeddings, constructing the index, and saving all outputs to disk.
def search(question: 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
ok = min(cfg.top_k_tiles, len(index.metas))
qv = backend.embed_texts([query])
dscores, dids = index.search_dense(qv, ok)
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(question, ok) 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.objects(), 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"],
"supply": m["source"], "form": m["kind"], "rating": 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"],
"web page": m["page"], "y0": m["y0"], "y1": m["y1"],
"rrf": spherical(fscore, 6),
"cosine": spherical(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(question: str, outcomes: List[Dict[str, Any]]) -> None:
print(f"n 33[1mQ: {query} 33[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']})")
prime = r["tiles"][0]
print(f" tile {prime['tile_id']} y={prime['y0']}-{prime['y1']}")
if prime["snippet"]:
print(f" 33[2m{top['snippet'][:150]}... 33[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)", model="1.0")
@app.get("/well being")
def well being():
return {"standing": "okay", "tiles": len(index.metas),
"docs": len({m["doc_id"] for m in index.metas}),
"backend": getattr(backend, "title", "unknown")}
@app.submit("/search")
def do_search(req: SearchRequest):
return {"outcomes": [
{"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 begin(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(goal=self.server.run, daemon=True)
self.thread.begin()
for _ in vary(40):
time.sleep(0.25)
strive:
if requests.get(f"http://127.0.0.1:{self.cfg.server_port}/well being",
timeout=2).okay:
log.information("Search API reside on http://127.0.0.1:%d", self.cfg.server_port)
return True
besides Exception:
proceed
log.warning("Server didn't come up in time.")
return False
def cease(self) -> None:
if self.server:
self.server.should_exit = True
if self.thread:
self.thread.be part of(timeout=5)
We implement hybrid retrieval by combining dense vector rankings and OCR-based BM25 rankings by means of reciprocal rank fusion. We combination matching tiles into document-level outcomes whereas retaining the strongest proof tiles, similarity scores, and OCR snippets for inspection. We additionally expose the retrieval system by means of a FastAPI server with well being and search endpoints that run on a background Uvicorn thread.
def consider(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, need in queries:
docs = search(q, index, backend, cfg, n_docs=10)
hit = subsequent((i for i, d in enumerate(docs) if need.decrease() in d["doc_id"].decrease()), None)
ranks.append(hit)
if not quiet:
obtained = docs[0]["doc_id"] if docs else "-"
mark = "OK " if hit == 0 else (f"@{hit + 1}" if hit isn't None else "MISS")
print(f" [{mark:>4}] {q[:56]:<58} -> {obtained[:32]}")
n = len(ranks)
m = {
"recall@1": sum(r == 0 for r in ranks) / n,
"recall@3": sum(r isn't None and r < 3 for r in ranks) / n,
"recall@5": sum(r isn't None and r < 5 for r in ranks) / n,
"mrr": sum(1.0 / (r + 1) for r in ranks if r isn't None) / n,
}
print(f" 33[1m{label} 33[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 personal OCR textual content / title."""
import random
rng = random.Random(0)
pairs: List[Tuple[str, int]] = []
for idx, t in enumerate(tiles):
textual content = (t.ocr_text or "").strip()
cands: List[str] = []
if len(textual content) > 80:
phrases = textual content.cut up()
for _ in vary(max_per_tile):
if len(phrases) <= 14:
break
s = rng.randint(0, len(phrases) - 14)
span = " ".be part of(phrases[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 utilized to each question and tile vectors."""
def __init__(self, dim: int, hidden: int = 512, system: str = "cpu"):
import torch
import torch.nn as nn
self.system = system
self.web = nn.Sequential(
nn.Linear(dim, hidden), nn.GELU(), nn.Linear(hidden, dim)
).to(system)
for p in self.web[-1].parameters():
torch.nn.init.zeros_(p)
self.logit_scale = torch.nn.Parameter(torch.tensor(2.996, system=system))
self.dim = dim
def forward_t(self, x):
import torch
y = x + self.web(x)
return torch.nn.purposeful.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.system)
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 coaching "
"(allow OCR or add paperwork).", len(pairs))
return None
log.information("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.information("Pre-embedding pseudo-queries (frozen encoder, completed as soon as)...")
Q = torch.from_numpy(backend.embed_texts(q_texts))
V = torch.from_numpy(index.vectors)
system = "cuda" if torch.cuda.is_available() else "cpu"
advert = ResidualAdapter(index.dim, system=system)
Q, V = Q.to(system), V.to(system)
choose = torch.optim.AdamW(checklist(advert.web.parameters()) + [ad.logit_scale], lr=lr, weight_decay=1e-2)
n = len(pairs)
doc_ids = torch.from_numpy(t_idx).to(system)
for ep in vary(epochs):
perm = torch.randperm(n, system=system)
whole, steps = 0.0, 0
for i in vary(0, n, batch):
sel = perm[i:i + batch]
if sel.numel() < 4:
proceed
qb = advert.forward_t(Q[sel])
docs = doc_ids[sel]
vb = advert.forward_t(V[docs])
logits = advert.logit_scale.exp().clamp(max=100) * qb @ vb.T
similar = docs[:, None] == docs[None, :]
eye = torch.eye(len(sel), dtype=torch.bool, system=system)
logits = logits.masked_fill(similar & ~eye, float("-inf"))
labels = torch.arange(len(sel), system=system)
loss = 0.5 * (torch.nn.purposeful.cross_entropy(logits, labels) +
torch.nn.purposeful.cross_entropy(logits.T, labels))
choose.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(advert.web.parameters(), 1.0)
choose.step()
whole += loss.detach().merchandise()
steps += 1
if ep % 3 == 0 or ep == epochs - 1:
log.information(" epoch %second/%d InfoNCE loss %.4f", ep + 1, epochs, whole / max(steps, 1))
return advert
class AdaptedBackend:
"""Wraps a frozen backend so queries cross by means of the educated adapter."""
def __init__(self, backend, adapter: ResidualAdapter):
self.backend, self.adapter = backend, adapter
self.dim = backend.dim
self.title = f"{getattr(backend, 'title', '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, photographs, bs=None):
return self.adapter.apply_np(self.backend.embed_images(photographs, bs=bs))
def answer_with_vlm(question: str, outcomes: List[Dict[str, Any]], cfg: Config,
max_tiles: int = 3) -> str:
"""
Retrieval returns pixels, so era should settle for pixels. Any VLM works;
Qwen2.5-VL-3B is an inexpensive Colab-sized default (~7 GB obtain).
"""
strive:
import torch
from PIL import Image
from transformers import AutoProcessor, AutoModelForImageTextToText
besides Exception as exc:
return f"[VLM unavailable: {exc}]"
paths = [t["path"] for r in outcomes for t in r["tiles"]][:max_tiles]
if not paths:
return "[no retrieved tiles]"
log.information("Loading VLM %s ...", cfg.vlm_model_id)
proc = AutoProcessor.from_pretrained(cfg.vlm_model_id)
mannequin = AutoModelForImageTextToText.from_pretrained(
cfg.vlm_model_id,
torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
device_map="auto")
photographs = [Image.open(p).convert("RGB") for p in paths]
content material = [{"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.nnQuestion: {query}"}]
immediate = proc.apply_chat_template([{"role": "user", "content": content}],
add_generation_prompt=True, tokenize=False)
inputs = proc(textual content=[prompt], photographs=photographs, return_tensors="pt").to(mannequin.system)
with torch.no_grad():
out = mannequin.generate(**inputs, max_new_tokens=256, do_sample=False)
textual content = proc.batch_decode(out[:, inputs["input_ids"].form[1]:],
skip_special_tokens=True)[0]
return textual content.strip()
def show_results(question: str, outcomes: List[Dict[str, Any]], max_tiles: int = 3) -> None:
strive:
import matplotlib.pyplot as plt
from PIL import Image
besides 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 checklist(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: {question}", fontsize=12)
plt.tight_layout()
plt.present()
We consider retrieval high quality utilizing Recall@1, Recall@3, Recall@5, and imply reciprocal rank throughout a small benchmark. We mine pseudo-query and tile pairs from OCR content material, practice a residual contrastive adapter, and apply the realized transformation to each question and picture embeddings. We additionally help grounded reply era with a vision-language mannequin and visualize the highest-ranked screenshot tiles with their retrieval scores.
def fundamental(cfg: Config = CFG) -> Dict[str, Any]:
Path(cfg.work_dir).mkdir(mother and father=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 — textual content question in opposition to 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 = consider(index, backend, cfg, label="baseline")
if cfg.use_ocr_hybrid:
cfg.use_ocr_hybrid = False
print("n -- ablation: dense solely (OCR/BM25 disabled) --")
consider(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 isn't None:
index.reproject(adapter.apply_np(index.vectors))
active_backend = AdaptedBackend(backend, adapter)
if cfg.enable_eval:
print("n -- after adapter --")
after = consider(index, active_backend, cfg, label="tailored")
if metrics_before:
d = after["mrr"] - metrics_before["mrr"]
print(f" MRR delta: {d:+.3f} "
f"({'improved' if d > 0 else 'no acquire — anticipated 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.begin():
import requests
r = requests.submit(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 question={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: software/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's 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 solutions "
"instantly 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 listed : {len(index.metas)} throughout {n_docs} paperwork")
print(f" embedding dim : {index.dim} backend: {getattr(active_backend, 'title', '?')}")
print(f" index on disk : {Path(cfg.index_dir).resolve()}")
print(f" tiles on disk : {Path(cfg.work_dir).resolve() / 'tiles'}")
print("""
Try subsequent:
* CFG.urls -> level at your individual pages, then re-run fundamental()
* CFG.backend = "qwen3vl" -> upstream's Qwen3-VL-Embedding-2B (wants an enormous GPU)
* CFG.device_scale = 2.0 -> sharper tiles, higher small-text retrieval
* CFG.tile_overlap = 256 -> increased recall on prose, extra vectors to retailer
* render_pdf("/content material/your.pdf", CFG, Path(CFG.work_dir)/"tiles")
* The actual deal: git clone https://github.com/StarTrail-org/PixelRAG
uv sync --package pixelrag-index && pixelrag-index construct
""")
return {"index": index, "backend": active_backend, "tiles": tiles, "server": server,
"search": lambda q, ok=5: pretty_print(q, search(q, index, active_backend, cfg, ok))}
if __name__ == "__main__":
parser = argparse.ArgumentParser(add_help=False)
parser.add_argument("--no-server", motion="store_true")
parser.add_argument("--no-train", motion="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 = fundamental(CFG)
We join each part by means of the principle execution workflow and run the whole PixelRAG tutorial from finish to finish. We show search, benchmark the baseline system, evaluate dense-only retrieval, practice the adapter, launch the API, and optionally generate solutions from retrieved photographs. We lastly show index statistics, saved output areas, extension choices, and command-line controls for disabling the server, coaching stage, or altering the embedding backend.
In conclusion, we carried out the whole PixelRAG workflow, from rendering paperwork into screenshot tiles to retrieving and serving related visible proof by means of a searchable API. We mixed dense vision-language embeddings, OCR-derived sparse retrieval, reciprocal rank fusion, FAISS indexing, document-level rating aggregation, and contrastive adapter coaching inside a single runnable pipeline. We additionally measured the system with retrieval benchmarks and inspected outcomes visually, which permits us to evaluate configurations as a substitute of relying solely on qualitative outputs. By working instantly with rendered pixels, we preserved doc construction, tables, photographs, mathematical notation, code blocks, and visible structure that conventional text-only pipelines regularly discard, whereas creating a versatile basis that we will lengthen to non-public paperwork, bigger corpora, stronger multimodal embedding fashions, and absolutely grounded vision-language era.
Check out the FULL CODES here. Also, be at liberty to observe us on Twitter and don’t neglect to be part of our 150k+ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
Need to accomplice with us for selling your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar and so forth.? Connect with us
The submit Pixel-Native RAG: A Practical Guide to Visual Document Indexing appeared first on MarkTechPost.
