The tutorial builds a complete pixel-native retrieval-augmented generation pipeline from scratch. The process renders web pages and PDF documents as images, divides them into overlapping tiles, and stores the resulting vectors in a FAISS index for efficient similarity search. The system strengthens retrieval with OCR-based BM25 scoring and reciprocal rank fusion, aggregates tile-level evidence into document-level results, and exposes the system through a FastAPI search service. Evaluation uses Recall@k and mean reciprocal rank, while a lightweight residual adapter trains with contrastive learning. The strongest evidence tiles optionally pass to a vision-language model for grounded answer generation.
Setup and dependencies
The code defines global configuration, evaluation queries, logging behaviour, and runtime settings. It installs required Python and system dependencies, including Playwright, Chromium, Tesseract, FAISS, and transformer libraries. An asynchronous execution helper allows browser-rendering coroutines to run reliably inside Google Colab and Jupyter environments.
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")
The script sets up the pipeline parameters and installs the necessary libraries. It checks for existing modules and installs missing ones like Pillow, NumPy, and FAISS. If the hybrid OCR mode is enabled, it ensures Pytesseract is present. The code also handles the installation of Tesseract-ocr as a system package if the Python package is missing. A marker file tracks whether Playwright Chromium has been downloaded successfully. If that step fails, the system falls back to a text renderer.
A helper function runs coroutines from Jupyter or Colab cells. Since Colab already owns a running event loop, Playwright’s synchronous API raises errors. The solution hands the coroutine to a private loop on a private thread. This approach avoids monkey-patching with nest_asyncio.
Tile processing and deduplication
The code defines a Tile dataclass to store metadata for each image segment. It includes fields for tile ID, document ID, source, kind, page number, sequence, coordinates, file path, OCR text, and title. A function generates a document ID from the source URL by stripping extensions and replacing special characters. It limits the ID to 80 characters or uses an MD5 hash if the name is empty.
A 64-bit average hash function detects near-duplicates for repeated headers. It converts the image to grayscale, resizes it to a small square, and flattens the array into bits. The function compares two hashes using a Hamming distance calculation
![Per-pixel bounding-box regression + DBSCAN for handwritten word detection – visual walkthrough of WordDetectorNet [P]](https://ai-maestro.online/wp-content/uploads/2026/05/per-pixel-bounding-box-regression-dbscan-for-handwritten-wor-768x768.jpg){kind=tracking}
