像素级 RAG 实战：把文档渲染成图像从零构建检索管线

一、背景：为什么传统 RAG 在富版式文档上翻车

经典 RAG 管线是”文本中心”的：PyPDF 解析文字 → 按字符数切块 → 文本 embedding → 向量检索。这一套在纯文本 PDF 上工作得不错，但一碰到真实世界的文档就碎了：双栏排版的论文（解析出来左右栏文字搅在一起）、带表格的财报（行列关系全丢）、扫描版合同（根本没有文本层）、网页（正文广告导航混成一锅）、PPT（文字全在文本框里，顺序错乱）。

像素级 RAG（Pixel-native RAG）换了一条路：不解析、不切文本块，直接把文档渲染成图像，按视觉切片（tile）索引，用 SigLIP/CLIP 这类多模态 embedding 把”文字 query”和”文档截图”放进同一个向量空间做检索。表格、图表、版式、印章这些视觉信号全部保留，OCR 只做辅助的稀疏检索通道。

本教程从零构建一条完整管线：文档渲染切片 → 多模态嵌入 → FAISS 索引 → 稠密+稀疏混合检索（RRF 融合）→ FastAPI 服务 → 评测（Recall@k/MRR）→ 对比学习微调残差适配器 → VLM 接入生成答案。

二、原理：像素检索为什么 work

2.1 核心思想：版式即信息

人类读财报时，表格线、加粗、aden 表头位置本身就是语义。文本抽取把这些全扔了，像素保留了全部。SigLIP 这类图文对比预训练模型，见过海量”截图 ↔ 描述”数据，天生会把”2024年营收 5.2 亿”这样的 query 映射到包含该数字表格的截图附近。

2.2 管线全景

文档（URL / PDF / 文本）
  → Playwright 渲染成整页截图
  → 纵向重叠切片（tile，高 800~1000px，重叠 100~200px）
  → 去空白、去重
  → 双路索引：
      A. 稠密路：SigLIP/CLIP 图像 embedding → FAISS
      B. 稀疏路：OCR 文字 → BM25
  → 查询时双路召回 → RRF 融合 → tile 聚合成文档
  → （可选）Top tiles 喂给 VLM 生成 grounded 答案

2.3 关键设计决策

重叠切片：切在表格/段落中间是不可避免的，重叠 15%~20% 保证任何一行内容都完整出现在至少一个 tile 里。
tile 聚合到文档：检索命中的是 tile，但用户要的是文档。用”文档得分 = Top tiles 得分聚合（max 或加权和）”回切到文档级。
RRF（倒数排名融合）：稠密和稀疏两路的分数不可比，融合排名而非分数：score(d) = Σ 1/(k + rank)，k 常取 60。简单、鲁棒、无需调参。
残差适配器：冻结 SigLIP 主干，只学一个轻量线性残差层，用 OCR 自动构造的（query, 正 tile, 负 tile）三元组做对比学习，让 embedding 适配你的文档分布。

2.4 评测指标

Recall@k：正确答案出现在前 k 个结果里的比例，检索系统的头号指标。
MRR（平均倒数排名）：正确答案排名的倒数均值，考验”第一屏就有答案”的能力。

三、环境准备

# 系统依赖（Debian/Ubuntu）
apt install -y tesseract-ocr tesseract-ocr-chi-sim chromium-browser

# Python 依赖
pip install playwright faiss-cpu transformers torch pillow \
            pytesseract rank-bm25 fastapi uvicorn pdf2image scikit-learn

# Playwright 浏览器
playwright install chromium

需要一块能跑 SigLIP 的 GPU（CPU 也能跑，慢一个数量级）。网络要能下载 HuggingFace 模型（google/siglip-base-patch16-224）。

四、分步实战

4.1 步骤一：全局配置与异步执行辅助

# config.py
from dataclasses import dataclass, field

@dataclass
class PixelConfig:
    tile_height: int = 900       # 切片高（px）
    overlap: int = 150           # 重叠（px）
    tile_width: int = 800        # 统一缩放宽
    embed_model: str = "google/siglip-base-patch16-224"
    batch_size: int = 16
    top_k: int = 10
    rrf_k: int = 60
    index_dir: str = "./pixel_index"

# 评测 query：每个 query 标注正确答案所在的文档 id
EVAL_QUERIES = [
    {"q": "2024年公司营收是多少？", "gold_doc": "report_2024"},
    {"q": "退货政策是几天？", "gold_doc": "policy"},
    {"q": "模型训练的学习率设置？", "gold_doc": "paper"},
]

# async_helper.py —— Jupyter/Colab 里可靠运行 Playwright 协程
import asyncio

def run_coro(coro):
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(coro)
    else:  # 已在事件循环内（Jupyter），用新线程跑
        import concurrent.futures
        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex:
            return ex.submit(asyncio.run, coro).result()

4.2 步骤二：文档渲染层（网页 / 文本 / PDF → tiles）

# render.py
from playwright.async_api import async_playwright
from PIL import Image
import io

async def render_url(url: str, width: int = 800) -> Image.Image:
    async with async_playwright() as pw:
        browser = await pw.chromium.launch()
        page = await browser.new_page(viewport={"width": width, "height": 600})
        await page.goto(url, wait_until="networkidle")
        # 清理干扰元素：广告、弹窗、cookie 条
        await page.evaluate("""() => {
          document.querySelectorAll('[class*=ad],[class*=popup],[class*=cookie]')
                  .forEach(e => e.remove());
        }""")
        shot = await page.screenshot(full_page=True)
        await browser.close()
    return Image.open(io.BytesIO(shot))

def render_text(text: str, width: int = 800) -> Image.Image:
    """文本/PDF不可用时的兜底：白底黑字渲染成图，保证管线不断。”
    from PIL import ImageDraw
    img = Image.new("RGB", (width, 60 + 22 * (len(text) // 40 + 1)), "white")
    d = ImageDraw.Draw(img)
    y = 20
    for i in range(0, len(text), 40):
        d.text((20, y), text[i:i+40], fill="black")
        y += 22
    return img

def slice_tiles(img: Image.Image, tile_h: int = 900,
                overlap: int = 150, width: int = 800) -> list:
    img = img.resize((width, int(img.height * width / img.width)))
    tiles, y = [], 0
    while y < img.height:
        tile = img.crop((0, y, width, min(y + tile_h, img.height)))
        tiles.append(tile)
        if y + tile_h >= img.height:
            break
        y += tile_h - overlap
    # 去空白 tile：像素方差过小的跳过
    import numpy as np
    kept = [t for t in tiles if np.asarray(t).std() > 5]
    # 去重：相邻 tile 内容几乎一致时去重（phash 简化版用均值哈希）
    return kept

生产建议：PDF 用 pdf2image/pymupdf 按页渲染（比 Playwright 快且保真），网页才走 Playwright；渲染结果落盘缓存，改 embedding 模型时不用重渲染。

4.3 步骤三：OCR + 多模态嵌入后端

# embed.py
import torch, pytesseract
from PIL import Image
from transformers import AutoProcessor, AutoModel

class SiglipBackend:
    def __init__(self, model_id="google/siglip-base-patch16-224"):
        self.processor = AutoProcessor.from_pretrained(model_id)
        self.model = AutoModel.from_pretrained(model_id).eval()
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.model.to(self.device)

    @torch.no_grad()
    def embed_images(self, images: list[Image.Image]):
        inp = self.processor(images=images, return_tensors="pt",
                             padding=True).to(self.device)
        feats = self.model.get_image_features(**{k: v for k, v in inp.items()
                                                 if k in ("pixel_values",)})
        return (feats / feats.norm(dim=-1, keepdim=True)).cpu().numpy()

    @torch.no_grad()
    def embed_texts(self, texts: list[str]):
        inp = self.processor(text=texts, return_tensors="pt",
                             padding=True, truncation=True).to(self.device)
        feats = self.model.get_text_features(**inp)
        return (feats / feats.norm(dim=-1, keepdim=True)).cpu().numpy()

# 同一接口可再实现 ClipBackend（openai/clip-vit-base-patch32）
# 与 Qwen3VLBackend（Qwen3-VL 做 query/图像联合编码，可选）

def ocr_tile(tile: Image.Image, lang: str = "chi_sim+eng") -> str:
    try:
        return pytesseract.image_to_string(tile, lang=lang)
    except Exception:
        return ""

接口统一的价值：评测时可以一键对比 SigLIP vs CLIP vs Qwen3-VL 在你的文档上的 Recall@k，用数据选模型而不是跟风。

4.4 步骤四：PixelIndex（FAISS + BM25 + 持久化）

# index.py
import json, numpy as np, faiss
from rank_bm25 import BM25Okapi

class PixelIndex:
    def __init__(self, dim: int):
        self.dim = dim
        self.index = faiss.IndexFlatIP(dim)  # 小规模精确检索
        self.metas: list[dict] = []          # tile 元数据
        self.ocr_texts: list[str] = []
        self.bm25 = None

    def add(self, vecs: np.ndarray, metas: list[dict], ocr_texts: list[str]):
        self.index.add(vecs.astype("float32"))
        self.metas.extend(metas)
        self.ocr_texts.extend(ocr_texts)

    def build_bm25(self):
        tokenized = [t.split() for t in self.ocr_texts]
        self.bm25 = BM25Okapi(tokenized)

    def dense_search(self, qvec: np.ndarray, k: int):
        D, I = self.index.search(qvec.astype("float32"), k)
        return list(zip(I[0].tolist(), D[0].tolist()))

    def sparse_search(self, query: str, k: int):
        scores = self.bm25.get_scores(query.split())
        top = np.argsort(scores)[::-1][:k]
        return [(int(i), float(scores[i])) for i in top]

    def save(self, d: str):
        import os
        os.makedirs(d, exist_ok=True)
        faiss.write_index(self.index, f"{d}/pix.faiss")
        json.dump({"metas": self.metas, "ocr": self.ocr_texts},
                  open(f"{d}/meta.json", "w", ensure_ascii=False))

大规模（百万 tiles+）把 IndexFlatIP 换成 IndexIVFFlat（先 train 再 add），查询时设 nprobe 平衡速度与召回。

4.5 步骤五：端到端建索引（一键 groundbreaking）

# build.py —— 渲染→OCR→嵌入→索引→落盘
from config import PixelConfig
from render import render_text, slice_tiles
from embed import SiglipBackend, ocr_tile
from index import PixelIndex
import numpy as np

cfg = PixelConfig()

# 演示文档集（含表格文本，考验像素检索对版式+数字的捕捉）
DOCS = {
    "report_2024": "2024年公司营收5.2亿元，同比增长18%。\n年份 营收 利润\n2022 3.8亿 0.6亿\n2023 4.4亿 0.8亿\n2024 5.2亿 1.1亿",
    "policy": "退货政策：7天无理由退货，15天质量问题换新。运费满99包邮。",
    "paper": "模型训练：学习率3e-4，batch size 128，AdamW优化器，训练50 epoch。",
}

backend = SiglipBackend(cfg.embed_model)
pix = PixelIndex(dim=768)  # siglip-base 维度，按实际模型调整

for doc_id, text in DOCS.items():
    img = render_text(text)
    tiles = slice_tiles(img, cfg.tile_height, cfg.overlap, cfg.tile_width)
    ocrs = [ocr_tile(t) for t in tiles]
    vecs = backend.embed_images(tiles)
    metas = [{"doc_id": doc_id, "tile_no": i,
              "n_tiles": len(tiles)} for i in range(len(tiles))]
    pix.add(np.array(vecs), metas, ocrs)
    print(f"{doc_id}: {len(tiles)} tiles indexed")

pix.build_bm25()
pix.save(cfg.index_dir)
print("索引已落盘：", cfg.index_dir)

4.6 步骤六：混合检索（RRF 融合 + tile→文档聚合）

# search.py
def rrf_fuse(rank_lists: list[list[int]], k: int = 60) -> dict[int, float]:
    fused: dict[int, float] = {}
    for ranks in rank_lists:
        for r, idx in enumerate(ranks):
            fused[idx] = fused.get(idx, 0.0) + 1.0 / (k + r + 1)
    return fused

def hybrid_search(query: str, backend, pix, cfg, top_k: int = 5):
    qvec = backend.embed_texts([query])
    dense = pix.dense_search(qvec, k=top_k * 2)
    sparse = pix.sparse_search(query, k=top_k * 2)
    fused = rrf_fuse([[i for i, _ in dense], [i for i, _ in sparse]],
                     k=cfg.rrf_k)
    # tile→文档聚合：文档得分取其 Top tiles 融合分之和
    doc_scores: dict[str, float] = {}
    for idx, s in sorted(fused.items(), key=lambda x: -x[1])[:top_k * 2]:
        doc = pix.metas[idx]["doc_id"]
        doc_scores[doc] = doc_scores.get(doc, 0.0) + s
    ranked = sorted(doc_scores.items(), key=lambda x: -x[1])[:top_k]
    return [{"doc_id": d, "score": round(s, 4)} for d, s in ranked]

if __name__ == "__main__":
    from config import PixelConfig, EVAL_QUERIES
    from embed import SiglipBackend
    import json
    cfg = PixelConfig()
    backend = SiglipBackend(cfg.embed_model)
    import faiss
    from index import PixelIndex
    pix = PixelIndex(dim=768)
    pix.index = faiss.read_index(f"{cfg.index_dir}/pix.faiss")
    meta = json.load(open(f"{cfg.index_dir}/meta.json"))
    pix.metas, pix.ocr_texts = meta["metas"], meta["ocr"]
    pix.build_bm25()
    for item in EVAL_QUERIES:
        print(item["q"], "->", hybrid_search(item["q"], backend, pix, cfg))

4.7 步骤七：评测（Recall@k / MRR）

# eval.py
def evaluate(queries, search_fn, k: int = 3):
    recalls, rrs = [], []
    for item in queries:
        res = search_fn(item["q"])
        ids = [r["doc_id"] for r in res[:k]]
        hit = item["gold_doc"] in ids
        recalls.append(1.0 if hit else 0.0)
        if hit:
            rrs.append(1.0 / (ids.index(item["gold_doc"]) + 1))
        else:
            rrs.append(0.0)
    print(f"Recall@{k} = {sum(recalls)/len(recalls):.3f}")
    print(f"MRR@{k} = {sum(rrs)/len(rrs):.3f}")

评测时消融三组：纯稠密、纯稀疏、混合 RRF。像素 RAG 的典型结论：表格/数字类 query 稠密占优，精确术语 query 稀疏占优，混合最稳。

4.8 步骤八：残差适配器微调（对比学习）

# adapter.py —— 冻结主干，只学一个残差矩阵
import torch, torch.nn as nn

class ResidualAdapter(nn.Module):
    def __init__(self, dim: int):
        super().__init__()
        self.A = nn.Parameter(torch.zeros(dim, dim))
    def forward(self, x):
        return (x + x @ self.A).renorm(p=2, dim=1)

# 训练对自动构造：OCR 文本含 query 关键词的 tile 为正样本，
# 同 batch 其他 tile 为负样本（InfoNCE）
def info_nce(q, pos, negs, tau=0.07):
    import torch.nn.functional as F
    logits = torch.cat([q @ pos.T, q @ negs.T], dim=1) / tau
    return F.cross_entropy(logits, torch.zeros(len(q), dtype=torch.long))

几百个自动三元组、CPU 几分钟就能训完，Recall@k 常有 3~8 个点的提升。注意：适配器绑定你的文档分布，换文档集要重训。

4.9 步骤九：FastAPI 服务 + VLM 生成答案

# serve.py
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="PixelRAG search service")
# 启动时加载 backend / pix（略，见 search.py 主函数）

class Q(BaseModel):
    query: str
    top_k: int = 5

@app.post("/search")
def search(q: Q):
    return {"results": hybrid_search(q.query, backend, pix, cfg, q.top_k)}

# VLM 生成（可选）：把 Top-2 tile 原图 + query 喂给视觉语言模型，
# 要求它只依据图像作答并给出 tile 编号，实现 grounded 回答。

截图可视化别忘了：把检索命中的 tile 原图并排存下来人工看一眼——像素检索最大的调试优势就是”看得见”，命中错了一眼便知。

五、常见坑

切片不重叠：表格行被切成两半，两边都检不中。重叠至少 15%，表格密集文档给到 25%。
空白 tile 进索引：页边距、大片留白会污染索引且浪费显存。方差/边缘检测过滤空白tile 是必备步骤。
只用稠密不用稀疏：精确型号、电话号码、人名这类”一字之差”query，稠密 embedding 经常翻车，BM25 是救命通道。
tile 聚合策略拍脑袋：max 适合”一 tile 含答案”，sum 适合”答案分散多 tile”。先看 bad case 再选，别直接抄。
embedding 不归一化：FAISS 内积索引要求向量归一化后才等价于余弦相似度，漏掉这一步检索质量雪崩。
渲染与线上不一致：训练时 800px 宽、线上 1200px 宽，版式一变 embedding 分布漂移。渲染参数要版本化并与评测对齐。
OCR 语言包缺失：中文文档没装 tesseract-ocr-chi-sim，稀疏路全空还查不出原因——先对 OCR 做抽查，稀疏召回为 0 先查 OCR。
适配器过拟合小样本：自动三元组有噪声（OCR 含关键词≠语义相关），训练集太小会过拟合。用评测集早停，涨不动就停。

六、总结

像素级 RAG 用”渲染+视觉切片+多模态嵌入”绕开了文本抽取这个最大丢信息环节，是富版式文档检索的务实路线。
双路召回（SigLIP 稠密 + OCR/BM25 稀疏）+ RRF 融合 + tile→文档聚合，是经过验证的最小可用架构。
Recall@k/MRR 评测、残差适配器微调、tile 可视化调试，构成迭代三件套。
Top tiles 直喂 VLM 即得 grounded 问答，引用精确到截图编号。
从今天起：拿你最头疼的那份扫描版 PDF 跑一遍本教程，和传统文本 RAG 对比 Recall@3，数据会告诉你值不值得切像素路线。

参考资料：PixelRAG GitHub 项目与 SigLIP/CLIP 官方文档。点击阅读原文

# AI教程 # 教程

文章版权归作者所有，未经允许请勿转载。

暂无评论

暂无评论...