像素原生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.

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.