ドキュメントを画像化して検索する「Pixel-Native RAG」の実践ガイド

In this tutorial, we build a complete pixel-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.

Copy CodeCopiedUse a different Browser

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.

Copy CodeCopiedUse a different Browser

@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 = ""

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,

"""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),

]