STEP3-VL-10B实战教程：对接RAG系统——图文混合检索+上下文增强回答

1. 为什么需要图文混合检索？

想象一下这个场景：你是一家电商公司的运营，每天要处理海量的商品图片和描述文档。当你想找“去年夏天那款蓝色连衣裙的促销文案”时，传统的关键词搜索只能找到文字描述，却找不到对应的商品图片。或者反过来，你看到一张产品图，想知道它的详细规格和用户评价，却需要手动去翻找文档。

这就是传统RAG（检索增强生成）系统的痛点——它只能处理文字，对图片无能为力。而STEP3-VL-10B的出现，正好解决了这个问题。

STEP3-VL-10B是阶跃星辰开源的一个10B参数多模态模型，别看它参数不算特别大，但在视觉理解、文字识别、复杂推理这些能力上，表现相当出色。更重要的是，它支持OpenAI兼容的API，这意味着我们可以很方便地把它集成到现有的RAG系统中。

今天我就带你一步步搭建一个既能理解图片又能理解文字的智能检索系统。学完这篇教程，你就能让现有的RAG系统“长眼睛”，实现真正的图文混合检索和智能回答。

2. 准备工作：环境与数据

2.1 确保STEP3-VL-10B服务正常运行

首先，你需要确保STEP3-VL-10B的服务已经启动并可以访问。根据你提供的配置，服务应该已经在运行了。

检查服务状态很简单，打开终端执行：

supervisorctl status

你会看到类似这样的输出：

webui                           RUNNING   pid 12345, uptime 1:23:45

如果服务没有运行，用这个命令启动：

supervisorctl start webui

服务启动后，你的API地址通常是这样的格式：

https://你的服务器地址-7860.web.gpu.csdn.net/api/v1/chat/completions

记下这个地址，后面会用到。

2.2 准备你的图文数据

接下来要准备一些测试数据。我建议从简单的开始，比如：

1. 产品文档：包含产品描述、规格参数、使用说明的PDF或Word文档
2. 产品图片：对应的产品图片，最好有文件名能对应到文档
3. 用户评价：一些用户反馈的文字记录

为了方便演示，我准备了一个简单的例子。创建一个文件夹，里面放这些文件：

data/
├── documents/
│   ├── product_guide.pdf    # 产品说明书
│   ├── spec_sheet.docx      # 规格参数表
│   └── user_reviews.txt     # 用户评价
├── images/
│   ├── product_front.jpg    # 产品正面图
│   ├── product_side.jpg     # 产品侧面图
│   └── product_in_use.jpg   # 产品使用场景图
└── metadata.json           # 元数据文件，记录图文对应关系

metadata.json文件内容可以这样写：

{
  "product_guide.pdf": {
    "related_images": ["product_front.jpg", "product_in_use.jpg"],
    "category": "documentation",
    "description": "产品使用说明书"
  },
  "product_front.jpg": {
    "related_docs": ["product_guide.pdf", "spec_sheet.docx"],
    "description": "产品正面外观图"
  }
}

这个文件很重要，它建立了文字和图片之间的关联关系。

3. 构建图文混合向量数据库

传统的RAG系统只处理文字，我们要改造它，让它能同时处理图片和文字。

3.1 文字向量化处理

对于文字部分，我们可以用现有的文本嵌入模型。这里我推荐使用text-embedding-ada-002或者开源的BGE模型。

安装必要的库：

pip install langchain chromadb pypdf python-docx pillow

然后创建文字向量数据库：

from langchain.document_loaders import PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
import os

# 加载文档
def load_documents(doc_path):
    documents = []
    
    for file in os.listdir(doc_path):
        file_path = os.path.join(doc_path, file)
        
        if file.endswith('.pdf'):
            loader = PyPDFLoader(file_path)
        elif file.endswith('.txt'):
            loader = TextLoader(file_path)
        elif file.endswith('.docx'):
            # 对于docx文件，可以用其他库处理
            continue
        else:
            continue
            
        docs = loader.load()
        for doc in docs:
            doc.metadata["source"] = file
            doc.metadata["type"] = "text"
        documents.extend(docs)
    
    return documents

# 分割文本
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)

# 加载并处理文档
documents = load_documents("data/documents/")
texts = text_splitter.split_documents(documents)

# 创建向量数据库
embeddings = OpenAIEmbeddings()  # 或者使用其他嵌入模型
vectorstore = Chroma.from_documents(
    documents=texts,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

3.2 图片向量化处理

这是关键的一步。我们要用STEP3-VL-10B来理解图片内容，然后把理解的结果转换成文字描述，再用文本嵌入模型向量化。

import base64
import requests
from PIL import Image
import io

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

def describe_image_with_step3(image_path, api_url):
    """使用STEP3-VL-10B描述图片内容"""
    
    # 获取base64编码的图片
    base64_image = encode_image(image_path)
    
    # 准备请求数据
    headers = {
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": "Step3-VL-10B",
        "messages": [
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{base64_image}"
                        }
                    },
                    {
                        "type": "text",
                        "text": "请详细描述这张图片的内容，包括其中的文字、物体、场景、颜色等所有可见信息。描述要尽可能详细和准确。"
                    }
                ]
            }
        ],
        "max_tokens": 1024
    }
    
    try:
        response = requests.post(api_url, headers=headers, json=payload)
        response.raise_for_status()
        result = response.json()
        return result['choices'][0]['message']['content']
    except Exception as e:
        print(f"图片描述失败: {e}")
        return None

# 处理所有图片
def process_images(image_folder, api_url):
    image_descriptions = []
    
    for image_file in os.listdir(image_folder):
        if image_file.lower().endswith(('.png', '.jpg', '.jpeg')):
            image_path = os.path.join(image_folder, image_file)
            print(f"正在处理图片: {image_file}")
            
            description = describe_image_with_step3(image_path, api_url)
            if description:
                # 创建文档对象
                from langchain.schema import Document
                doc = Document(
                    page_content=description,
                    metadata={
                        "source": image_file,
                        "type": "image",
                        "original_path": image_path
                    }
                )
                image_descriptions.append(doc)
    
    return image_descriptions

# 你的STEP3-VL-10B API地址
api_url = "https://你的服务器地址-7860.web.gpu.csdn.net/api/v1/chat/completions"

# 处理图片并获取描述
image_docs = process_images("data/images/", api_url)

# 将图片描述也添加到向量数据库
if image_docs:
    vectorstore.add_documents(image_docs)

3.3 建立图文关联索引

现在我们有文字向量和图片描述向量，但还需要建立它们之间的关联关系。

import json

def build_cross_reference_index(metadata_file):
    """建立图文交叉引用索引"""
    with open(metadata_file, 'r', encoding='utf-8') as f:
        metadata = json.load(f)
    
    # 创建两个方向的索引
    doc_to_images = {}
    image_to_docs = {}
    
    for doc, info in metadata.items():
        if "related_images" in info:
            doc_to_images[doc] = info["related_images"]
            for img in info["related_images"]:
                if img not in image_to_docs:
                    image_to_docs[img] = []
                image_to_docs[img].append(doc)
    
    return doc_to_images, image_to_docs

# 加载元数据建立索引
doc_to_images, image_to_docs = build_cross_reference_index("data/metadata.json")

# 保存索引供后续使用
import pickle
with open("cross_reference.pkl", "wb") as f:
    pickle.dump({
        "doc_to_images": doc_to_images,
        "image_to_docs": image_to_docs
    }, f)

4. 实现混合检索与智能回答

有了向量数据库和交叉索引，现在我们可以实现真正的图文混合检索了。

4.1 混合检索器实现

from typing import List, Dict, Any
import numpy as np

class HybridRetriever:
    def __init__(self, vectorstore, cross_ref_path, step3_api_url):
        self.vectorstore = vectorstore
        self.step3_api_url = step3_api_url
        
        # 加载交叉引用索引
        with open(cross_ref_path, "rb") as f:
            cross_ref = pickle.load(f)
            self.doc_to_images = cross_ref["doc_to_images"]
            self.image_to_docs = cross_ref["image_to_docs"]
    
    def retrieve(self, query: str, k: int = 5):
        """混合检索：先检索文字，再根据关联找到图片"""
        
        # 1. 文字检索
        text_results = self.vectorstore.similarity_search(query, k=k)
        
        # 2. 根据文字结果找到关联图片
        related_images = []
        for doc in text_results:
            source = doc.metadata.get("source")
            if source in self.doc_to_images:
                related_images.extend(self.doc_to_images[source])
        
        # 去重
        related_images = list(set(related_images))
        
        # 3. 如果有图片，获取图片的详细描述
        image_docs = []
        for img in related_images[:3]:  # 最多取3张相关图片
            # 在向量库中查找图片描述
            img_query = f"图片 {img} 的描述"
            img_results = self.vectorstore.similarity_search(img_query, k=1)
            if img_results:
                image_docs.append(img_results[0])
        
        return {
            "text_results": text_results,
            "image_results": image_docs,
            "related_images": related_images
        }
    
    def retrieve_with_image(self, image_path: str, query: str = None, k: int = 5):
        """基于图片的检索"""
        
        # 1. 先用STEP3-VL-10B理解图片
        image_description = describe_image_with_step3(image_path, self.step3_api_url)
        
        if not image_description:
            return {"error": "图片理解失败"}
        
        # 2. 如果用户提供了额外查询，合并查询
        if query:
            combined_query = f"{query}。图片内容：{image_description}"
        else:
            combined_query = image_description
        
        # 3. 检索相关文档
        text_results = self.vectorstore.similarity_search(combined_query, k=k)
        
        # 4. 找到图片对应的文档
        image_file = os.path.basename(image_path)
        related_docs = self.image_to_docs.get(image_file, [])
        
        return {
            "image_description": image_description,
            "text_results": text_results,
            "related_documents": related_docs
        }

4.2 上下文增强回答生成

现在到了最精彩的部分——用STEP3-VL-10B生成基于上下文的智能回答。

class ContextEnhancedAnswerer:
    def __init__(self, retriever, step3_api_url):
        self.retriever = retriever
        self.step3_api_url = step3_api_url
    
    def build_context(self, retrieval_results):
        """构建上下文提示"""
        
        context_parts = []
        
        # 添加文字上下文
        if retrieval_results.get("text_results"):
            context_parts.append("## 相关文档内容：")
            for i, doc in enumerate(retrieval_results["text_results"][:3], 1):
                context_parts.append(f"{i}. {doc.page_content[:500]}...")
        
        # 添加图片上下文
        if retrieval_results.get("image_results"):
            context_parts.append("\n## 相关图片描述：")
            for i, img_doc in enumerate(retrieval_results["image_results"][:2], 1):
                context_parts.append(f"{i}. {img_doc.page_content[:300]}...")
        
        return "\n\n".join(context_parts)
    
    def generate_answer(self, query: str, use_images: bool = True):
        """生成增强回答"""
        
        # 1. 检索相关上下文
        if use_images:
            results = self.retriever.retrieve(query)
        else:
            # 纯文字检索
            results = {"text_results": self.retriever.vectorstore.similarity_search(query, k=5)}
        
        # 2. 构建上下文
        context = self.build_context(results)
        
        # 3. 准备STEP3-VL-10B的请求
        messages = [
            {
                "role": "system",
                "content": "你是一个专业的助手，能够根据提供的文档和图片信息回答问题。请基于以下上下文信息，准确、详细地回答用户的问题。如果上下文信息不足，请如实说明。"
            },
            {
                "role": "user",
                "content": f"上下文信息：\n{context}\n\n用户问题：{query}\n\n请基于以上上下文回答问题："
            }
        ]
        
        # 4. 如果有相关图片，把图片也加进去
        if use_images and results.get("related_images"):
            # 这里简化处理，实际可以发送图片或图片描述
            messages[1]["content"] += f"\n\n相关图片：{', '.join(results['related_images'][:2])}"
        
        # 5. 调用STEP3-VL-10B
        payload = {
            "model": "Step3-VL-10B",
            "messages": messages,
            "max_tokens": 1024,
            "temperature": 0.7
        }
        
        try:
            response = requests.post(self.step3_api_url, json=payload)
            response.raise_for_status()
            result = response.json()
            answer = result['choices'][0]['message']['content']
            
            return {
                "answer": answer,
                "sources": [doc.metadata.get("source", "未知") for doc in results.get("text_results", [])],
                "related_images": results.get("related_images", [])
            }
        except Exception as e:
            return {"error": f"生成回答失败: {str(e)}"}
    
    def answer_with_image(self, image_path: str, query: str = None):
        """基于图片回答问题"""
        
        # 检索与图片相关的内容
        results = self.retriever.retrieve_with_image(image_path, query)
        
        if "error" in results:
            return results
        
        # 构建基于图片的上下文
        context_parts = [f"## 图片描述：\n{results['image_description']}"]
        
        if results.get("text_results"):
            context_parts.append("\n## 相关文档内容：")
            for i, doc in enumerate(results["text_results"][:3], 1):
                context_parts.append(f"{i}. {doc.page_content[:500]}...")
        
        context = "\n\n".join(context_parts)
        
        # 准备问题
        if query:
            user_question = query
        else:
            user_question = "请根据图片和相关文档，详细描述和分析图片内容"
        
        # 调用STEP3-VL-10B
        messages = [
            {
                "role": "system",
                "content": "你是一个视觉分析专家，能够结合图片内容和相关文档进行综合分析。请基于提供的图片描述和相关文档，详细回答用户的问题。"
            },
            {
                "role": "user",
                "content": f"{context}\n\n问题：{user_question}\n\n请回答："
            }
        ]
        
        payload = {
            "model": "Step3-VL-10B",
            "messages": messages,
            "max_tokens": 1024
        }
        
        try:
            response = requests.post(self.step3_api_url, json=payload)
            response.raise_for_status()
            result = response.json()
            
            return {
                "answer": result['choices'][0]['message']['content'],
                "image_description": results['image_description'],
                "related_docs": results.get('related_documents', [])
            }
        except Exception as e:
            return {"error": str(e)}

5. 完整系统集成与使用示例

5.1 初始化系统

# 初始化整个系统
def initialize_system(step3_api_url, data_folder="./data"):
    """初始化图文混合RAG系统"""
    
    print("正在初始化系统...")
    
    # 1. 初始化向量数据库（如果不存在则创建）
    if not os.path.exists("./chroma_db"):
        print("创建向量数据库...")
        # 这里应该包含前面提到的文档加载和向量化代码
        # 为了简洁，这里省略具体实现
        pass
    
    # 加载现有的向量数据库
    embeddings = OpenAIEmbeddings()
    vectorstore = Chroma(
        persist_directory="./chroma_db",
        embedding_function=embeddings
    )
    
    # 2. 初始化检索器
    retriever = HybridRetriever(
        vectorstore=vectorstore,
        cross_ref_path="cross_reference.pkl",
        step3_api_url=step3_api_url
    )
    
    # 3. 初始化回答生成器
    answerer = ContextEnhancedAnswerer(retriever, step3_api_url)
    
    print("系统初始化完成！")
    return answerer

# 使用示例
api_url = "https://你的服务器地址-7860.web.gpu.csdn.net/api/v1/chat/completions"
answerer = initialize_system(api_url)

5.2 实际使用案例

现在让我们看看这个系统在实际场景中怎么用。

案例1：基于文字查询的混合检索

# 用户问关于产品的问题
query = "这款产品的最大特点是什么？有哪些用户反馈？"

result = answerer.generate_answer(query, use_images=True)

print("问题：", query)
print("\n回答：")
print(result["answer"])
print("\n参考来源：", result["sources"])
print("相关图片：", result["related_images"])

案例2：基于图片的智能分析

# 用户上传一张产品图片，想了解相关信息
image_path = "data/images/product_front.jpg"
query = "这个产品的主要功能是什么？适合什么人群使用？"

result = answerer.answer_with_image(image_path, query)

print("图片分析结果：")
print(result["answer"])
print("\n图片描述：", result["image_description"][:200], "...")
print("相关文档：", result["related_docs"])

案例3：纯文字检索（兼容传统RAG）

# 如果只想进行文字检索
query = "产品的技术规格参数"

result = answerer.generate_answer(query, use_images=False)
print(result["answer"])

5.3 Web API接口

为了让其他系统也能使用，我们可以提供一个简单的Web API：

from flask import Flask, request, jsonify
import os

app = Flask(__name__)

# 全局变量
answerer = None

@app.route('/api/query', methods=['POST'])
def handle_query():
    """处理文字查询"""
    data = request.json
    query = data.get('query', '')
    use_images = data.get('use_images', True)
    
    if not query:
        return jsonify({"error": "查询内容不能为空"}), 400
    
    try:
        result = answerer.generate_answer(query, use_images)
        return jsonify(result)
    except Exception as e:
        return jsonify({"error": str(e)}), 500

@app.route('/api/query_with_image', methods=['POST'])
def handle_image_query():
    """处理图片查询"""
    if 'image' not in request.files:
        return jsonify({"error": "没有上传图片"}), 400
    
    image_file = request.files['image']
    query = request.form.get('query', '')
    
    # 保存临时文件
    temp_path = f"/tmp/{image_file.filename}"
    image_file.save(temp_path)
    
    try:
        result = answerer.answer_with_image(temp_path, query)
        # 清理临时文件
        os.remove(temp_path)
        return jsonify(result)
    except Exception as e:
        if os.path.exists(temp_path):
            os.remove(temp_path)
        return jsonify({"error": str(e)}), 500

if __name__ == '__main__':
    # 初始化系统
    api_url = os.environ.get("STEP3_API_URL", "http://localhost:8000/v1/chat/completions")
    answerer = initialize_system(api_url)
    
    app.run(host='0.0.0.0', port=5000, debug=True)

启动这个API服务后，其他应用就可以通过HTTP请求来使用你的图文混合RAG系统了。

6. 优化建议与进阶技巧

6.1 性能优化

如果你的数据量很大，可以考虑以下优化：

1. 批量处理图片：不要一张一张处理，可以批量发送给STEP3-VL-10B
2. 缓存机制：对已经处理过的图片和文档，缓存它们的向量和描述
3. 异步处理：使用异步IO提高并发处理能力

import asyncio
import aiohttp

async def batch_describe_images(image_paths, api_url, batch_size=5):
    """批量描述图片"""
    async with aiohttp.ClientSession() as session:
        tasks = []
        for i in range(0, len(image_paths), batch_size):
            batch = image_paths[i:i+batch_size]
            task = asyncio.create_task(
                describe_images_batch(session, batch, api_url)
            )
            tasks.append(task)
        
        results = await asyncio.gather(*tasks)
        return results

6.2 精度提升

1. 多轮对话优化：对于复杂查询，可以使用多轮对话让STEP3-VL-10B逐步深入理解
2. 置信度评分：为检索结果添加置信度评分，只使用高置信度的内容
3. 来源验证：让模型在回答时注明信息来源，提高可信度

6.3 扩展功能

1. 多模态搜索：支持同时搜索文字和图片
2. 语义聚类：对检索结果进行聚类，提供更组织的答案
3. 实时更新：支持增量更新向量数据库
4. 权限控制：根据不同用户权限返回不同内容

7. 总结

通过这篇教程，我们完成了一个完整的图文混合RAG系统搭建。让我简单总结一下关键点：

核心价值：这个系统最大的价值在于打破了文字和图片之间的壁垒。现在你的RAG系统不仅能理解文档内容，还能“看懂”图片，并且能把两者关联起来，提供更全面、更准确的回答。

技术要点回顾：

1. 用STEP3-VL-10B理解图片内容，生成文字描述
2. 将图片描述和文档内容一起向量化存储
3. 建立图文交叉引用索引，实现混合检索
4. 基于检索到的上下文，用STEP3-VL-10B生成智能回答

实际应用场景：

• 电商客服：用户发来商品图片，系统能自动找到商品信息和用户评价
• 知识管理：企业内部的图文资料可以统一检索
• 内容审核：同时检查文字和图片内容是否合规
• 教育培训：根据教材图片自动关联相关知识点

开始你的实践：

1. 先从简单的数据开始，比如你自己的文档和图片
2. 逐步增加数据量，观察系统表现
3. 根据实际需求调整检索策略和提示词
4. 考虑加入更多优化，比如缓存、批处理等

这个系统的美妙之处在于它的灵活性。你可以根据自己的需求调整各个环节——用不同的向量模型、调整检索策略、优化提示词等等。STEP3-VL-10B的强大视觉理解能力为这个系统提供了坚实的基础，而OpenAI兼容的API让集成变得异常简单。

现在，你的RAG系统终于“长眼睛”了。去试试看，它能为你带来什么样的惊喜吧！

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。