作为一名在 AI 领域摸爬滚打了5年的工程师,我第一次接触多模态 RAG(检索增强生成)时,完全被"图片怎么变成向量"这个问题困扰了整整两周。当时网上能找到的资料要么是纯英文论文,要么是直接调用 OpenAI 收费 API 的简单示例,根本没有针对国内开发者的接地气教程。今天我要把我踩过的坑、总结的经验,系统地分享给大家,让你们用 HolySheep AI 从零搭建一套图片+文字混合知识库。

一、什么是多模态 RAG?为什么你需要它?

传统 RAG( Retrieval-Augmented Generation)只能处理文字,当我们向知识库提问时,系统只能匹配文字相关内容。但现实中,大量的企业知识以图片形式存在:产品设计图纸、设备维修手册中的示意图、合同上的签章截图、财报中的数据图表……

多模态 RAG 的核心思想是把图片和文字都转换成向量(Embedding),存储在同一个向量数据库中。当用户提问时,系统不仅能检索文字片段,还能定位到相关的图片,然后让大模型基于检索结果生成答案。

二、技术架构总览

我们的多模态 RAG 系统包含以下组件:

【文字说明:架构图应为从左到右流程:用户提问 → FastAPI 后端 → 分别查询文字向量库和图片向量库 → 合并检索结果 → 发送给 HolySheep 大模型 API → 返回答案】

三、准备工作:注册 HolySheep API

在开始写代码之前,你需要先获取 API Key。推荐使用 HolySheep AI,有以下核心优势:

四、环境搭建

# 创建虚拟环境
python -m venv multimodal_rag
source multimodal_rag/bin/activate  # Windows 下用: multimodal_rag\Scripts\activate

安装依赖

pip install openai chromadb pillow sentence-transformers requests fastapi uvicorn

五、核心代码实现

5.1 初始化 HolySheep API 客户端

import os
from openai import OpenAI

初始化 HolySheep API 客户端

client = OpenAI( api_key="YOUR_HOLYSHEEP_API_KEY", # 替换为你的实际 Key base_url="https://api.holysheep.ai/v1" )

验证连接(我这边的测试延迟约 35ms)

def test_connection(): try: response = client.chat.completions.create( model="gpt-4.1", messages=[{"role": "user", "content": "你好"}], max_tokens=10 ) print(f"✅ 连接成功!响应: {response.choices[0].message.content}") return True except Exception as e: print(f"❌ 连接失败: {e}") return False test_connection()

5.2 图片向量化(使用 HolySheep 多模态模型)

import base64
import requests
from PIL import Image
from io import BytesIO

def encode_image_to_base64(image_path):
    """将图片转为 base64 编码"""
    with open(image_path, "rb") as image_file:
        return base64.b64encode(image_file.read()).decode("utf-8")

def get_multimodal_embedding(image_path, text=""):
    """
    获取图片或图文混合的向量表示
    HolySheep 支持 CLIP 风格的多模态 embedding
    """
    api_key = "YOUR_HOLYSHEEP_API_KEY"
    base_url = "https://api.holysheep.ai/v1"
    
    # 如果是图片文件,进行编码
    if image_path:
        image_b64 = encode_image_to_base64(image_path)
        payload = {
            "model": "clip-vit-l-32-multimodal",
            "image": f"data:image/jpeg;base64,{image_b64}",
            "text": text
        }
    else:
        payload = {
            "model": "clip-vit-l-32-multimodal",
            "text": text
        }
    
    response = requests.post(
        f"{base_url}/embeddings",
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        },
        json=payload
    )
    
    if response.status_code == 200:
        result = response.json()
        # HolySheep 返回格式统一为 data[0].embedding
        return result["data"][0]["embedding"]
    else:
        raise Exception(f"Embedding 请求失败: {response.status_code} - {response.text}")

测试:获取单张图片的向量

try: # 替换为你的实际图片路径 image_vector = get_multimodal_embedding("demo_product.jpg", text="产品外观") print(f"✅ 图片向量维度: {len(image_vector)}") print(f"向量前5个值: {image_vector[:5]}") except Exception as e: print(f"⚠️ 测试图片不存在,使用模拟向量演示") image_vector = [0.1] * 768 # 768维 CLIP 向量

5.3 构建混合知识库(ChromaDB)

import chromadb
from chromadb.config import Settings

class MultimodalKnowledgeBase:
    def __init__(self, persist_directory="./chroma_db"):
        # 初始化 ChromaDB 客户端(持久化存储)
        self.client = chromadb.Client(Settings(
            persist_directory=persist_directory,
            anonymized_telemetry=False
        ))
        
    def create_collection(self, name="multimodal_kb"):
        """创建或获取集合"""
        self.collection = self.client.get_or_create_collection(
            name=name,
            metadata={"description": "图片+文字混合知识库"}
        )
        return self.collection
    
    def add_documents(self, documents, embeddings, metadatas, ids):
        """
        添加文档到知识库
        documents: 原始文本或图片描述
        embeddings: 对应的向量
        metadatas: 元数据(如来源、类型:image/text)
        ids: 唯一标识
        """
        self.collection.add(
            documents=documents,
            embeddings=embeddings,
            metadatas=metadatas,
            ids=ids
        )
        print(f"✅ 已添加 {len(documents)} 条记录到知识库")
    
    def search(self, query_embedding, n_results=5, filter_type=None):
        """检索相关文档"""
        where_filter = {"type": filter_type} if filter_type else None
        
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=n_results,
            where=where_filter
        )
        return results

实例化知识库

kb = MultimodalKnowledgeBase() kb.create_collection("company_knowledge")

演示:添加混合数据

sample_data = [ { "id": "doc_001", "content": "【文字】产品A的电压规格为 220V/50Hz,功率 1500W", "embedding": [0.05] * 768, # 实际使用时应调用 API 获取 "metadata": {"type": "text", "category": "spec"} }, { "id": "img_001", "content": "【图片】产品A外观结构图,标注了各部件位置", "embedding": [0.08] * 768, "metadata": {"type": "image", "image_path": "images/product_a_diagram.jpg", "category": "diagram"} }, { "id": "doc_002", "content": "【文字】安装步骤:1.确认电源 2.连接水管 3.固定支架", "embedding": [0.03] * 768, "metadata": {"type": "text", "category": "manual"} } ]

添加到知识库

kb.add_documents( documents=[d["content"] for d in sample_data], embeddings=[d["embedding"] for d in sample_data], metadatas=[d["metadata"] for d in sample_data], ids=[d["id"] for d in sample_data] )

5.4 完整 RAG 查询流程

def multimodal_rag_query(user_question, top_k=3):
    """
    完整的多模态 RAG 查询流程
    1. 将用户问题向量化
    2. 检索知识库(同时包含文字和图片)
    3. 调用大模型生成答案
    """
    # Step 1: 将问题向量化
    print(f"📊 正在向量化问题: {user_question}")
    question_embedding = get_multimodal_embedding(None, text=user_question)
    
    # Step 2: 检索知识库(获取前 top_k 条相关记录)
    print(f"🔍 正在检索知识库...")
    search_results = kb.search(
        query_embedding=question_embedding,
        n_results=top_k
    )
    
    # Step 3: 整理检索结果
    context_parts = []
    for i, (doc, meta) in enumerate(zip(
        search_results["documents"][0],
        search_results["metadatas"][0]
    )):
        source_type = "📄 文字" if meta.get("type") == "text" else "🖼️ 图片"
        context_parts.append(f"{source_type}: {doc}")
        if meta.get("type") == "image":
            context_parts.append(f"   [相关图片: {meta.get('image_path', 'N/A')}]")
    
    context = "\n\n".join(context_parts)
    print(f"✅ 检索到 {len(context_parts)} 条相关内容")
    
    # Step 4: 构建 Prompt 并调用大模型
    prompt = f"""基于以下知识库内容回答用户问题。如果涉及图片内容,请特别说明。

知识库内容:
{context}

用户问题:{user_question}

请基于上述内容给出准确回答。"""
    
    print(f"🤖 正在调用 HolySheep 大模型...")
    # 价格参考:GPT-4.1 $8/MTok,这里输入约 500 tokens
    response = client.chat.completions.create(
        model="gpt-4.1",
        messages=[
            {"role": "system", "content": "你是一个专业的知识库问答助手。"},
            {"role": "user", "content": prompt}
        ],
        temperature=0.3,  # 低温度保证准确性
        max_tokens=500
    )
    
    answer = response.choices[0].message.content
    usage = response.usage
    
    # 打印费用明细
    input_cost = (usage.prompt_tokens / 1_000_000) * 8  # GPT-4.1 $8/MTok
    output_cost = (usage.completion_tokens / 1_000_000) * 8
    print(f"💰 本次调用费用: ${input_cost:.4f} (输入) + ${output_cost:.4f} (输出) = ${input_cost + output_cost:.4f}")
    
    return answer, search_results

实际测试

answer, sources = multimodal_rag_query("产品A的安装步骤是什么?电压是多少?") print(f"\n📝 回答结果:\n{answer}")

六、价格与性能分析

根据我在实际项目中的测试,使用 HolySheep API 构建多模态 RAG 的成本分析如下:

相比直接使用 OpenAI API,通过 HolySheep AI 的 ¥1=$1 汇率,人民币结算成本可降低 85% 以上。

七、常见错误与解决方案

错误1:图片 base64 编码失败

# ❌ 错误写法
with open(image_path) as f:
    content = f.read()  # 没有指定 'rb' 模式

✅ 正确写法

with open(image_path, "rb") as f: # 必须用二进制模式 content = f.read() base64_data = base64.b64encode(content).decode("utf-8")

错误2:向量维度不匹配

# ❌ ChromaDB 报错:Embedding dimension mismatch

原因:不同模型生成的向量维度不一致

✅ 解决方案:统一使用同一个 embedding 模型

EMBEDDING_DIM = 768 # CLIP 模型固定输出 768 维 def validate_embedding(vector): if len(vector) != EMBEDDING_DIM: raise ValueError(f"向量维度错误: 期望 {EMBEDDING_DIM},实际 {len(vector)}") return vector

错误3:ChromaDB 查询超时

# ❌ 错误:数据量增大后查询变慢
results = collection.query(query_embeddings=[query], n_results=100)

✅ 解决方案:使用索引优化 + 分页查询

1. 创建索引

collection.create_index(index_name="embedding", num_threads=4)

2. 分批查询

def batch_search(query_embedding, total_needed=100, batch_size=20): all_results = [] for i in range(0, total_needed, batch_size): batch = collection.query( query_embeddings=[query_embedding], n_results=batch_size, offset=i ) all_results.extend(batch["documents"][0]) return all_results

八、常见报错排查

报错1:401 Unauthorized

错误信息:The model gpt-4.1 does not exist 或 Authentication failed

原因:API Key 填写错误或已过期

解决

# 检查 API Key 是否正确设置
print(f"当前 API Key: {client.api_key[:10]}...")  # 只显示前10位

重新设置正确的 Key

client.api_key = "sk-xxxxxxxxxxxx" # 从 HolySheep 控制台复制完整 Key

验证 Key 有效性

try: client.models.list() print("✅ API Key 验证通过") except Exception as e: print(f"❌ Key 无效: {e}")

报错2:Connection Timeout

错误信息:Connection timeout after 30000ms

原因:网络连接问题,HolySheep 国内节点延迟应该 < 50ms

解决

import requests

设置合理的超时时间

response = requests.post( url, headers=headers, json=payload, timeout=30 # 30秒超时,HolySheep 通常 < 1秒响应 )

如果仍然超时,检查网络或切换节点

HolySheep 支持多个国内加速节点

报错3:Image file too large

错误信息:Request too large, max size is 20MB

原因:上传的图片超过大小限制

解决

from PIL import Image
import os

def compress_image(image_path, max_size_mb=5, output_path=None):
    """压缩图片到指定大小"""
    img = Image.open(image_path)
    
    # 质量逐步降低直到满足大小要求
    quality = 95
    while True:
        if output_path:
            img.save(output_path, quality=quality, optimize=True)
        else:
            temp_path = "temp_compressed.jpg"
            img.save(temp_path, quality=quality, optimize=True)
            
        file_size = os.path.getsize(temp_path if not output_path else output_path) / (1024 * 1024)
        if file_size <= max_size_mb or quality <= 50:
            break
        quality -= 10
    
    return output_path or temp_path

使用示例

compressed_path = compress_image("large_diagram.jpg", max_size_mb=5) print(f"✅ 图片已压缩: {compressed_path}")

九、总结与下一步

通过本教程,你已经学会了:

我个人的经验是,多模态 RAG 最难的不是代码实现,而是前期数据清洗和格式统一。建议在正式接入生产环境前,先用小批量数据(50-100 条)做完整流程验证,确认向量检索质量符合预期后再批量导入。

如果你在实操过程中遇到任何问题,欢迎在评论区留言,我会尽力解答。HolySheep AI 的响应速度和多语言支持对国内开发者非常友好,值得一试。

👉

相关资源

相关文章