想象您在组装宜家（IKEA）家具。说明书的第 14 步摆在眼前，可就是弄不清哪颗螺丝该拧到哪儿。图示画了些什么，但就是对不上号。 

如果您问： “第 14 步里那个小金属件是干什么用的？”，标准的纯文本检索增强生成（RAG）系统就帮不上忙了。答案就在图里，而不是页面文字里。宜家说明书出了名地“无字”。

这正是多模态 RAG 大显身手的地方。

在本教程中，我将带您从零构建一个多模态 RAG 系统，能够回答包含图片、图表和混合内容文档中的问题。我们用宜家装配说明作为测试案例，因为它们是很好的“压力测试”：以视觉为主、文字极少，而且确实是人们需要帮助的场景。

我们将全程使用 OpenAI 的模型：用 GPT-4o 理解图像并生成答案，使用 text-embedding-3-small 创建可搜索的向量。若您想用不同框架搭建 RAG 流水线，推荐我们的 使用 LangChain 的 RAG 课程。

完成后，您将拥有一个可在本地运行、并可适配到您自有文档的工作系统。

什么是多模态 RAG？

多模态 RAG 将标准的 RAG 模式扩展到不仅处理文本。核心思想不变，但“上下文”现在包含图像、表格、图示以及其他非文本内容。

为什么这很重要？因为真实世界的文档从不只是文本。

想想您实际会接触到的东西： 

• 带有接线图的技术手册 
• 含实验曲线的研究论文 
• 带收入图表的财报
• 附部件图片的产品规格 

只处理文本的系统掌握的信息不完整。这就像只看配方的食材清单，却忽略那些告诉您“轻柔翻拌”到底长什么样的配图。

传统 RAG 把 PDF 当作字符流：抽取文本、切块、向量化，然后就结束。任何图像要么被完全跳过，要么被转换成 [IMAGE] 这样的占位符。表格会被打散成杂乱的文本行，丢失结构。

多模态 RAG 则不同：图像会被描述或直接向量化，表格会被解析为结构化格式。检索系统可以把“网络拓扑图”这样的查询同时匹配到网络架构的文字描述和真实的视觉图示。生成模型能够查看图像并推理其所表达的内容。

如果“多模态”听起来仍有些虚，本文这篇关于 多模态 AI 的博文有简洁明了的示例。

多模态 RAG 如何工作？

与纯文本 RAG 相比，这条流水线环节更多，但结构类似。数据被处理并向量化，写入向量库，在有人提问时再被检索出来。

流程如下：

文档摄取与解析

原始文档（PDF、图片、幻灯片）会被拆解为组成部分。您需要抽取文本、识别图像、解析表格，并识别它们在空间上的关联。第 5 页的示意图下方可能有一段说明，而第 4 页的正文里又有引用。您需要把这些关联关系记录下来。

内容预处理

不同内容类型需要不同处理方式。对图像，主要有两种方案：用视觉模型（如 GPT-4o）生成文字描述，或用多模态编码器（如 CLIP）直接向量化图像。描述方案更直接，并能与标准文本向量配合。表格可转为结构化格式，或用自然语言总结。

随后，一切都会被转换为向量。若采用描述方案，您只需用标准的文本嵌入模型对所有文本（原始文本 + 生成的图像描述）做向量化。关键在于，关于视觉内容的查询如今可以匹配到这些描述。

索引与存储 

向量存入带元数据的数据库，包括来源文档、页码、内容类型，以及图像的原始文件路径。稍后要把图像展示给生成模型时，您会用到这个路径。

检索

接到查询后，用同一模型对其向量化，搜索最近邻并返回 Top 结果。检索到的片段可能包含文本、图像或两者兼有。

生成

把查询连同检索到的上下文一起传给视觉-语言模型。若检索片段引用了图像，就加载这些图像并包含在提示中。模型同时看到文本上下文与真实图片，从而生成答案。

生成这一步是 GPT-4o 的强项。它能看着展示螺丝位置的宜家示意图，用通俗语言解释发生了什么；能读懂图表并总结趋势；还能注意到技术图纸中的细节，这是文字描述可能遗漏的。

实现多模态 RAG

我们动手搭建一个系统，能够回答关于宜家装配说明的问题，从环境搭建到可用的查询界面，一步步来。

第 0 步：选择框架与工具

本教程力求简单：

您当然可以替换为其他方案。如果需要规模化，向量库可用 Pinecone 或 Weaviate。生成模型也可用 Claude 或 Gemini，总体架构在这些情况下保持不变。

为什么不直接用 CLIP 或 ColPali 做嵌入？当然也可以。它们允许您在不先生成文字描述的情况下直接对图像向量化。但它们会增加复杂度，且往往需要 GPU。对于希望捕捉语义细节、而不只是视觉相似度的“文档型”内容，描述方案已能很好地胜任。

第 1 步：环境搭建

我们先设置依赖与环境变量，搭好项目环境。

安装依赖

首先，您需要安装 poppler 以处理 PDF。

• Ubuntu/Debian：sudo apt-get install poppler-utils

• macOS（使用 Homebrew）：brew install poppler

• Windows：从发布页下载并加入 PATH

接着安装 Python 包。除 ChromaDB 外，我们还需要 pdf2image（将 PDF 转为图像）、pillow（图像处理）和 requests（下载 PDF 文件）。

pip install openai chromadb pdf2image pillow requests技术支持

设置 API 密钥与文件

要使用模型，请将您的 OpenAI API 密钥设置为环境变量。

Unix/Linux/macOS：

export OPENAI_API_KEY="your-api-key-here"技术支持

Windows: 

setx OPENAI_API_KEY="your-api-key-here"技术支持

现在，我们来搭建项目结构并拉取示例数据。脚本已包含下载 URL，如果您想先查看 PDF，这里是 MALM 书桌 和 MALM 床架 的直链。 

创建名为 step1_setup.py 的文件：

"""
Step 1: Setup and Download Sample Data
"""
import os
import requests

# Create directories
for d in ["data", "data/images", "data/cache"]:
    os.makedirs(d, exist_ok=True)
print("✓ Created directories")

# Download IKEA assembly instructions
SAMPLE_PDFS = {
    "malm_desk.pdf": "https://www.ikea.com/us/en/assembly_instructions/malm-desk-white__AA-516949-7-2.pdf",
    "malm_bed.pdf": "https://www.ikea.com/us/en/assembly_instructions/malm-bed-frame-low__AA-75286-15_pub.pdf",
}

for filename, url in SAMPLE_PDFS.items():
    filepath = f"data/{filename}"
    if not os.path.exists(filepath):
        print(f"Downloading {filename}...")
        response = requests.get(url, timeout=30)
        with open(filepath, "wb") as f:
            f.write(response.content)
        print(f"  ✓ Saved ({len(response.content) / 1024:.1f} KB)")
    else:
        print(f"✓ {filename} already exists")技术支持

完成初始设置后，运行脚本：

python step1_setup.py技术支持

您应能在 data/ 文件夹中看到下载好的宜家 PDF。

第 2 步：预处理文档

宜家说明书很“棘手”。它们几乎全是图示，文字极少。我们需要把 PDF 页面转为图像，再用 GPT-4o 描述每页展示的内容。

创建 step2_preprocess.py：

"""
Step 2: Preprocess Documents
Convert PDFs to images and generate descriptions with GPT-4o
"""
import os
import json
import base64
from pdf2image import convert_from_path
from openai import OpenAI

client = OpenAI()

DATA_DIR = "data"
IMAGES_DIR = "data/images"
CACHE_FILE = "data/cache/descriptions.json"


def convert_pdf_to_images(pdf_path):
    """Convert a PDF to page images."""
    pdf_name = os.path.splitext(os.path.basename(pdf_path))[0]
    images = convert_from_path(pdf_path, dpi=150)
    
    image_paths = []
    for i, image in enumerate(images):
        image_path = f"{IMAGES_DIR}/{pdf_name}_page_{i+1:03d}.png"
        image.save(image_path, "PNG")
        image_paths.append(image_path)
    
    return image_paths


def describe_page(image_path):
    """Use GPT-4o to describe an instruction page."""
    with open(image_path, "rb") as f:
        base64_image = base64.b64encode(f.read()).decode("utf-8")
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": """Describe this IKEA assembly instruction page in detail.
Include: step numbers, parts shown, tools needed, actions demonstrated, 
quantities (like "2x"), warnings, and part numbers if visible.
This description will help people find this page when they have questions."""
            },
            {
                "role": "user",
                "content": [
                    {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{base64_image}"}},
                    {"type": "text", "text": "Describe this instruction page."}
                ]
            }
        ],
        max_tokens=500
    )
    return response.choices[0].message.content


def main():
    # Load cache if exists
    cache = {}
    if os.path.exists(CACHE_FILE):
        with open(CACHE_FILE, "r") as f:
            cache = json.load(f)
    
    all_pages = []
    
    # Process each PDF
    for pdf_file in os.listdir(DATA_DIR):
        if not pdf_file.endswith(".pdf"):
            continue
        
        print(f"\nProcessing {pdf_file}...")
        pdf_path = f"{DATA_DIR}/{pdf_file}"
        
        # Convert to images
        print("  Converting to images...")
        image_paths = convert_pdf_to_images(pdf_path)
        print(f"  ✓ {len(image_paths)} pages")
        
        # Generate descriptions
        print("  Generating descriptions with GPT-4o...")
        for i, image_path in enumerate(image_paths):
            if image_path in cache:
                description = cache[image_path]
                status = "cached"
            else:
                description = describe_page(image_path)
                cache[image_path] = description
                # Save cache after each page
                with open(CACHE_FILE, "w") as f:
                    json.dump(cache, f, indent=2)
                status = "generated"
            
            all_pages.append({
                "source_pdf": pdf_file,
                "page_number": i + 1,
                "image_path": image_path,
                "description": description
            })
            print(f"    Page {i+1}/{len(image_paths)} [{status}]")
    
    # Save all page data
    with open("data/cache/all_pages.json", "w") as f:
        json.dump(all_pages, f, indent=2)
    
    print(f"\n✓ Processed {len(all_pages)} pages total")


if __name__ == "__main__":
    main()技术支持

这些函数分别在做什么：

• convert_pdf_to_images(pdf_path)：使用 pdf2image 以 150 DPI 将每个 PDF 页面渲染为 PNG 图像。这样可提取那些无法选择文本的图示，并按统一命名保存，如 malm_desk_page_001.png。

• describe_page(image_path)：将图像编码为 base64，连同定制的系统提示发送给 GPT-4o。模型返回包含步骤、部件与数量等细节的文本描述，用于语义搜索。

• main()：协调一切：加载缓存以避免重复处理，将 PDF 转成图像，按需生成新描述，增量保存缓存，并输出带索引元数据的 all_pages.json。

像这样运行脚本：

python step2_preprocess.py技术支持

这一步需要几分钟。GPT-4o 会分析每一页并生成详细描述。描述会被缓存，因此再次运行时会跳过已处理页面。

这里到底发生了什么？我们正在把视觉内容“转写”为可搜索的文本。当有人问“我该如何安装桌腿？”时，检索器会在这些描述中搜索，并找到提到安装桌腿的页面，即便原始页面根本没有文字。

当您把基础流水线搭建起来后，接下来的提升点来自检索技巧，比如混合搜索、重排序或查询改写。要熟悉这些，请阅读这篇博文：高级 RAG 技术。

第 3 步：存储与索引多模态数据

现在我们对描述做嵌入，并将其存入 ChromaDB。

创建 step3_index.py：

"""
Step 3: Create Embeddings and Index
"""
import os
import json
from openai import OpenAI
import chromadb

client = OpenAI()

PAGES_FILE = "data/cache/all_pages.json"
CHROMA_DIR = "data/chroma_db"
COLLECTION_NAME = "ikea_instructions"


def get_embedding(text):
    """Get embedding from OpenAI."""
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding


def main():
    # Load page descriptions
    with open(PAGES_FILE, "r") as f:
        pages = json.load(f)
    print(f"Loaded {len(pages)} pages")
    
    # Initialize ChromaDB with persistent storage
    chroma_client = chromadb.PersistentClient(path=CHROMA_DIR)
    
    # Delete existing collection if present
    try:
        chroma_client.delete_collection(name=COLLECTION_NAME)
    except:
        pass
    
    collection = chroma_client.create_collection(
        name=COLLECTION_NAME,
        metadata={"hnsw:space": "cosine"}
    )
    print(f"Created collection: {COLLECTION_NAME}")
    
    # Index each page
    print("\nCreating embeddings...")
    for i, page in enumerate(pages):
        # Combine source info with description
        text_to_embed = f"Source: {page['source_pdf']}, Page {page['page_number']}\n\n{page['description']}"
        
        embedding = get_embedding(text_to_embed)
        
        collection.add(
            ids=[f"page_{i}"],
            embeddings=[embedding],
            documents=[page["description"]],
            metadatas=[{
                "source_pdf": page["source_pdf"],
                "page_number": page["page_number"],
                "image_path": page["image_path"]
            }]
        )
        
        if (i + 1) % 5 == 0 or i == len(pages) - 1:
            print(f"  Indexed {i + 1}/{len(pages)} pages")
    
    print(f"\n✓ Index created with {collection.count()} documents")
    
    # Quick retrieval test
    print("\nTesting retrieval...")
    test_query = "How do I attach the legs?"
    query_embedding = get_embedding(test_query)
    results = collection.query(query_embeddings=[query_embedding], n_results=2)
    
    print(f"Query: '{test_query}'")
    for meta in results["metadatas"][0]:
        print(f"  → {meta['source_pdf']}, Page {meta['page_number']}")


if __name__ == "__main__":
    main()技术支持

这里，get_embedding(text) 使用 OpenAI 的 text-embedding-3-small 模型将描述与来源信息转换为稠密向量。该模型输出 1536 维向量，能捕捉语义用于相似度搜索。

main() 会从上一步的 all_pages.json 读取已处理页面，初始化持久化的 ChromaDB，删除/重建集合，为每页生成嵌入，连同丰富元数据（来源、页码、图像路径）写入，并做一次快速检索测试验证索引可用。

运行它：

python step3_index.py技术支持

这里的元数据很重要。我们存了 image_path，以便稍后在生成时取回真实图像并展示给 GPT-4o。检索的是描述，但展示的是图像。

第 4 步：构建 RAG 流水线

下面是完整的多模态 RAG 流水线：对查询向量化，检索相关页面，加载图像，GPT-4o 基于文本描述与视觉共同生成答案。

创建 step4_query.py：

"""
Step 4: Query the Multimodal RAG System
"""
import os
import sys
import base64
from openai import OpenAI
import chromadb

client = OpenAI()

CHROMA_DIR = "data/chroma_db"
COLLECTION_NAME = "ikea_instructions"
TOP_K = 3

# Initialize ChromaDB
chroma_client = chromadb.PersistentClient(path=CHROMA_DIR)


def get_embedding(text):
    """Get embedding from OpenAI."""
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding


def retrieve(query, top_k=TOP_K):
    """Find relevant instruction pages."""
    collection = chroma_client.get_collection(name=COLLECTION_NAME)
    query_embedding = get_embedding(query)
    
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k,
        include=["documents", "metadatas"]
    )
    return results


def generate_answer(query, retrieved_results):
    """Generate answer using GPT-4o with retrieved images."""
    if not retrieved_results["ids"][0]:
        return "No relevant pages found.", []
    
    # Build image content for GPT-4o
    image_content = []
    sources = []
    
    for doc, metadata in zip(retrieved_results["documents"][0], retrieved_results["metadatas"][0]):
        image_path = metadata["image_path"]
        
        if not os.path.exists(image_path):
            continue
        
        with open(image_path, "rb") as f:
            base64_image = base64.b64encode(f.read()).decode("utf-8")
        
        image_content.append({
            "type": "image_url",
            "image_url": {"url": f"data:image/png;base64,{base64_image}"}
        })
        image_content.append({
            "type": "text",
            "text": f"[Page {metadata['page_number']} from {metadata['source_pdf']}]"
        })
        
        sources.append(f"{metadata['source_pdf']}, Page {metadata['page_number']}")
    
    # Generate with GPT-4o
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": """You answer questions about IKEA assembly based on instruction pages.
Be specific, reference page numbers when helpful, and say if something isn't clear from the images."""
            },
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": f"Question: {query}\n\nRelevant instruction pages:"},
                    *image_content,
                    {"type": "text", "text": "Answer based on these pages."}
                ]
            }
        ],
        max_tokens=600
    )
    
    return response.choices[0].message.content, sources


def answer_question(query):
    """Full RAG pipeline."""
    print(f"\nQuery: {query}")
    print("-" * 50)
    
    print("Retrieving relevant pages...")
    results = retrieve(query)
    print(f"  Found {len(results['ids'][0])} pages")
    
    print("Generating answer...")
    answer, sources = generate_answer(query, results)
    
    return answer, sources


def main():
    # Check index exists
    try:
        collection = chroma_client.get_collection(name=COLLECTION_NAME)
        print(f"Connected to index ({collection.count()} documents)")
    except:
        print("ERROR: Index not found. Run step2_index.py first.")
        return
    
    # Single query from command line
    if len(sys.argv) > 1:
        query = " ".join(sys.argv[1:])
        answer, sources = answer_question(query)
        print(f"\nAnswer:\n{answer}")
        print(f"\nSources: {', '.join(sources)}")
        return
    
    # Interactive mode
    print("\nAsk questions about IKEA assembly. Type 'quit' to exit.\n")
    
    while True:
        try:
            query = input("Question: ").strip()
        except KeyboardInterrupt:
            break
        
        if not query or query.lower() in ["quit", "exit"]:
            break
        
        answer, sources = answer_question(query)
        print(f"\nAnswer:\n{answer}")
        print(f"\nSources: {', '.join(sources)}\n")


if __name__ == "__main__":
    main()技术支持

关键函数说明：

• get_embedding(text)：使用 text-embedding-3-small 为查询生成向量嵌入，以便在 ChromaDB 中进行语义相似检索。

• retrieve(query, top_k=TOP_K)：对查询做嵌入，检索集合中的前 k 个最近邻，通过 include=["documents", "metadatas"] 返回文档与元数据（如图像路径）。

• generate_answer(query, retrieved_results)：将 Top 图像以 base64 载入，构建交替插入页面标签与图像的多模态 GPT-4o 提示，生成包含宜家场景细节的回答。

• answer_question(query)：编排检索与生成，带有进度日志。

• main()：校验索引是否存在，支持命令行单次查询或交互式聊天模式（输入 “quit”“exit” 或按 Ctrl+C 退出）。

第 5 步：运行 RAG 流水线

您可以直接运行后输入问题，或携带特定问题运行，如下：

python step4_query.py "How do I attach the legs to the desk?"技术支持

Retrieving relevant pages...
Found 3 pages
Generating answer...

Answer:
To attach the legs to the MALM desk, follow these steps shown on pages 5-6:

1. First, insert the wooden dowels into the pre-drilled holes on the leg 
   pieces (you'll see "4x" indicating four dowels are needed)

2. Apply the cam locks to the desk top using the Phillips screwdriver

3. Align each leg with the corresponding holes on the underside of the desk

4. Tighten the cam locks by rotating them clockwise until the legs are 
   firmly attached

The instructions show this should be done with the desk top face-down on 
a soft surface to avoid scratching.

Sources: malm_desk.pdf, Page 5, malm_desk.pdf, Page 6, malm_desk.pdf, Page 7技术支持

看起来不错！系统根据描述检索页面，加载真实图像，GPT-4o 直接查看图示并生成答案。

多模态 RAG 评估

构建一个多模态 RAG 系统是一回事，判断它是否表现良好又是另一回事，而让它在生产中稳定运行更是多了一层挑战。 

标准的 RAG 指标仍然适用，但需要为视觉内容做扩展。

检索质量

Precision@k 衡量检索到的页面中有多少是真正相关的。它会惩罚 Top-k 中的无关页面，但对生成来说，拿到关键图片往往比做到完美的 Top-k 更重要。

Recall@k 衡量正确页面在您 Top k 结果中出现的频次。对我们的宜家场景更相关，因为我们希望系统在 Top-k（例如 k=3）内至少找到一个相关页面。这样 GPT-4o 就能获得视觉上下文来正确回答，即使其他结果是噪声。 

对于我们的宜家系统，您可以构建一个包含约 50 个问题的测试集，并为每个问题标注“正确”的页面。然后，用如下函数进行评估：

def evaluate_retrieval(test_cases, top_k=3):
    """
    test_cases: [{"query": str, "relevant_pages": ["image_path1", ...]}]
    """
    hits = 0
    for test in test_cases:
        results = retrieve(test["query"], top_k=top_k)
        retrieved = [m["image_path"] for m in results["metadatas"][0]]
        if any(p in retrieved for p in test["relevant_pages"]):
            hits += 1
    return hits / len(test_cases)技术支持

生成质量

生成质量更难自动评测。 

您可以让人工对答案的准确性与有用性打分，或使用 LLM 作为评审，判断答案在给定源图像下是否正确。 

对于带编号步骤的图示，要验证模型是否正确对应步骤编号。

多模态特定检查

进一步的评估可以回答以下问题：

• 当答案是视觉性的，系统是否能找到以图像为主的答案？ 

• 当模型引用图像中的某个元素时，这种引用是否准确？ 

这些至少需要对部分查询进行人工复核。

多模态 RAG 最佳实践

若要构建可用于生产的多模态 RAG 流水线，需要注意以下几点：

• 预处理成本会累积：为图像生成描述的视觉调用才是昂贵部分。对于大型文档库，请在低峰时段批量处理，并积极缓存描述。若质量可接受，可考虑更便宜的模型（在本例中如 GPT-4o-mini）用于描述生成。

• 查询时延很关键：携带图像的 GPT-4o 每次查询约需 5–10 秒。通过流式输出可让用户更快看到部分结果，缓存常见问题可加速重复查询。若 2 页上下文就够用，不要取 5 页，能明显减少 Token。

• 监控对的指标：跟踪检索时延（小集合应低于 100ms）、生成时延（带图像预计 5–10 秒）、单次查询 API 成本，以及如有收集则关注用户反馈。

• 错误处理能救场：总会出问题。要有降级方案，如下面的代码片段。

def answer_question_safe(query):
    """Wrapper with error handling."""
    try:
        return answer_question(query)
    except Exception as e:
        print(f"Error: {e}")
        # Fall back to text-only using descriptions
        results = retrieve(query)
        if results["documents"][0]:
            context = results["documents"][0][0][:500]
            return f"Based on the instructions: {context}...", []
        return "Sorry, couldn't process that question.", []技术支持

当您的流水线需要多步处理（搜索 → 验证 → 重搜 → 作答），Agentic RAG 模式能帮助您清晰地组织流程。

结语

多模态 RAG 将检索增强生成扩展到真实世界文档的完整内容形态，包括图像、图示、表格与图表。 

我们成功地将其用于查询以视觉为主的宜家说明书，纯文本方法在这里并不可行。整条流水线除 API 调用外，其余均可在本地运行。

关键洞见是：您不需要复杂的多模态嵌入也能实现这一点。通过为视觉内容生成高质量的文字描述，您可以用标准的文本检索找到相关图片，然后让视觉-语言模型在生成阶段发力。

如果您想在此基础上继续深入，欢迎从AI 工程师职业路径开启您的旅程！