AI教程 9 分钟
多模态 RAG 实战：图文混合检索的完整 pipeline 搭建与性能调优
从 CLIP embedding 到图文对齐检索，从 OCR 预处理到融合评分，完整搭建一个支持图片+文本混合查询的 RAG 系统，覆盖 3 种架构方案的召回率/延迟/成本对比。
AinoCode 编辑部
多模态 RAG 实战：图文混合检索的完整 pipeline 搭建与性能调优
你遇到过这种场景吗：
用户上传了一张产品截图，问”这个报错怎么解决”。你的 RAG 系统只存了文本，搜不到任何东西——因为知识库里的解决方案写在另一篇带截图的文档里，而那段文字描述和用户的截图措辞完全不同。
或者：用户上传了一份合同扫描件，问”违约金条款在第几条”。纯文本 RAG 只能匹配到 OCR 提取的文字，但丢失了表格结构、印章位置、手写批注这些只有图片才能传达的信息。
这就是纯文本 RAG 的天花板。多模态 RAG 不是”锦上添花”，而是处理图文混合知识库的”必选项”。
这篇文章完整搭建一套多模态 RAG pipeline，覆盖三种架构方案，用同一套 5000 页图文混合文档库做实测，给出 Recall@10 / 延迟 / 成本的完整对比。
一、三种多模态 RAG 架构
架构 A：文本化（Image-to-Text）
把图片转成文字描述，然后用纯文本 RAG 的流程走到底。
图片 → VLM 描述生成 → 文本 → 文本 embedding → 向量检索
优点：实现最简单，复用已有的文本 RAG 基础设施。 缺点：描述生成的信息损失不可控。VLM 的 caption 会丢失颜色、布局、精确数值等细节。
架构 B：双塔独立检索（Dual-Encoder）
图片和文本各自独立 embedding，分别检索，最后融合排序。
查询文本 → 文本 embedding → 文本向量检索 ──┐
├──→ 融合排序 → 最终结果
图片 → CLIP embedding → 图片向量检索 ──┘
优点：保留图片的原始视觉信息，不会因 caption 化而丢失。 缺点：两套索引，维护成本高。融合排序策略需要精细调优。
架构 C：统一多模态嵌入（Unified Multimodal Embedding）
用同一个 embedding 模型同时处理文本和图片，映射到同一个向量空间。
查询文本/图片 → 统一 embedding 模型 → 单一向量空间检索
优点：一套索引，天然支持跨模态检索（文搜图、图搜图、图文混搜）。 缺点：模型选择有限，目前 CLIP 系列在文本检索精度上不如专用文本模型。
二、测试数据集
我构建了一个包含 5000 页的企业文档库：
| 文档类型 | 页数 | 特点 |
|---|---|---|
| 技术手册（PDF） | 2000 | 代码截图 + 架构图 + 文字说明混合 |
| 产品规格书 | 1000 | 大量参数表格、对比图、尺寸标注 |
| 故障排查指南 | 800 | 报错截图 + 文字步骤，截图中的错误码是关键 |
| 合同/法律文档扫描件 | 700 | 扫描件、手写批注、表格、印章 |
| 会议纪要（含白板照片） | 500 | 白板照片中的架构图是核心信息 |
标注了 200 条查询样本，覆盖五种查询类型：
| 查询类型 | 示例 | 数量 |
|---|---|---|
| 纯文本查文本 | ”Gunicorn 的 worker 数量怎么设置” | 40 |
| 图片查文本 | 上传报错截图 → 查找解决方案 | 40 |
| 纯文本查图片 | ”找一下那个系统架构图” | 40 |
| 图文混合查 | ”这个表格里的 QPS 数据和文档描述一致吗” | 40 |
| OCR 依赖型 | ”合同第三条的违约金比例是多少”（需要从扫描件提取） | 40 |
三、架构 A 实现：文本化方案
3.1 Pipeline 设计
文档输入
├── 纯文本页面 → 直接分块
├── 含图片页面 → VLM 生成描述 → 拼接原文
└── 扫描件 → OCR → VLM 描述 → 拼接
分块策略：每段不超过 512 tokens，图片和相邻文本绑定
↓
文本 Embedding（BGE-M3）
↓
ChromaDB 向量索引
↓
查询 → 文本 embedding → 检索 → 返回 top-K
3.2 关键实现
VLM 描述生成
from openai import OpenAI
def generate_image_description(image_path: str, context: str = "") -> str:
"""
用 VLM 生成图片的技术性描述。
context 是图片周围的文字，用于引导描述方向。
"""
client = OpenAI()
prompt = f"""请作为技术文档助手，详细描述这张图片。
当前页面上下文：{context[:200]}
要求：
1. 如果是架构图，描述组件名称、连接关系、数据流向
2. 如果是截图，描述界面元素、报错信息、按钮状态
3. 如果是表格，提取所有行和列的内容（结构化输出）
4. 如果是照片（白板/手写），尽量识别文字内容
5. 保留精确的数值、版本号和错误码"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "user", "content": [
{"type": "text", "text": prompt},
{"type": "image_url", "image_url": {
"url": f"data:image/jpeg;base64,{encode_image(image_path)}",
"detail": "high"
}}
]}
],
max_tokens=1024,
)
return response.choices[0].message.content
OCR 预处理
import pytesseract
from pdf2image import convert_from_path
def extract_from_scan(pdf_path: str, page_num: int) -> str:
"""从扫描件 PDF 提取文本（支持中文 OCR）"""
images = convert_from_path(pdf_path, first_page=page_num, last_page=page_num, dpi=300)
image = images[0]
# 中文 OCR
text = pytesseract.image_to_string(image, lang='chi_sim+eng')
return text.strip()
图文绑定分块
class MultimodalChunker:
"""
图片和相邻文本绑定的分块策略。
不能把图片和它的说明文字拆到不同的 chunk 里。
"""
def __init__(self, max_tokens: int = 512):
self.max_tokens = max_tokens
def chunk_document(self, pages: list[dict]) -> list[dict]:
"""
pages: [
{"type": "text", "content": "..."},
{"type": "image", "path": "...", "caption": "..."},
{"type": "text", "content": "..."},
]
"""
chunks = []
current_chunk = {"text": "", "images": []}
current_tokens = 0
for page in pages:
if page["type"] == "text":
tokens = estimate_tokens(page["content"])
if current_tokens + tokens > self.max_tokens:
if current_chunk["text"].strip():
chunks.append(current_chunk)
current_chunk = {"text": page["content"], "images": []}
current_tokens = tokens
else:
current_chunk["text"] += "\n" + page["content"]
current_tokens += tokens
elif page["type"] == "image":
# 图片描述作为文本的一部分绑定
desc = generate_image_description(page["path"], current_chunk["text"])
current_chunk["text"] += f"\n[图片描述] {desc}\n"
current_chunk["images"].append(page["path"])
if current_chunk["text"].strip():
chunks.append(current_chunk)
return chunks
3.3 实测结果
| 查询类型 | Recall@10 | 延迟（中位数） | 备注 |
|---|---|---|---|
| 纯文本查文本 | 85% | 120ms | 接近纯文本 RAG 水平 |
| 图片查文本 | 52% | 2.1s | VLM 描述质量是瓶颈 |
| 纯文本查图片 | 38% | 120ms | 依赖描述的匹配度 |
| 图文混合查 | 31% | 2.2s | 描述信息损失导致漏召回 |
| OCR 依赖型 | 61% | 1.8s | OCR 错误率影响大 |
| 综合 | 53.4% | 1.6s |
关键问题：架构 A 的核心瓶颈在于 VLM 描述生成的信息损失。一张包含 30 个参数的表格，VLM 的描述可能只提取了 15 个；一张架构图中的颜色编码和箭头方向，描述中常常丢失。
四、架构 B 实现：双塔独立检索
4.1 架构设计
文本索引 图片索引
┌─────────────┐ ┌─────────────┐
│ 文本分块 │ │ 图片缩略图 │
│ BGE-M3 │ │ CLIP ViT-L/14│
│ embedding │ │ embedding │
│ → ChromaDB │ │ → ChromaDB │
└──────┬──────┘ └──────┬──────┘
│ │
│ 文本查询 │ 图片查询
▼ ▼
┌─────────────────────────────────────────┐
│ Fusion Scoring Layer │
│ │
│ score = α × text_score + β × img_score │
│ + γ × cross_modal_score │
│ │
│ 动态权重： │
│ - 纯文本查询：α=0.9, β=0.1 │
│ - 图片查询：α=0.3, β=0.7 │
│ - 混合查询：α=0.5, β=0.5 │
└──────────────────┬──────────────────────┘
▼
排序 + 去重 + 返回
4.2 关键实现
图片 Embedding
import torch
from PIL import Image
from transformers import CLIPModel, CLIPProcessor
class ImageEncoder:
"""使用 CLIP 模型编码图片"""
def __init__(self, model_name: str = "openai/clip-vit-large-patch14"):
self.model = CLIPModel.from_pretrained(model_name)
self.processor = CLIPProcessor.from_pretrained(model_name)
self.model.eval()
@torch.no_grad()
def encode(self, image_path: str) -> torch.Tensor:
image = Image.open(image_path).convert("RGB")
inputs = self.processor(images=image, return_tensors="pt")
embedding = self.model.get_image_features(**inputs)
# L2 归一化
embedding = embedding / embedding.norm(p=2, dim=-1, keepdim=True)
return embedding.squeeze().numpy()
@torch.no_grad()
def encode_text(self, text: str) -> torch.Tensor:
"""用 CLIP 的文本编码器编码查询文本"""
inputs = self.processor(text=[text], return_tensors="pt", padding=True)
embedding = self.model.get_text_features(**inputs)
embedding = embedding / embedding.norm(p=2, dim=-1, keepdim=True)
return embedding.squeeze().numpy()
融合排序
import numpy as np
class FusionScorer:
"""
双塔检索结果的融合排序。
核心挑战：两套 embedding 空间的分数不可直接比较。
"""
def __init__(self):
# 通过校准集统计得到的分数分布参数
self.text_mean, self.text_std = 0.72, 0.15
self.img_mean, self.img_std = 0.58, 0.20
def normalize_score(self, raw_score: float, space: str) -> float:
"""将原始余弦相似度归一化到 [0, 1] 区间"""
if space == "text":
mean, std = self.text_mean, self.text_std
else:
mean, std = self.img_mean, self.img_std
return max(0, min(1, (raw_score - mean) / std))
def fuse(
self,
text_results: list[dict],
img_results: list[dict],
query_type: str,
) -> list[dict]:
"""融合排序"""
weights = {
"text_only": {"text": 0.90, "img": 0.10},
"image_only": {"text": 0.30, "img": 0.70},
"hybrid": {"text": 0.50, "img": 0.50},
}
w = weights.get(query_type, weights["hybrid"])
# 构建文档 ID → 分数的映射
scores: dict[str, dict] = {}
for r in text_results:
doc_id = r["id"]
scores[doc_id] = {
"text": self.normalize_score(r["score"], "text"),
"img": 0.0,
"metadata": r["metadata"],
}
for r in img_results:
doc_id = r["id"]
if doc_id not in scores:
scores[doc_id] = {
"text": 0.0,
"img": self.normalize_score(r["score"], "img"),
"metadata": r["metadata"],
}
else:
scores[doc_id]["img"] = self.normalize_score(r["score"], "img")
# 计算融合分数
fused = []
for doc_id, s in scores.items():
combined = w["text"] * s["text"] + w["img"] * s["img"]
fused.append({
"id": doc_id,
"score": combined,
"text_score": s["text"],
"img_score": s["img"],
"metadata": s["metadata"],
})
return sorted(fused, key=lambda x: x["score"], reverse=True)
完整的检索 Pipeline
class DualTowerRAG:
def __init__(self):
self.text_index = ChromaClient(collection="text_chunks")
self.img_index = ChromaClient(collection="image_embeddings")
self.image_encoder = ImageEncoder()
self.fusion_scorer = FusionScorer()
self.text_embedder = BGEM3Embedder() # 假设的 BGE-M3 封装
def ingest_document(self, doc: Document):
"""文档入库：文本和图片分别索引"""
for chunk in doc.text_chunks:
embedding = self.text_embedder.encode(chunk.text)
self.text_index.add(
embeddings=[embedding],
documents=[chunk.text],
metadatas=[{"doc_id": doc.id, "chunk_id": chunk.id}],
ids=[f"text_{doc.id}_{chunk.id}"],
)
for img in doc.images:
embedding = self.image_encoder.encode(img.path)
self.img_index.add(
embeddings=[embedding.tolist()],
documents=[img.description], # CLIP 也存文本描述做辅助
metadatas=[{
"doc_id": doc.id,
"img_id": img.id,
"page": img.page_num,
}],
ids=[f"img_{doc.id}_{img.id}"],
)
def query(self, query_text: str, query_image: str = None) -> list[dict]:
"""多模态查询"""
query_type = "image_only" if query_image and not query_text else \
"text_only" if query_text and not query_image else "hybrid"
# 文本检索
text_results = []
if query_text:
text_embedding = self.text_embedder.encode(query_text)
text_results = self.text_index.query(
query_embeddings=[text_embedding], n_results=20
)
# 图片检索
img_results = []
if query_image:
img_embedding = self.image_encoder.encode(query_image)
img_results = self.img_index.query(
query_embeddings=[img_embedding.tolist()], n_results=20
)
# 融合排序
return self.fusion_scorer.fuse(text_results, img_results, query_type)
4.3 实测结果
| 查询类型 | Recall@10 | 延迟（中位数） | 备注 |
|---|---|---|---|
| 纯文本查文本 | 88% | 130ms | 文本塔用 BGE-M3，质量高 |
| 图片查文本 | 71% | 85ms | CLIP 直接匹配，无需 VLM 描述 |
| 纯文本查图片 | 65% | 85ms | 文搜图，CLIP 跨模态能力 |
| 图文混合查 | 58% | 170ms | 双塔融合，比架构 A 提升 87% |
| OCR 依赖型 | 45% | 170ms | 双塔不解决 OCR 问题 |
| 综合 | 65.4% | 129ms |
关键发现：架构 B 在图片相关查询上大幅领先架构 A（71% vs 52%），因为 CLIP 直接编码图片，不经过 VLM 描述的信息损失。但 OCR 依赖型查询反而更差——双塔架构没有 OCR 预处理步骤。
五、架构 C 实现：统一多模态嵌入
5.1 模型选择
2026 年可用的统一多模态 embedding 模型：
| 模型 | 维度 | 文搜图 | 图搜文 | 图搜图 | 文搜文 |
|---|---|---|---|---|---|
| CLIP ViT-L/14 | 768 | ⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐ |
| BLIP-2 | 768 | ⭐⭐⭐ | ⭐⭐ | ⭐⭐⭐ | ⭐⭐ |
| EVA-CLIP-8B | 1024 | ⭐⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐⭐ |
| Jina CLIP v2 | 768 | ⭐⭐⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐⭐⭐ |
选型：Jina CLIP v2——文本检索精度最接近专用文本模型，同时保持了良好的跨模态对齐能力。
5.2 完整实现
from jina_clip import JinaCLIPEncoder
class UnifiedMultimodalRAG:
"""
统一多模态 RAG：文本和图片映射到同一向量空间。
核心优势：一套索引，天然支持所有查询类型。
"""
def __init__(self, index_path: str = "./multimodal_index"):
self.encoder = JinaCLIPEncoder(model="jina-clip-v2")
self.index = ChromaClient(path=index_path, collection="unified")
def ingest(self, doc: Document):
"""文档入库：文本和图片统一编码"""
items = []
for chunk in doc.text_chunks:
embedding = self.encoder.encode_text(chunk.text)
items.append({
"id": f"text_{doc.id}_{chunk.id}",
"embedding": embedding,
"content": chunk.text,
"modality": "text",
"doc_id": doc.id,
})
for img in doc.images:
embedding = self.encoder.encode_image(img.path)
items.append({
"id": f"img_{doc.id}_{img.id}",
"embedding": embedding,
"content": img.description, # 辅助文本
"modality": "image",
"doc_id": doc.id,
"image_path": img.path,
})
# 批量入库
self.index.add(
embeddings=[item["embedding"] for item in items],
documents=[item["content"] for item in items],
metadatas=[{
"modality": item["modality"],
"doc_id": item["doc_id"],
} for item in items],
ids=[item["id"] for item in items],
)
def query(
self,
query_text: str = None,
query_image: str = None,
top_k: int = 10,
) -> list[dict]:
"""
统一查询：无论查询是文本还是图片，都走同一个 embedding 空间。
"""
if query_image and not query_text:
embedding = self.encoder.encode_image(query_image)
elif query_text and not query_image:
embedding = self.encoder.encode_text(query_text)
else:
# 混合查询：对文本和图片 embedding 取平均
text_emb = self.encoder.encode_text(query_text)
img_emb = self.encoder.encode_image(query_image)
embedding = (text_emb + img_emb) / 2
results = self.index.query(
query_embeddings=[embedding], n_results=top_k * 2
)
# 后处理：同一文档去重，按文档聚合
return self._deduplicate_and_aggregate(results, top_k)
def _deduplicate_and_aggregate(self, results: dict, top_k: int) -> list[dict]:
"""同一文档的多个 chunk/图片聚合为一个结果"""
doc_scores: dict[str, float] = {}
for i, doc_id in enumerate(results["ids"][0]):
actual_doc_id = results["metadatas"][0][i]["doc_id"]
score = 1 - results["distances"][0][i]
if actual_doc_id not in doc_scores or score > doc_scores[actual_doc_id]:
doc_scores[actual_doc_id] = score
sorted_docs = sorted(doc_scores.items(), key=lambda x: x[1], reverse=True)
return [{"doc_id": d, "score": s} for d, s in sorted_docs[:top_k]]
5.3 实测结果
| 查询类型 | Recall@10 | 延迟（中位数） | 备注 |
|---|---|---|---|
| 纯文本查文本 | 79% | 95ms | 不如 BGE-M3（88%） |
| 图片查文本 | 73% | 90ms | 和架构 B 相当 |
| 纯文本查图片 | 70% | 90ms | 比架构 B 略好 |
| 图文混合查 | 62% | 95ms | 天然支持混合查询 |
| OCR 依赖型 | 48% | 95ms | 仍需额外 OCR 预处理 |
| 综合 | 66.4% | 93ms |
关键发现：架构 C 在综合指标上领先，主要是因为纯文本查询的差距不大（79% vs 88%），而图片相关查询和混合查询都表现良好。延迟也最低，因为只查一个索引。
六、三方案横评
6.1 综合对比
| 维度 | A: 文本化 | B: 双塔 | C: 统一嵌入 |
|---|---|---|---|
| 综合 Recall@10 | 53.4% | 65.4% | 66.4% |
| 纯文本查文本 | 85% | 88% | 79% |
| 图片查文本 | 52% | 71% | 73% |
| 纯文本查图片 | 38% | 65% | 70% |
| 图文混合查 | 31% | 58% | 62% |
| 中位延迟 | 1.6s | 129ms | 93ms |
| 索引存储 | 800MB | 1.5GB（双份） | 900MB |
| 实现复杂度 | ⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐⭐ |
| 维护成本 | 低 | 高（双索引） | 中 |
6.2 决策树
你的知识库中图片占比？
├── < 10%（以文字为主）
│ └── 选 A（文本化）——成本最低，图片查询需求少
├── 10%~40%
│ └── 你的查询类型？
│ ├── 主要是文本查图片 → 选 C（统一嵌入）
│ ├── 主要是图片查文本 → 选 B（双塔）或 C
│ └── 两种都有 → 选 C（统一嵌入）
└── > 40%（图片密集型）
├── 需要纯文本最高精度？
│ ├── 是 → 选 B（双塔，文本塔用专用模型）
│ └── 否 → 选 C（统一嵌入）
└── 有 OCR 需求？
└── 是 → 无论选哪个，都要加 OCR 预处理层
七、生产级 Pipeline 完整方案
7.1 架构选型：C + OCR 预处理
经过实测，架构 C（统一嵌入）+ 独立 OCR 层是性价比最高的方案：
文档输入 Pipeline
├── 文本页面 → 分块
├── 含图片页面 → 图片 + 图片描述绑定
├── 扫描件/图片PDF → OCR → 文本 + 原文图片
└── 表格页 → 表格解析（Markdown） + 表格截图
统一 Embedding（Jina CLIP v2）
├── 文本 chunk → text embedding
├── 图片 → image embedding
└── OCR 文本 → text embedding
↓
ChromaDB 单一索引（带 modality 标记）
↓
查询路由
├── 纯文本 → text embedding → 检索
├── 图片 → image embedding → 检索
└── 混合 → 加权平均 → 检索
7.2 完整的文档处理 Pipeline
import os
from pathlib import Path
from PIL import Image
from pdf2image import convert_from_path
class MultimodalDocProcessor:
"""
多模态文档处理 Pipeline。
输入：PDF / 图片 / Markdown
输出：统一格式的 chunks，可直接送入 embedding 模型
"""
def __init__(self, ocr_enabled: bool = True):
self.ocr_enabled = ocr_enabled
self.vlm = VLMClient()
self.ocr = TesseractOCR(lang="chi_sim+eng")
def process_pdf(self, pdf_path: str) -> list[Chunk]:
chunks = []
images = convert_from_path(pdf_path, dpi=300)
for page_num, img in enumerate(images):
page_chunks = self._process_page(img, page_num, pdf_path)
chunks.extend(page_chunks)
return chunks
def _process_page(self, page_image: Image.Image, page_num: int, pdf_path: str) -> list[Chunk]:
chunks = []
# Step 1: 检测页面类型
page_type = self._detect_page_type(page_image)
if page_type == "scan":
# 扫描件：OCR 为主，保留原图
text = self.ocr.extract(page_image)
desc = self.vlm.describe(page_image, context=text[:200])
chunks.append(Chunk(
text=f"[OCR] {text}\n[描述] {desc}",
image_path=self._save_page_image(page_image, pdf_path, page_num),
page_num=page_num,
chunk_type="ocr_scan",
))
elif page_type == "mixed":
# 混合页面：提取文字 + 处理图片
text = self._extract_text_from_pdf_page(pdf_path, page_num)
images = self._extract_images_from_page(page_image)
for idx, sub_img in enumerate(images):
desc = self.vlm.describe(sub_img, context=text[:300])
chunks.append(Chunk(
text=f"[图片] {desc}",
image_path=self._save_sub_image(sub_img, pdf_path, page_num, idx),
page_num=page_num,
chunk_type="image_with_text",
nearby_text=text[:500], # 绑定相邻文本
))
# 纯文本部分
if text.strip():
chunks.append(Chunk(
text=text,
image_path=None,
page_num=page_num,
chunk_type="text_only",
))
elif page_type == "text_only":
text = self._extract_text_from_pdf_page(pdf_path, page_num)
chunks.append(Chunk(
text=text,
image_path=None,
page_num=page_num,
chunk_type="text_only",
))
return chunks
def _detect_page_type(self, image: Image.Image) -> str:
"""简单的页面类型检测"""
# 可以用 LayoutLM / DocLayout-YOLO 等模型做精确检测
# 这里用启发式规则
gray = image.convert("L")
# 低对比度 + 噪声 = 扫描件
if self._estimate_noise_level(gray) > 0.15:
return "scan"
# 有大面积均匀区域 = 含图片
if self._has_large_uniform_regions(gray) > 0.3:
return "mixed"
return "text_only"
def _estimate_noise_level(self, gray_image) -> float:
"""估计图像噪声水平（判断是否为扫描件）"""
import numpy as np
arr = np.array(gray_image)
# 局部方差均值
from scipy.ndimage import uniform_filter
mean = uniform_filter(arr, size=3)
variance = uniform_filter(arr*arr, size=3) - mean*mean
return float(np.mean(variance) / 255**2)
def _has_large_uniform_regions(self, gray_image) -> float:
"""检测大面积均匀区域（判断是否有图片/图表）"""
import numpy as np
arr = np.array(gray_image)
from scipy.ndimage import uniform_filter
variance = uniform_filter(arr.astype(float)**2, size=16) - uniform_filter(arr.astype(float), size=16)**2
return float(np.mean(variance < 100))
7.3 检索优化：Metadata 过滤
class OptimizedMultimodalRetriever:
"""
带 metadata 过滤的多模态检索器。
解决：纯语义召回会混入"相似但不相关"的结果。
"""
def __init__(self, index, encoder):
self.index = index
self.encoder = encoder
def query(
self,
text: str = None,
image: str = None,
filter_doc_type: str = None,
filter_page_range: tuple = None,
prefer_modality: str = None,
top_k: int = 10,
) -> list[dict]:
"""
支持多维度过滤的检索。
Args:
prefer_modality: "text" 或 "image"，优先返回某种模态的结果
"""
# 编码
if image and not text:
embedding = self.encoder.encode_image(image)
elif text and not image:
embedding = self.encoder.encode_text(text)
else:
text_emb = self.encoder.encode_text(text or "")
img_emb = self.encoder.encode_image(image) if image else text_emb
embedding = (text_emb + img_emb) / 2
# 构建过滤条件
where_filter = {}
if filter_doc_type:
where_filter["doc_type"] = filter_doc_type
if filter_page_range:
where_filter["page_num"] = {
"$gte": filter_page_range[0],
"$lte": filter_page_range[1],
}
# 检索
results = self.index.query(
query_embeddings=[embedding.tolist()],
n_results=top_k * 3, # 多取一些用于后续排序
where=where_filter if where_filter else None,
)
# 模态偏好排序
if prefer_modality:
scored = []
for i, meta in enumerate(results["metadatas"][0]):
score = 1 - results["distances"][0][i]
if prefer_modality == "text" and meta.get("modality") == "text":
score *= 1.2 # 文本结果加权
elif prefer_modality == "image" and meta.get("modality") == "image":
score *= 1.2
scored.append((score, i))
scored.sort(reverse=True)
top_indices = [i for _, i in scored[:top_k]]
else:
top_indices = list(range(min(top_k, len(results["ids"][0]))))
return [
{
"id": results["ids"][0][i],
"text": results["documents"][0][i],
"metadata": results["metadatas"][0][i],
"score": 1 - results["distances"][0][i],
}
for i in top_indices
]
八、踩坑记录
踩坑 1：CLIP 对中文文本的理解能力不足
现象：用 CLIP 做纯文本检索时，中文查询的 Recall@10 只有 45%，远低于英文查询的 82%。
原因：CLIP 的训练数据以英文为主，中文语义空间对齐较差。
修复：切换到 Jina CLIP v2（专门优化了多语言支持），中文 Recall@10 提升到 79%。如果必须用原始 CLIP，可以先用 LLM 把中文查询翻译成英文再检索。
踩坑 2：图片描述和 OCR 文本的重复导致 embedding 偏差
现象：同一页面的 OCR 文本和图片描述高度相似（都描述了同一张表格），导致两个 chunk 的 embedding 几乎相同，检索时同时返回，浪费 top-K 名额。
修复：在入库前做语义去重——如果两个 chunk 的 embedding 余弦相似度 > 0.95，只保留信息量更大的那个（以文本长度 + 包含的实体数量衡量）。
踩坑 3：PDF 中的矢量图 vs 光栅图处理差异
现象：PDF 中的矢量图（用 matplotlib 生成的图表）提取为 PNG 后，CLIP embedding 质量很好；但扫描件中的模糊图表，embedding 质量显著下降。
修复：对页面做质量评分（分辨率、对比度、噪声水平），低于阈值时标记为”低质量图片”，检索时降低该结果的权重。同时为低质量图片强制走 OCR + VLM 描述的增强路径。
踩坑 4：混合查询时文本和图片 embedding 的尺度不一致
现象：查询文本和图片各编码后取平均，但两者的 embedding 尺度不同，导致某一种模态主导了检索结果。
修复：在取平均前各自做 L2 归一化，确保两者在同一尺度上：
text_emb = text_emb / np.linalg.norm(text_emb)
img_emb = img_emb / np.linalg.norm(img_emb)
embedding = 0.5 * text_emb + 0.5 * img_emb
embedding = embedding / np.linalg.norm(embedding) # 再次归一化
踩坑 5：ChromaDB 的 metadata 过滤和向量检索的组合查询性能
现象：加上了 where 过滤条件后，查询延迟从 90ms 飙升到 800ms。
原因：ChromaDB 的 metadata 过滤不是基于索引的，而是先做向量检索再过滤，过滤条件越严格，需要检索的候选集越大。
修复：
- 增大
n_results到 top_k 的 3-5 倍，给过滤留出足够候选 - 对高频过滤字段（如
doc_type）做预分片——按文档类型建多个 collection，而不是用 metadata 过滤 - 考虑切换到 Qdrant，它的 payload 过滤是基于 HNSW 索引的，性能更好
九、总结
多模态 RAG 不是”把图片转成文字然后走老路”。它有自己独特的架构选择、检索策略和性能陷阱。
核心结论：
- 架构 C（统一嵌入）是当前最优解——一套索引，天然支持所有查询类型，延迟最低，维护成本可控。
- OCR 是独立层——无论选哪个架构，扫描件都需要独立的 OCR 预处理，这不是 embedding 模型能解决的。
- 中文场景选对模型——原始 CLIP 对中文支持差，必须选多语言优化的模型（Jina CLIP v2 / EVA-CLIP-8B）。
- Metadata 过滤要谨慎——ChromaDB 的过滤性能差，高频过滤场景考虑用 Qdrant 或预分片策略。
如果你的知识库里有超过 10% 的图片内容，纯文本 RAG 已经在暗中丢失 30-50% 的召回率。多模态 RAG 的投入是值得的。