跨模态搜索新范式：通义千问3-VL-Reranker-8B+LangChain构建智能问答系统
想象一下，你有一个包含大量产品手册、设计图纸和用户反馈图片的知识库。当用户问“帮我找找去年那款蓝色运动鞋的防水性能说明”时，传统的关键词搜索可能完全失效——因为文档里写的是“靛蓝色”，图片里也没有“防水”这个词，只有一张水滴测试的图表。
这就是跨模态搜索要解决的难题：让机器能像人一样，同时理解文字、图片、视频，并在它们之间建立联系。今天，我们就来聊聊如何用通义千问最新的多模态模型Qwen3-VL-Reranker-8B，结合LangChain框架，搭建一个真正能“看懂”图文内容的智能问答系统。
1. 为什么需要跨模态搜索？
先看几个实际场景：
电商客服场景：用户发来一张衣服照片问“这件有没有M码？”传统系统只能靠图片文件名或标签来猜，准确率很低。跨模态系统能直接看懂图片内容，匹配商品库。
医疗辅助场景：医生上传一张X光片，问“这个阴影可能是什么？”系统需要同时理解医学影像和文本病历，给出综合判断。
教育资料库：学生问“牛顿第二定律的公式推导过程”，系统需要从教科书扫描图、教学视频、讲义文本中找出相关内容。
这些场景的共同点是：信息散落在不同模态里，用户的问题可能指向任何形式的内容。传统基于文本的检索系统在这里完全不够用。
通义千问团队最近开源的Qwen3-VL-Embedding和Qwen3-VL-Reranker模型，正好解决了这个问题。它们能把文本、图像、视频都映射到同一个语义空间，让不同形式的内容可以相互比较、检索。
2. 系统架构设计：两阶段检索增强
我们的系统采用经典的RAG（检索增强生成）架构，但在检索环节做了重要升级：
用户问题 → 多模态编码 → 向量检索（初筛） → 重排序（精排） → 上下文构建 → 大模型生成答案关键创新在于中间的“向量检索+重排序”两阶段流程。让我用大白话解释一下：
第一阶段：快速初筛想象你在图书馆找书。先根据书名关键词（向量检索）快速找出50本可能相关的书。Qwen3-VL-Embedding模型就是这个“快速找书员”，它能把问题和文档都转换成向量，然后计算相似度。
第二阶段：精细排序但这50本书里，哪些真正最相关？这时候需要“专业图书管理员”仔细翻看每本书的内容。Qwen3-VL-Reranker模型就是这个管理员，它对每个候选结果进行深度分析，重新打分排序。
为什么要两阶段？因为如果直接用重排序模型处理海量文档，速度太慢。先用Embedding快速缩小范围，再用Reranker精细挑选，兼顾了速度和精度。
3. 环境搭建与模型部署
3.1 基础环境准备
首先确保你的Python环境在3.8以上，然后安装必要的包：
pip install langchain langchain-community chromadb pydantic pip install transformers torch pip install pillow requests对于多模态模型，我们主要需要两个：
# 模型导入 from transformers import AutoModel, AutoTokenizer import torch # Embedding模型（用于快速检索） embedding_model_name = "Qwen/Qwen3-VL-Embedding-8B" # Reranker模型（用于精细排序） reranker_model_name = "Qwen/Qwen3-VL-Reranker-8B"如果你的显存有限（比如只有16GB），可以考虑使用2B版本，或者启用量化：
# 使用量化版本节省显存 model = AutoModel.from_pretrained( "Qwen/Qwen3-VL-Embedding-2B", torch_dtype=torch.float16, device_map="auto" )3.2 LangChain多模态文档加载
LangChain提供了丰富的文档加载器，支持各种格式：
from langchain.document_loaders import ( TextLoader, PyPDFLoader, UnstructuredImageLoader, CSVLoader ) from langchain.schema import Document import os class MultiModalDocumentLoader: """统一的多模态文档加载器""" def __init__(self, folder_path): self.folder_path = folder_path def load_documents(self): documents = [] # 遍历文件夹中的所有文件 for filename in os.listdir(self.folder_path): file_path = os.path.join(self.folder_path, filename) try: if filename.endswith('.txt'): # 加载文本文件 loader = TextLoader(file_path) docs = loader.load() for doc in docs: doc.metadata["type"] = "text" doc.metadata["filename"] = filename documents.extend(docs) elif filename.endswith('.pdf'): # 加载PDF文件（可能包含图片） loader = PyPDFLoader(file_path) docs = loader.load() for doc in docs: doc.metadata["type"] = "pdf" doc.metadata["filename"] = filename documents.extend(docs) elif filename.lower().endswith(('.png', '.jpg', '.jpeg')): # 加载图片文件 loader = UnstructuredImageLoader(file_path) docs = loader.load() for doc in docs: doc.metadata["type"] = "image" doc.metadata["filename"] = filename # 存储图片路径供后续处理 doc.metadata["image_path"] = file_path documents.extend(docs) elif filename.endswith('.csv'): # 加载CSV文件 loader = CSVLoader(file_path) docs = loader.load() for doc in docs: doc.metadata["type"] = "table" doc.metadata["filename"] = filename documents.extend(docs) except Exception as e: print(f"加载文件 {filename} 时出错: {e}") continue return documents这个加载器能处理文本、PDF、图片、表格等多种格式，并为每个文档添加类型标签，方便后续处理。
4. 核心实现：混合检索策略
4.1 多模态向量化存储
传统的文本向量数据库只能处理文字，我们需要扩展它来支持多模态内容：
from langchain.vectorstores import Chroma from langchain.embeddings.base import Embeddings import numpy as np from PIL import Image import base64 from io import BytesIO class QwenVLEmbeddings(Embeddings): """适配Qwen3-VL-Embedding的LangChain Embeddings接口""" def __init__(self, model_name="Qwen/Qwen3-VL-Embedding-2B"): from transformers import AutoModel, AutoTokenizer self.tokenizer = AutoTokenizer.from_pretrained(model_name) self.model = AutoModel.from_pretrained( model_name, torch_dtype=torch.float16, device_map="auto" ) self.model.eval() def _encode_text(self, text): """编码文本""" inputs = self.tokenizer(text, return_tensors="pt", padding=True, truncation=True) with torch.no_grad(): outputs = self.model(**inputs.to(self.model.device)) # 取[EOS] token的隐藏状态作为向量 embeddings = outputs.last_hidden_state[:, -1, :] return embeddings.cpu().numpy() def _encode_image(self, image_path): """编码图片""" # 将图片转换为base64格式（Qwen-VL支持的格式） with open(image_path, "rb") as img_file: img_data = img_file.read() img_base64 = base64.b64encode(img_data).decode('utf-8') # 构建多模态输入 multimodal_input = f"<image>{img_base64}</image>" return self._encode_text(multimodal_input) def embed_documents(self, texts, metadatas=None): """批量编码文档""" embeddings = [] for i, text in enumerate(texts): metadata = metadatas[i] if metadatas else {} if metadata.get("type") == "image" and "image_path" in metadata: # 如果是图片，使用图片编码 emb = self._encode_image(metadata["image_path"]) else: # 否则使用文本编码 emb = self._encode_text(text) embeddings.append(emb[0]) # 去掉batch维度 return embeddings def embed_query(self, text): """编码查询""" return self._encode_text(text)[0]这个类实现了LangChain的标准Embeddings接口，但内部能根据文档类型选择不同的编码方式。图片文档会被特殊处理，提取视觉特征。
4.2 构建向量数据库
有了编码器，我们就可以创建向量数据库了：
def create_vector_store(documents, persist_directory="./chroma_db"): """创建并持久化向量数据库""" # 初始化嵌入模型 embeddings = QwenVLEmbeddings() # 提取文档内容和元数据 texts = [doc.page_content for doc in documents] metadatas = [doc.metadata for doc in documents] # 创建向量存储 vector_store = Chroma.from_texts( texts=texts, embedding=embeddings, metadatas=metadatas, persist_directory=persist_directory, collection_name="multimodal_docs" ) # 持久化到磁盘 vector_store.persist() print(f"向量数据库已创建，包含 {len(documents)} 个文档") return vector_store4.3 重排序优化检索结果
向量检索找到的top-k结果可能不够精准，这时候就需要重排序模型出场：
class QwenVLReranker: """Qwen3-VL-Reranker重排序器""" def __init__(self, model_name="Qwen/Qwen3-VL-Reranker-2B"): from transformers import AutoModelForSequenceClassification, AutoTokenizer self.tokenizer = AutoTokenizer.from_pretrained(model_name) self.model = AutoModelForSequenceClassification.from_pretrained( model_name, torch_dtype=torch.float16, device_map="auto" ) self.model.eval() def rerank(self, query, candidates, top_n=5): """对候选结果进行重排序""" scores = [] for candidate in candidates: # 构建重排序输入 text = candidate.page_content metadata = candidate.metadata # 根据文档类型构建不同的输入 if metadata.get("type") == "image": # 图片文档：查询 vs 图片 with open(metadata["image_path"], "rb") as f: img_base64 = base64.b64encode(f.read()).decode('utf-8') document_input = f"<image>{img_base64}</image>" else: # 文本文档：查询 vs 文本 document_input = text # 构建模型输入 input_text = f"<|im_start|>system\n判断文档是否与查询相关。只回答'是'或'否'。\n<|im_end|>\n<|im_start|>user\n查询: {query}\n文档: {document_input}\n<|im_end|>" # 计算相关性分数 inputs = self.tokenizer(input_text, return_tensors="pt", truncation=True, max_length=4096) with torch.no_grad(): outputs = self.model(**inputs.to(self.model.device)) # 获取"是"的概率作为相关性分数 logits = outputs.logits yes_score = torch.softmax(logits, dim=-1)[0, self.tokenizer.convert_tokens_to_ids("是")].item() scores.append(yes_score) # 按分数排序 sorted_indices = np.argsort(scores)[::-1] # 降序 reranked_candidates = [candidates[i] for i in sorted_indices[:top_n]] reranked_scores = [scores[i] for i in sorted_indices[:top_n]] return reranked_candidates, reranked_scores重排序模型的核心思想是：对每个查询-文档对进行深度分析，计算一个精细的相关性分数，而不是简单的向量相似度。
5. 完整问答系统集成
5.1 检索增强生成流程
现在我们把所有组件组装起来：
from langchain.chains import RetrievalQA from langchain.llms import HuggingFacePipeline from transformers import pipeline class MultimodalQASystem: """多模态问答系统""" def __init__(self, vector_store_path="./chroma_db"): # 加载向量数据库 self.embeddings = QwenVLEmbeddings() self.vector_store = Chroma( persist_directory=vector_store_path, embedding_function=self.embeddings, collection_name="multimodal_docs" ) # 初始化重排序器 self.reranker = QwenVLReranker() # 初始化大语言模型（用于生成最终答案） self.llm = self._init_llm() # 检索器配置 self.retriever = self.vector_store.as_retriever( search_kwargs={"k": 20} # 初筛取20个结果 ) def _init_llm(self): """初始化大语言模型""" # 这里可以使用任何你喜欢的LLM，比如Qwen、ChatGLM等 from transformers import AutoModelForCausalLM, AutoTokenizer, TextStreamer model_name = "Qwen/Qwen2.5-7B-Instruct" tokenizer = AutoTokenizer.from_pretrained(model_name) model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=torch.float16, device_map="auto" ) # 创建文本生成pipeline text_generation_pipeline = pipeline( "text-generation", model=model, tokenizer=tokenizer, max_new_tokens=512, temperature=0.7, do_sample=True ) return HuggingFacePipeline(pipeline=text_generation_pipeline) def ask(self, question, use_reranker=True): """回答用户问题""" # 第一步：向量检索（初筛） print(" 正在进行向量检索...") initial_results = self.retriever.get_relevant_documents(question) if not initial_results: return "抱歉，没有找到相关信息。" # 第二步：重排序（可选） if use_reranker and len(initial_results) > 1: print(" 正在进行重排序优化...") final_results, scores = self.reranker.rerank(question, initial_results, top_n=5) # 打印重排序结果 print("\n重排序结果：") for i, (doc, score) in enumerate(zip(final_results, scores)): print(f"{i+1}. [{doc.metadata.get('type', 'unknown')}] {doc.metadata.get('filename', 'unknown')}") print(f" 相关性分数: {score:.4f}") print(f" 内容摘要: {doc.page_content[:100]}...") print() else: final_results = initial_results[:5] scores = [1.0] * len(final_results) # 第三步：构建上下文 context = self._build_context(final_results, question) # 第四步：生成答案 print("💭 正在生成答案...") prompt = self._build_prompt(question, context) answer = self.llm(prompt) return answer, final_results, scores def _build_context(self, documents, question): """从检索结果构建上下文""" context_parts = [] for i, doc in enumerate(documents): doc_type = doc.metadata.get("type", "text") filename = doc.metadata.get("filename", "unknown") if doc_type == "image": context_part = f"[图片文档 {i+1}: {filename}]\n内容描述: {doc.page_content}\n" elif doc_type == "table": context_part = f"[表格文档 {i+1}: {filename}]\n表格内容: {doc.page_content}\n" else: context_part = f"[文本文档 {i+1}: {filename}]\n{doc.page_content}\n" context_parts.append(context_part) return "\n".join(context_parts) def _build_prompt(self, question, context): """构建提示词""" prompt_template = """你是一个智能助手，请根据提供的上下文信息回答用户的问题。 上下文信息： {context} 用户问题：{question} 请根据上下文信息给出准确、详细的回答。如果上下文信息不足以回答问题，请如实说明。 回答：""" return prompt_template.format(context=context, question=question)5.2 实际使用示例
让我们看一个完整的例子：
def main(): # 1. 准备文档（假设我们有一个包含多种格式文档的文件夹） data_folder = "./knowledge_base" # 2. 加载文档 print(" 正在加载文档...") loader = MultiModalDocumentLoader(data_folder) documents = loader.load_documents() print(f"已加载 {len(documents)} 个文档") # 3. 创建向量数据库（首次运行需要，之后可以跳过） if not os.path.exists("./chroma_db"): print("🛠 正在创建向量数据库...") vector_store = create_vector_store(documents) else: print(" 向量数据库已存在，直接加载") # 4. 初始化问答系统 print(" 初始化问答系统...") qa_system = MultimodalQASystem() # 5. 开始问答 while True: print("\n" + "="*50) question = input("\n请输入您的问题（输入'退出'结束）：") if question.lower() in ['退出', 'exit', 'quit']: break # 回答问题 answer, sources, scores = qa_system.ask(question) print("\n" + "="*50) print(" 答案：") print(answer) print("\n 参考来源：") for i, (source, score) in enumerate(zip(sources, scores)): print(f"{i+1}. {source.metadata.get('filename')} ({source.metadata.get('type')}) - 相关性: {score:.3f}") if __name__ == "__main__": main()6. 效果对比与优化建议
6.1 重排序带来的提升
在实际测试中，重排序能显著提升检索质量。比如对于问题“展示一下我们产品的用户界面设计”：
不使用重排序：可能返回一些包含“用户”、“界面”等关键词的文本文档，但可能不是最新的设计图。
使用重排序后：系统能理解“展示”意味着需要视觉内容，优先返回产品UI截图、设计原型图等图片文档。
6.2 性能优化技巧
- 批量处理：对大量文档进行编码时，使用批量处理可以提高效率：
def batch_encode_documents(documents, batch_size=4): """批量编码文档""" embeddings = [] for i in range(0, len(documents), batch_size): batch = documents[i:i+batch_size] # 批量编码逻辑 ... return embeddings- 缓存机制：对已编码的文档进行缓存，避免重复计算：
import hashlib import pickle class CachedEmbeddings(QwenVLEmbeddings): def __init__(self, cache_file="./embeddings_cache.pkl"): super().__init__() self.cache_file = cache_file self.cache = self._load_cache() def _get_cache_key(self, content, doc_type): """生成缓存键""" key_str = f"{doc_type}:{content[:100]}" return hashlib.md5(key_str.encode()).hexdigest() def embed_documents(self, texts, metadatas=None): embeddings = [] uncached_indices = [] uncached_texts = [] uncached_metadatas = [] for i, (text, metadata) in enumerate(zip(texts, metadatas or [{}])): cache_key = self._get_cache_key( text if metadata.get("type") != "image" else metadata.get("image_path", ""), metadata.get("type", "text") ) if cache_key in self.cache: embeddings.append(self.cache[cache_key]) else: embeddings.append(None) uncached_indices.append(i) uncached_texts.append(text) uncached_metadatas.append(metadata) # 编码未缓存的文档 if uncached_texts: new_embeddings = super().embed_documents(uncached_texts, uncached_metadatas) # 更新缓存和结果 for idx, emb, text, metadata in zip(uncached_indices, new_embeddings, uncached_texts, uncached_metadatas): cache_key = self._get_cache_key( text if metadata.get("type") != "image" else metadata.get("image_path", ""), metadata.get("type", "text") ) self.cache[cache_key] = emb embeddings[idx] = emb # 保存缓存 self._save_cache() return embeddings- 混合检索策略：对于某些查询，可以结合关键词检索和向量检索：
def hybrid_retrieval(query, vector_store, keyword_weight=0.3): """混合检索：向量相似度 + 关键词匹配""" # 向量检索结果 vector_results = vector_store.similarity_search_with_score(query, k=10) # 简单关键词匹配（示例） keyword_scores = {} for doc in vector_store.get()["documents"]: # 计算关键词匹配分数 score = sum(1 for word in query.split() if word.lower() in doc.lower()) keyword_scores[doc] = score / len(query.split()) # 合并分数 final_results = [] for doc, vector_score in vector_results: keyword_score = keyword_scores.get(doc.page_content, 0) combined_score = (1 - keyword_weight) * vector_score + keyword_weight * keyword_score final_results.append((doc, combined_score)) # 按合并分数排序 final_results.sort(key=lambda x: x[1], reverse=True) return final_results7. 总结
用下来感觉这套方案确实能解决很多实际问题。Qwen3-VL-Reranker-8B的重排序效果比我想象的要好，特别是对于图文混合的内容，它能真正理解不同模态之间的语义联系。
LangChain的框架让整个系统搭建起来比较顺畅，各种文档加载器和向量数据库的集成都做得不错。不过在实际部署时，需要注意显存占用问题——8B模型对硬件要求不低，如果资源有限，2B版本也是不错的选择，效果虽然略有下降，但运行效率高很多。
如果你正在构建需要处理多种格式文档的知识库系统，或者想要让现有的问答系统支持图片、表格等内容，这套方案值得一试。建议先从简单的场景开始，比如处理产品手册、技术文档这类结构化程度较高的内容，等跑通了再扩展到更复杂的场景。
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
