PixelRAG：将网页和PDF作为图像进行原生视觉索引的实战指南

人工智能

Model Context Protocol

在本教程中，我们从零开始构建一个完整的像素原生检索增强生成（RAG）流水线，并探讨如何在不依赖传统 HTML 解析、文本提取或固定分块策略的情况下实现文档检索。我们将网页和 PDF 文档渲染为图像，将其分割为重叠的图块（tile），使用 SigLIP、CLIP 或可选的 Qwen3-VL 作为后端生成多模态嵌入向量，然后将生成的向量存储在 FAISS 索引中以进行高效的相似性搜索。我们还通过基于 OCR 的 BM25 评分和互惠排序融合（reciprocal rank fusion）来增强检索效果，将图块级证据聚合为文档级结果，并通过 FastAPI 搜索服务对外暴露该系统。在此过程中，我们使用 Recall@k 和平均互惠秩（mean reciprocal rank）来评估检索质量，使用对比学习训练轻量级残差适配器，可视化检索到的截图，并可选择将最强证据图块传递给视觉语言模型以进行有依据的答案生成。

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:
       

我们定义了 PixelRAG 流水线的全局配置、评估查询、日志行为和运行时设置。我们安装了所需的 Python 和系统依赖，包括 Playwright、Chromium、Tesseract、FAISS 和 Transformer 库。我们还创建了一个异步执行辅助函数，允许浏览器渲染协程在 Google Colab 和 Jupyter 环境中可靠运行。

@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_tile

我们创建文档渲染层，将网页、文本内容和 PDF 文件转换为结构化的图像瓦片。我们使用 Playwright 捕获网页，清理干扰页面元素，应用重叠的垂直切片，并移除空白或重复的瓦片。我们还提供文本渲染和合成 PDF 回退方案，使流水线在浏览器渲染或外部内容不可用时仍能继续运行。

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.a

我们从每个渲染瓦片中提取 OCR 文本，以支持稀疏检索和自动训练对生成。我们实现了 SigLIP、CLIP 和 Qwen3-VL 嵌入后端，将文本查询和文档截图映射到共享的向量空间中。然后我们批量处理瓦片图像，生成可立即用于相似度索引的归一化嵌入。

我们构建了 PixelIndex 类，将归一化的瓦片嵌入存储在 FAISS 内积索引中。支持小数据集的精确 flat 搜索、大规模集合的 IVF 搜索、OCR 文本的 BM25 索引，以及向量和元数据的持久化存储。我们还通过渲染文档、执行 OCR、生成嵌入、构建索引并将所有输出保存到磁盘来编排完整的索引管道。

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)
                   re

我们通过倒数排名融合（reciprocal rank fusion）将密集向量排序与基于 OCR 的 BM25 排序相结合，实现混合检索。我们在文档级别聚合匹配的瓦片，同时保留最相关的证据瓦片、相似度分数和 OCR 片段供检查。我们还通过 FastAPI 服务器暴露检索系统，其 health 和 search 端点运行在后台的 Uvicorn 线程上。

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,
   }
   return m