大模型应用:多模态RAG实现

传统 RAG 只处理纯文本,但真实文档往往包含图片、表格、图表等多种元素。多模态 RAG 将这些内容统一纳入检索和生成流程,显著提升文档问答的效果。

多模态 RAG 的核心挑战

标准的文本 RAG 流程:文档分块 -> 文本 Embedding -> 向量检索 -> LLM 生成。多模态 RAG 需要处理的额外问题:

  1. 文档解析:怎么从 PDF/Word 中提取图片、表格,并保留与周围文本的关联
  2. 多模态 Embedding:文本和图片需要映射到同一个向量空间
  3. 检索策略:怎么同时检索文本和图片,如何排序
  4. 生成阶段:把图片和文本一起传给多模态 LLM

文档解析

文档解析是整个流程的基础。推荐使用 unstructured 库,它能处理 PDF、Word、HTML 等多种格式,并自动识别文本、表格、图片等元素。

from unstructured.partition.pdf import partition_pdf

elements = partition_pdf(
    filename="report.pdf",
    strategy="hi_res",          # 高精度模式,使用 OCR + 布局检测
    infer_table_structure=True,  # 推断表格结构
    extract_images_in_pdf=True,  # 提取嵌入的图片
    extract_image_block_output_dir="./extracted_images",
)

# elements 包含多种类型
for elem in elements:
    print(type(elem).__name__, elem.text[:80] if elem.text else "[image]")
    # NarrativeText, Table, Image, Title, ListItem, ...

strategy="hi_res" 模式会调用布局检测模型(基于 detectron2)来识别页面元素的位置和类型。对于扫描版 PDF 效果显著,但处理速度较慢。如果是文字版 PDF,用 "fast" 策略就够了。

表格会被解析为 HTML 格式的字符串,方便后续处理:

tables = [el for el in elements if type(el).__name__ == "Table"]
for table in tables:
    print(table.metadata.text_as_html)
    # <table><tr><td>Q1</td><td>Revenue</td>...</tr>...</table>

多模态 Embedding

要让文本和图片可以互相检索,需要将它们映射到同一个向量空间。CLIP(Contrastive Language-Image Pre-training)是最常用的选择。

from sentence_transformers import SentenceTransformer
from PIL import Image

# 加载 CLIP 模型
model = SentenceTransformer("clip-ViT-L-14")

# 文本 Embedding
text_embeddings = model.encode(["公司Q2营收增长图", "技术架构示意图"])

# 图片 Embedding
img = Image.open("extracted_images/figure-1.jpg")
img_embedding = model.encode(img)

# 两者在同一个向量空间,可以计算相似度
from sentence_transformers.util import cos_sim
similarity = cos_sim(text_embeddings[0], img_embedding)
print(f"相似度: {similarity.item():.4f}")

CLIP 的优势是天然支持跨模态检索:用文本查询可以找到相关图片,反之亦然。缺点是对中文的理解能力有限,可以考虑使用 Chinese-CLIP 或者多模态 Embedding 服务(比如 Cohere 的 embed-v3)。

对于表格,建议同时保留两种表示:

def process_table(table_element):
    '''将表格转换为文本描述 + 原始 HTML'''
    html = table_element.metadata.text_as_html
    # 用 LLM 生成表格的文字摘要
    summary = llm.invoke(
        f"用一段话描述这个表格的内容和关键数据:\n{html}"
    )
    return {
        "type": "table",
        "html": html,
        "summary": summary,
        "embedding": model.encode(summary),
    }

用文字摘要做检索(因为用户查询通常是文本),用原始 HTML 做生成(保留精确数据)。

构建多模态索引

把解析出的所有元素统一存入向量数据库:

import chromadb

client = chromadb.PersistentClient(path="./multimodal_db")
collection = client.get_or_create_collection(
    name="documents",
    metadata={"hnsw:space": "cosine"},
)

def index_document(elements, doc_id):
    '''将文档的所有元素索引到向量数据库'''
    for i, elem in enumerate(elements):
        elem_type = type(elem).__name__

        if elem_type == "Image":
            img_path = elem.metadata.image_path
            img = Image.open(img_path)
            embedding = model.encode(img).tolist()
            metadata = {
                "type": "image",
                "doc_id": doc_id,
                "image_path": img_path,
                "page": elem.metadata.page_number,
            }
            text_content = f"[图片: 第{elem.metadata.page_number}页]"

        elif elem_type == "Table":
            processed = process_table(elem)
            embedding = processed["embedding"].tolist()
            metadata = {
                "type": "table",
                "doc_id": doc_id,
                "html": processed["html"],
                "page": elem.metadata.page_number,
            }
            text_content = processed["summary"]

        else:
            if not elem.text or len(elem.text.strip()) < 20:
                continue
            embedding = model.encode(elem.text).tolist()
            metadata = {
                "type": "text",
                "doc_id": doc_id,
                "page": elem.metadata.page_number,
            }
            text_content = elem.text

        collection.add(
            ids=[f"{doc_id}_{i}"],
            embeddings=[embedding],
            documents=[text_content],
            metadatas=[metadata],
        )

检索策略

多模态检索需要考虑几个问题:

def multimodal_retrieve(query, top_k=5):
    '''多模态检索:同时返回文本、表格、图片'''
    query_embedding = model.encode(query).tolist()

    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k * 2,  # 多取一些,后面过滤
    )

    # 按类型分组
    texts, tables, images = [], [], []
    for doc, meta, dist in zip(
        results["documents"][0],
        results["metadatas"][0],
        results["distances"][0],
    ):
        item = {"content": doc, "metadata": meta, "score": 1 - dist}
        if meta["type"] == "text":
            texts.append(item)
        elif meta["type"] == "table":
            tables.append(item)
        else:
            images.append(item)

    # 保证多样性:至少包含不同类型的结果
    final = []
    final.extend(texts[:3])
    final.extend(tables[:2])
    final.extend(images[:2])

    # 按分数重新排序
    final.sort(key=lambda x: x["score"], reverse=True)
    return final[:top_k]

实践中还可以加入 reranking 步骤,用交叉编码器(Cross-Encoder)对检索结果做精排。

多模态生成

最终生成阶段,需要把文本和图片一起传给多模态 LLM:

import base64
from openai import OpenAI

client = OpenAI()

def generate_answer(query, retrieved_items):
    '''使用多模态 LLM 生成回答'''
    messages = [
        {"role": "system", "content": "基于提供的文档内容回答问题。内容可能包含文本、表格和图片。"},
    ]

    # 构建用户消息:混合文本和图片
    user_content = [{"type": "text", "text": f"问题:{query}\n\n相关内容:\n"}]

    for item in retrieved_items:
        if item["metadata"]["type"] == "image":
            img_path = item["metadata"]["image_path"]
            with open(img_path, "rb") as f:
                b64 = base64.b64encode(f.read()).decode()
            user_content.append({
                "type": "image_url",
                "image_url": {"url": f"data:image/jpeg;base64,{b64}"},
            })
            user_content.append({
                "type": "text",
                "text": f"(以上图片来自第{item['metadata']['page']}页)",
            })
        elif item["metadata"]["type"] == "table":
            user_content.append({
                "type": "text",
                "text": f"表格内容:\n{item['metadata']['html']}",
            })
        else:
            user_content.append({
                "type": "text",
                "text": item["content"],
            })

    messages.append({"role": "user", "content": user_content})

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        max_tokens=2000,
    )
    return response.choices[0].message.content

效果提升点

几个实践中有效的优化方向:

分块策略:不要按固定长度分块,按语义段落分。unstructuredchunk_by_title 可以按标题层级分块,保留文档结构。

上下文窗口:检索到图片时,同时把图片前后的文本段落一起送给 LLM,帮助模型理解图片上下文。

表格处理:对于复杂表格,用 LLM 预处理生成多角度的文字描述("这个表格展示了..."、"从表中可以看出..."),每个描述单独索引,提升召回率。

Embedding 模型选择:纯文本部分可以用专门的文本 Embedding 模型(如 bge-large),图片相关部分用 CLIP。混合索引在检索时分别查询再合并结果。

多模态 RAG 的工程量比纯文本 RAG 大不少,但对于包含丰富图表的文档(财报、技术手册、研究报告),效果提升非常明显。