问题背景与适用场景

在企业知识库、智能客服、设计资产检索等场景中,传统 RAG 系统只能处理纯文本,无法理解和检索图片内容。Multimodal RAG 通过整合视觉-语言模型,实现跨模态的语义检索能力,让用户可以通过文字描述找到相关图片,或通过图片找到关联文档。本教程详细讲解如何基于 Python 构建支持图片检索的 Multimodal RAG 系统,并提供 2026 年最新的工程实践。

前置条件

配置步骤详解

步骤 1:安装核心依赖库

首先安装支持多模态处理的基础包,包括向量数据库客户端、图像处理库和 API 调用框架。

步骤 2:配置 HolySheep API 环境

设置 base_url 为 HolySheheep 官方端点,配置 API Key 用于认证和计费。

步骤 3:初始化向量数据库

创建集合并配置向量维度,确保与嵌入模型输出维度匹配(通常为 1024 或 1536)。

步骤 4:加载和预处理图片

使用 Pillow 读取图片,进行尺寸归一化和格式转换,准备输入给 Vision Model。

步骤 5:生成多模态向量

调用 Vision Model 生成图片描述和向量表示,将文本-图片配对存入向量数据库。


import os
import base64
from io import BytesIO
from PIL import Image
import requests
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

HolySheep API 配置

HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" HOLYSHEEP_BASE_URL = "https://api.holysheep.ai"

初始化 Qdrant 向量数据库

qdrant_client = QdrantClient(host="localhost", port=6333) collection_name = "multimodal_rag_2026"

创建集合(向量维度 1024,对应 CLIP/Vision Model)

def create_collection(): try: qdrant_client.recreate_collection( collection_name=collection_name, vectors_config=VectorParams(size=1024, distance=Distance.COSINE) ) print(f"Collection '{collection_name}' created successfully") except Exception as e: print(f"Collection may already exist: {e}")

编码图片为 base64

def encode_image(image_path: str) -> str: with open(image_path, "rb") as img_file: return base64.b64encode(img_file.read()).decode("utf-8")

调用 HolySheep Vision API 生成图片向量

def get_image_embedding(image_path: str) -> list: image_base64 = encode_image(image_path) payload = { "model": "clip-vit-large-patch14", "input": { "image": image_base64, "task": "image_embedding" } } headers = { "Authorization": f"Bearer {HOLYSHEEP_API_KEY}", "Content-Type": "application/json" } response = requests.post( f"{HOLYSHEEP_BASE_URL}/v1/embeddings", headers=headers, json=payload ) if response.status_code == 200: return response.json()["data"][0]["embedding"] else: raise Exception(f"API Error: {response.status_code} - {response.text}")

批量索引图片到向量数据库

def index_images(image_folder: str): image_files = [f for f in os.listdir(image_folder) if f.endswith(('.png', '.jpg', '.jpeg'))] for idx, img_file in enumerate(image_files): img_path = os.path.join(image_folder, img_file) print(f"Processing: {img_file}") embedding = get_image_embedding(img_path) point = PointStruct( id=idx, vector=embedding, payload={ "filename": img_file, "path": img_path, "type": "image" } ) qdrant_client.upsert( collection_name=collection_name, points=[point] ) print(f"Indexed: {img_file}") if __name__ == "__main__": create_collection() index_images("./test_images/")

完整代码示例

以下是通过自然语言查询检索相关图片的完整流程,包括 RAG 检索和上下文增强生成:


安装依赖(命令行安装方式)

pip install qdrant-client pillow requests openai langchain

测试 HolySheep API 连接性

curl -X POST "https://api.holysheep.ai/v1/embeddings" \ -H "Authorization: Bearer YOUR_HOLYSHEEP_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "clip-vit-large-patch14", "input": { "image": "base64_encoded_image_data", "task": "image_embedding" } }'

启动 Qdrant Docker 服务

docker run -d --name qdrant \ -p 6333:6333 \ -p 6334:6334 \ qdrant/qdrant

运行图片索引脚本

python multimodal_rag_indexer.py --folder ./product_images --batch-size 10

执行向量检索查询

python multimodal_rag_retriever.py --query "红色运动鞋侧面照" --top-k 5

检索和生成完整实现

import requests from qdrant_client import QdrantClient from qdrant_client.models import Filter, FieldCondition, MatchText HOLYSHEEP_BASE_URL = "https://api.holysheep.ai" HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" qdrant_client = QdrantClient(host="localhost", port=6333) def text_to_vector(query: str) -> list: """将文本查询转换为向量""" payload = { "model": "text-embedding-3-large", "input": query } headers = { "Authorization": f"Bearer {HOLYSHEEP_API_KEY}", "Content-Type": "application/json" } response = requests.post( f"{HOLYSHEEP_BASE_URL}/v1/embeddings", headers=headers, json=payload ) return response.json()["data"][0]["embedding"] def search_similar_images(query: str, top_k: int = 5): """语义检索相似图片""" query_vector = text_to_vector(query) search_results = qdrant_client.search( collection