当前位置：首页 > news >正文

Lychee Rerank MM代码实例：与Elasticsearch结合构建多模态混合检索系统

news 2026/8/31 20:03:14

Lychee Rerank MM代码实例：与Elasticsearch结合构建多模态混合检索系统

1. 项目概述与核心价值

在现代信息检索场景中，用户往往需要通过多种形式的内容来寻找所需信息——可能是用文字描述需求，也可能是直接上传一张图片。传统的文本检索系统难以处理这种多模态的搜索需求，而Lychee Rerank MM正是为了解决这个问题而生。

Lychee Rerank MM基于Qwen2.5-VL多模态大模型构建，能够理解文字和图片之间的深层语义关联。通过与Elasticsearch结合，我们可以构建一个既能处理海量数据检索，又能精准理解多模态语义的混合检索系统。

这种组合的优势很明显：Elasticsearch负责快速从百万级文档中初步筛选候选结果，Lychee Rerank MM则对这些候选进行精细的重排序，确保最终返回的结果真正符合用户的搜索意图。

2. 环境准备与依赖安装

开始之前，我们需要准备好Python环境和必要的依赖包。建议使用Python 3.10或更高版本。

# 创建虚拟环境 python -m venv lychee_es_env source lychee_es_env/bin/activate # Linux/Mac # 或者 lychee_es_env\Scripts\activate # Windows # 安装核心依赖 pip install lychee-rerank-mm elasticsearch7 transformers torch

除了Python包，我们还需要确保有足够的硬件资源。Qwen2.5-VL模型需要约16-20GB的显存，建议使用A100、RTX 3090或更高性能的GPU。如果没有GPU，也可以使用CPU模式，但推理速度会慢很多。

3. Elasticsearch多模态数据索引构建

要让Elasticsearch支持多模态检索，我们需要精心设计索引结构。以下是一个包含文本和图像特征的多模态文档索引示例：

from elasticsearch import Elasticsearch from elasticsearch.helpers import bulk # 连接Elasticsearch es = Elasticsearch(["http://localhost:9200"]) # 创建多模态索引 index_mapping = { "mappings": { "properties": { "id": {"type": "keyword"}, "title": {"type": "text", "analyzer": "ik_max_word"}, "content": {"type": "text", "analyzer": "ik_max_word"}, "image_path": {"type": "keyword"}, "text_embedding": { "type": "dense_vector", "dims": 1024, "index": True, "similarity": "cosine" }, "image_tags": {"type": "keyword"}, "created_time": {"type": "date"} } } } # 创建索引 es.indices.create(index="multimodal_docs", body=index_mapping)

在实际应用中，我们还需要将文档内容向量化后存入Elasticsearch。这里使用一个简单的文本嵌入模型示例：

from sentence_transformers import SentenceTransformer # 加载文本嵌入模型 text_encoder = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2') def index_documents(docs): """索引多模态文档到Elasticsearch""" actions = [] for doc in docs: # 生成文本嵌入 text_embedding = text_encoder.encode(doc['content']).tolist() action = { "_index": "multimodal_docs", "_id": doc['id'], "_source": { "id": doc['id'], "title": doc['title'], "content": doc['content'], "image_path": doc.get('image_path', ''), "text_embedding": text_embedding, "image_tags": doc.get('image_tags', []), "created_time": doc['created_time'] } } actions.append(action) # 批量导入 bulk(es, actions)

4. Lychee Rerank MM集成与初始化

现在我们来初始化Lychee Rerank MM模型，并创建与Elasticsearch的集成类：

import torch from lychee_rerank_mm import LycheeRerankMM from PIL import Image import base64 from io import BytesIO class MultimodalSearchSystem: def __init__(self): # 初始化重排序模型 self.rerank_model = LycheeRerankMM( model_name_or_path="Qwen/Qwen2.5-VL-7B-Instruct", device="cuda" if torch.cuda.is_available() else "cpu", use_flash_attention2=True ) # 连接Elasticsearch self.es = Elasticsearch(["http://localhost:9200"]) def encode_query(self, query): """处理多模态查询""" if isinstance(query, str): # 文本查询 return {"text": query, "type": "text"} elif hasattr(query, 'read'): # 文件对象或图片 image = Image.open(query).convert('RGB') return {"image": image, "type": "image"} else: raise ValueError("不支持的查询类型") def search_elasticsearch(self, query, top_k=100): """在Elasticsearch中进行初步检索""" query_vector = text_encoder.encode(query).tolist() search_body = { "query": { "script_score": { "query": {"match_all": {}}, "script": { "source": "cosineSimilarity(params.query_vector, 'text_embedding') + 1.0", "params": {"query_vector": query_vector} } } }, "size": top_k } response = self.es.search( index="multimodal_docs", body=search_body ) return [hit["_source"] for hit in response["hits"]["hits"]]

5. 多模态检索完整流程实现

有了基础组件后，我们来实现完整的多模态检索流程：

def multimodal_search(self, query, top_k=10): """多模态混合检索完整流程""" # 1. 解析查询 parsed_query = self.encode_query(query) # 2. Elasticsearch初步检索 if parsed_query["type"] == "text": initial_results = self.search_elasticsearch(parsed_query["text"], top_k=100) else: # 对于图像查询，我们可以使用标签或元数据检索 initial_results = self.search_by_image_metadata(parsed_query["image"]) # 3. 准备重排序数据 rerank_inputs = [] for doc in initial_results: if parsed_query["type"] == "text": # 文本查询 vs 文本文档 rerank_inputs.append({ "query": parsed_query["text"], "document": doc["content"] }) else: # 图像查询 vs 文本文档（或包含图像的文档） rerank_inputs.append({ "query_image": parsed_query["image"], "document": doc["content"] }) # 4. 重排序 scores = self.rerank_model.rerank_batch(rerank_inputs) # 5. 组合结果 ranked_results = [] for score, doc in zip(scores, initial_results): ranked_results.append({ "doc": doc, "score": score, "rerank_score": score }) # 按重排序分数排序 ranked_results.sort(key=lambda x: x["rerank_score"], reverse=True) return ranked_results[:top_k] def search_by_image_metadata(self, image, top_k=100): """基于图像标签或元数据检索""" # 这里可以集成图像识别模型来生成标签 # 简化示例：返回所有文档 search_body = { "query": {"match_all": {}}, "size": top_k } response = self.es.search( index="multimodal_docs", body=search_body ) return [hit["_source"] for hit in response["hits"]["hits"]]

6. 实际应用示例与效果展示

让我们通过几个具体例子来看看这个系统在实际场景中如何工作：

# 初始化系统 search_system = MultimodalSearchSystem() # 示例1：文本搜索 - 寻找关于人工智能的文档 text_query = "人工智能的最新发展" results = search_system.multimodal_search(text_query, top_k=5) print("文本搜索结果:") for i, result in enumerate(results): print(f"{i+1}. {result['doc']['title']} (得分: {result['score']:.3f})") # 示例2：图像搜索 - 通过产品图片寻找相关文档 with open("product_image.jpg", "rb") as img_file: image_results = search_system.multimodal_search(img_file, top_k=5) print("\n图像搜索结果:") for i, result in enumerate(image_results): print(f"{i+1}. {result['doc']['title']} (得分: {result['score']:.3f})")

在实际测试中，我们发现这种混合检索方式相比纯文本检索有显著提升。特别是在处理跨模态查询时（比如用文字搜索图片内容，或者用图片搜索相关文档），Lychee Rerank MM的深度语义理解能力发挥了关键作用。

7. 性能优化与实践建议

构建生产级的多模态检索系统时，还需要考虑一些优化策略：

def optimize_performance(self): """性能优化措施""" # 1. 批量处理优化 self.rerank_model.set_batch_size(8) # 根据GPU内存调整 # 2. 缓存常用查询结果 self.setup_caching() # 3. 异步处理支持 self.setup_async_processing() def setup_caching(self): """设置查询缓存""" from cachetools import TTLCache self.query_cache = TTLCache(maxsize=1000, ttl=300) # 5分钟缓存 def setup_async_processing(self): """设置异步处理""" import asyncio from concurrent.futures import ThreadPoolExecutor self.executor = ThreadPoolExecutor(max_workers=4) self.loop = asyncio.get_event_loop() async def async_multimodal_search(self, query, top_k=10): """异步多模态搜索""" return await self.loop.run_in_executor( self.executor, self.multimodal_search, query, top_k )

另外还有一些实用建议：