Lychee-Rerank-MM部署教程：Python API封装与RESTful接口调用示例

Lychee-Rerank-MM部署教程：Python API封装与RESTful接口调用示例

1. 项目概述

Lychee-Rerank-MM是一个基于Qwen2.5-VL的多模态重排序模型，专门用于图文检索场景的精排任务。这个模型能够同时处理文本和图像输入，为搜索、推荐系统提供精准的相关性评分。

核心特性：

• 支持文本到文本、文本到图像、图像到文本、图像到图像四种检索模式
• 提供单文档和批量处理两种工作方式
• 指令感知能力，可根据不同场景优化性能
• 高性能推理，支持Flash Attention 2加速

技术规格：

• 参数规模：7B（实际8.29B）
• 推理精度：BF16
• 服务端口：7860

2. 环境准备与快速部署

2.1 系统要求

在开始部署前，请确保你的环境满足以下要求：

硬件要求：

• GPU显存：建议16GB以上（如NVIDIA RTX 4090、A100等）
• 系统内存：建议32GB以上
• 存储空间：至少30GB可用空间

软件依赖：

• Python 3.8或更高版本
• PyTorch 2.0或更高版本
• CUDA 11.7或更高版本

2.2 模型下载与准备

模型文件需要放置在指定路径：

# 创建模型目录 mkdir -p /root/ai-models/vec-ai/lychee-rerank-mm # 下载模型文件（请从ModelScope获取） # 将下载的模型文件放置到上述目录

2.3 一键部署启动

Lychee-Rerank-MM提供了多种启动方式：

# 进入项目目录 cd /root/lychee-rerank-mm # 方式1：使用启动脚本（推荐） ./start.sh # 方式2：直接运行Python应用 python app.py # 方式3：后台运行并记录日志 nohup python app.py > /tmp/lychee_server.log 2>&1 &

服务启动后，可以通过以下地址访问：

• 本地访问：http://localhost:7860
• 远程访问：http://<你的服务器IP>:7860

3. Python API封装实战

3.1 基础客户端封装

下面是一个完整的Python客户端封装示例，让你能够轻松集成Lychee-Rerank-MM到自己的项目中：

import requests import json import base64 from typing import List, Union, Dict, Optional class LycheeRerankClient: def __init__(self, base_url: str = "http://localhost:7860"): self.base_url = base_url.rstrip('/') self.session = requests.Session() def _encode_image(self, image_path: str) -> str: """将图像文件编码为base64字符串""" with open(image_path, "rb") as image_file: return base64.b64encode(image_file.read()).decode('utf-8') def single_rerank( self, instruction: str, query: Union[str, Dict], document: Union[str, Dict], timeout: int = 30 ) -> float: """ 单文档重排序 参数: instruction: 指令文本 query: 查询内容，可以是文本或图像字典 document: 文档内容，可以是文本或图像字典 timeout: 请求超时时间（秒） 返回: float: 相关性得分（0-1） """ # 构建请求数据 payload = { "instruction": instruction, "query": query, "document": document } try: response = self.session.post( f"{self.base_url}/rerank/single", json=payload, timeout=timeout ) response.raise_for_status() result = response.json() return result.get('score', 0.0) except requests.exceptions.RequestException as e: print(f"请求失败: {e}") return 0.0 def batch_rerank( self, instruction: str, query: Union[str, Dict], documents: List[Union[str, Dict]], timeout: int = 60 ) -> List[Dict]: """ 批量文档重排序 参数: instruction: 指令文本 query: 查询内容 documents: 文档列表 timeout: 请求超时时间（秒） 返回: List[Dict]: 排序后的文档列表，包含得分和排名 """ # 构建请求数据 payload = { "instruction": instruction, "query": query, "documents": documents } try: response = self.session.post( f"{self.base_url}/rerank/batch", json=payload, timeout=timeout ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f"批量请求失败: {e}") return [] def health_check(self) -> bool: """检查服务是否健康""" try: response = self.session.get(f"{self.base_url}/health", timeout=5) return response.status_code == 200 except: return False # 使用示例 if __name__ == "__main__": # 初始化客户端 client = LycheeRerankClient("http://localhost:7860") # 健康检查 if client.health_check(): print("服务正常运行") else: print("服务不可用")

3.2 多模态输入处理工具

为了更方便地处理文本和图像输入，我们可以创建一个工具类：

class MultiModalUtils: @staticmethod def create_text_input(text: str) -> Dict: """创建文本输入""" return {"type": "text", "content": text} @staticmethod def create_image_input(image_path: str) -> Dict: """创建图像输入""" return { "type": "image", "content": LycheeRerankClient()._encode_image(image_path) } @staticmethod def create_mixed_input(text: str, image_path: str = None) -> Dict: """创建图文混合输入""" if image_path: return { "type": "mixed", "text": text, "image": LycheeRerankClient()._encode_image(image_path) } return {"type": "text", "content": text} # 使用示例 utils = MultiModalUtils() # 创建不同类型的输入 text_query = utils.create_text_input("可爱的猫咪") image_doc = utils.create_image_input("/path/to/cat.jpg") mixed_query = utils.create_mixed_input("查找类似图片", "/path/to/example.jpg")

4. RESTful接口调用详解

4.1 单文档重排序接口

接口地址：POST /rerank/single

请求格式：

{ "instruction": "Given a web search query, retrieve relevant passages that answer the query", "query": { "type": "text", "content": "What is the capital of China?" }, "document": { "type": "text", "content": "The capital of China is Beijing." } }

Python调用示例：

import requests import json def single_rerank_example(): url = "http://localhost:7860/rerank/single" # 构建请求数据 payload = { "instruction": "Given a web search query, retrieve relevant passages that answer the query", "query": {"type": "text", "content": "What is the capital of China?"}, "document": {"type": "text", "content": "The capital of China is Beijing."} } # 设置请求头 headers = { "Content-Type": "application/json", "Accept": "application/json" } try: response = requests.post(url, json=payload, headers=headers, timeout=30) response.raise_for_status() result = response.json() print(f"相关性得分: {result.get('score', 0)}") return result except requests.exceptions.RequestException as e: print(f"请求失败: {e}") return None # 执行示例 single_rerank_example()

4.2 批量重排序接口

接口地址：POST /rerank/batch

请求格式：

{ "instruction": "Given a web search query, retrieve relevant passages that answer the query", "query": { "type": "text", "content": "机器学习入门" }, "documents": [ {"type": "text", "content": "机器学习基础概念讲解..."}, {"type": "text", "content": "深度学习与机器学习的区别..."}, {"type": "text", "content": "Python机器学习库使用指南..."} ] }

Python调用示例：

def batch_rerank_example(): url = "http://localhost:7860/rerank/batch" # 构建请求数据 payload = { "instruction": "Given a web search query, retrieve relevant passages that answer the query", "query": {"type": "text", "content": "机器学习入门"}, "documents": [ {"type": "text", "content": "机器学习基础概念讲解，包括监督学习、无监督学习和强化学习..."}, {"type": "text", "content": "深度学习是机器学习的一个子领域，使用神经网络进行特征学习..."}, {"type": "text", "content": "Scikit-learn是Python中最流行的机器学习库之一，提供了各种算法实现..."} ] } headers = {"Content-Type": "application/json"} try: response = requests.post(url, json=payload, headers=headers, timeout=60) response.raise_for_status() results = response.json() print("批量重排序结果:") for i, item in enumerate(results, 1): print(f"{i}. 得分: {item['score']:.4f} - 内容: {item['content'][:50]}...") return results except requests.exceptions.RequestException as e: print(f"批量请求失败: {e}") return [] # 执行批量重排序 batch_rerank_example()

4.3 健康检查接口

接口地址：GET /health

响应格式：

{ "status": "healthy", "model_loaded": true, "timestamp": "2024-01-15T10:30:00Z" }

5. 实战应用案例

5.1 电商商品搜索重排序

def ecommerce_search_example(): """电商商品搜索重排序示例""" client = LycheeRerankClient() # 用户查询 query = "红色连衣裙夏季新款" # 候选商品（模拟数据） products = [ {"id": 1, "title": "红色雪纺连衣裙", "description": "夏季新款红色雪纺连衣裙，透气舒适"}, {"id": 2, "title": "蓝色牛仔裤", "description": "修身蓝色牛仔裤，百搭款式"}, {"id": 3, "title": "红色T恤", "description": "纯棉红色T恤，休闲款式"}, {"id": 4, "title": "红色碎花连衣裙", "description": "夏季新款红色碎花连衣裙，清新风格"} ] # 构建文档列表 documents = [ utils.create_text_input(f"{p['title']} - {p['description']}") for p in products ] # 使用商品推荐指令 instruction = "Given a product search query, retrieve relevant products that match the query" # 执行重排序 results = client.batch_rerank(instruction, query, documents) print("商品搜索重排序结果:") for i, result in enumerate(results, 1): product_id = products[i-1]['id'] print(f"{i}. 商品ID: {product_id}, 得分: {result['score']:.4f}") return results # 运行示例 ecommerce_search_example()

5.2 多模态内容检索

def multimodal_retrieval_example(): """多模态内容检索示例""" client = LycheeRerankClient() # 图像查询（需要实际图像路径） image_query = utils.create_image_input("/path/to/query_image.jpg") # 文本文档候选 text_documents = [ "这是一张风景照片，展示山水美景", "城市建筑景观，现代都市风光", "人物肖像照片，表情自然", "动物图片，可爱宠物" ] # 使用多模态检索指令 instruction = "Given an image query, retrieve relevant textual descriptions" # 执行重排序 results = client.batch_rerank( instruction, image_query, [utils.create_text_input(doc) for doc in text_documents] ) print("多模态检索结果:") for i, result in enumerate(results, 1): print(f"{i}. 得分: {result['score']:.4f} - 内容: {result['content'][:30]}...") return results

6. 性能优化与最佳实践

6.1 批量处理优化

对于大量文档的重排序，建议使用批量模式：

def optimized_batch_processing(documents, batch_size=10): """优化批量处理，避免单次请求过大""" client = LycheeRerankClient() all_results = [] for i in range(0, len(documents), batch_size): batch = documents[i:i+batch_size] try: results = client.batch_rerank( "Given a web search query, retrieve relevant passages that answer the query", "你的查询文本", batch ) all_results.extend(results) except Exception as e: print(f"处理批次 {i//batch_size + 1} 时出错: {e}") return all_results

6.2 错误处理与重试机制

import time from tenacity import retry, stop_after_attempt, wait_exponential class RobustLycheeClient(LycheeRerankClient): @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) def robust_single_rerank(self, instruction, query, document): """带重试机制的单文档重排序""" try: return self.single_rerank(instruction, query, document) except Exception as e: print(f"重排序失败: {e}") raise def safe_batch_rerank(self, instruction, query, documents, max_retries=3): """安全的批量重排序，带错误处理""" for attempt in range(max_retries): try: return self.batch_rerank(instruction, query, documents) except Exception as e: print(f"第 {attempt + 1} 次尝试失败: {e}") if attempt < max_retries - 1: time.sleep(2 ** attempt) # 指数退避 else: return []

7. 总结

通过本教程，你已经掌握了Lychee-Rerank-MM模型的完整部署和使用方法。这个强大的多模态重排序模型能够为你的搜索和推荐系统提供精准的相关性评估。

关键要点回顾：

1. 快速部署：通过简单的命令即可启动服务，支持多种运行方式
2. Python集成：提供了完整的客户端封装，方便集成到现有项目
3. RESTful API：标准的接口设计，支持单文档和批量处理
4. 多模态支持：能够处理文本、图像以及图文混合的输入
5. 实战应用：适用于电商搜索、内容检索等多种场景

最佳实践建议：

• 根据具体场景选择合适的指令模板
• 使用批量处理提高效率，特别是文档数量较多时
• 实现适当的错误处理和重试机制
• 监控服务健康状态，确保服务稳定性

现在你已经具备了使用Lychee-Rerank-MM的所有知识，可以开始在你的项目中集成这个强大的多模态重排序模型了。

获取更多AI镜像

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