Lychee Rerank MM代码实例调用batch rerank接口处理100文档的完整Python脚本1. 项目概述与环境准备Lychee Rerank MM是一个基于Qwen2.5-VL构建的高性能多模态重排序系统专门解决多模态检索场景中的精准语义匹配问题。这个系统由哈工大深圳自然语言处理团队开发能够处理文本-文本、图像-文本、文本-图像以及图文-图文的全模态重排序任务。1.1 系统核心优势多模态深度对齐支持多种模态组合的匹配任务基于先进模型使用8B参数级别的Qwen2.5-VL多模态大模型双模式操作支持单条分析和批量重排序工程优化包含Flash Attention 2加速、显存清理和模型缓存机制1.2 环境要求与安装在开始编写批量处理脚本前需要确保环境满足以下要求# 基础环境要求 Python版本: 3.10 GPU显存: 16GB-20GB建议A10、A100或RTX 3090以上显卡 CUDA版本: 11.8或更高2. 批量重排序接口详解2.1 接口基本信息Lychee Rerank MM提供了RESTful API接口用于批量重排序任务。主要接口信息如下接口地址:http://localhost:8080/api/batch_rerank请求方法: POSTContent-Type: application/json超时设置: 建议设置为300-600秒根据文档数量调整2.2 请求参数结构{ query: 搜索查询内容或图片URL, documents: [ 文档1内容, 文档2内容, # ...更多文档 ], instruction: 重排序指令可选, batch_size: 8 # 批处理大小可选 }2.3 响应数据结构成功调用后接口返回JSON格式的响应{ status: success, results: [ { document: 文档内容, score: 0.85, rank: 1 }, # ...更多排序结果 ], processing_time: 12.45 }3. 完整Python批量处理脚本下面是一个完整的Python脚本用于处理100文档的批量重排序任务。3.1 脚本基础结构#!/usr/bin/env python3 # -*- coding: utf-8 -*- import requests import json import time import logging from typing import List, Dict, Optional from concurrent.futures import ThreadPoolExecutor, as_completed # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) logger logging.getLogger(__name__) class LycheeBatchReranker: Lychee Rerank MM批量处理客户端 def __init__(self, base_url: str http://localhost:8080): self.base_url base_url.rstrip(/) self.session requests.Session() self.session.timeout 300 # 5分钟超时 def health_check(self) - bool: 检查服务是否健康 try: response self.session.get(f{self.base_url}/health, timeout10) return response.status_code 200 except Exception as e: logger.error(f健康检查失败: {e}) return False3.2 核心批量处理函数def batch_rerank( self, query: str, documents: List[str], instruction: str Given a web search query, retrieve relevant passages that answer the query., batch_size: int 8, max_retries: int 3 ) - Optional[List[Dict]]: 执行批量重排序 Args: query: 查询内容文本或图片URL documents: 文档列表 instruction: 重排序指令 batch_size: 批处理大小 max_retries: 最大重试次数 Returns: 排序后的结果列表或None失败时 # 准备请求数据 payload { query: query, documents: documents, instruction: instruction, batch_size: batch_size } # 重试机制 for attempt in range(max_retries): try: logger.info(f开始批量重排序文档数量: {len(documents)}) start_time time.time() response self.session.post( f{self.base_url}/api/batch_rerank, jsonpayload, timeout300 ) if response.status_code 200: result response.json() processing_time time.time() - start_time logger.info( f重排序完成处理时间: {processing_time:.2f}秒, f文档数量: {len(documents)} ) return result.get(results, []) else: logger.warning( f请求失败 (尝试 {attempt 1}/{max_retries}): f状态码 {response.status_code} ) except requests.exceptions.Timeout: logger.warning(f请求超时 (尝试 {attempt 1}/{max_retries})) except requests.exceptions.ConnectionError: logger.warning(f连接错误 (尝试 {attempt 1}/{max_retries})) except Exception as e: logger.error(f未知错误: {e}) # 指数退避重试 if attempt max_retries - 1: wait_time 2 ** attempt logger.info(f等待 {wait_time}秒后重试...) time.sleep(wait_time) logger.error(所有重试尝试均失败) return None3.3 大文档集分块处理对于超过100个文档的大规模处理建议使用分块处理策略def process_large_document_set( self, query: str, all_documents: List[str], chunk_size: int 50, **kwargs ) - List[Dict]: 处理大规模文档集分块处理 Args: query: 查询内容 all_documents: 所有文档列表 chunk_size: 每块处理的文档数量 **kwargs: 其他传递给batch_rerank的参数 Returns: 合并后的排序结果 all_results [] total_docs len(all_documents) logger.info(f开始处理大规模文档集总数: {total_docs}) # 分块处理 for i in range(0, total_docs, chunk_size): chunk all_documents[i:i chunk_size] logger.info(f处理块 {i//chunk_size 1}/{(total_docs-1)//chunk_size 1}) chunk_results self.batch_rerank(query, chunk, **kwargs) if chunk_results: all_results.extend(chunk_results) else: logger.error(f第 {i//chunk_size 1} 块处理失败) # 对所有结果进行最终排序 sorted_results sorted( all_results, keylambda x: x.get(score, 0), reverseTrue ) # 重新分配排名 for rank, result in enumerate(sorted_results, 1): result[rank] rank logger.info(f大规模文档处理完成总共处理: {len(sorted_results)}个文档) return sorted_results3.4 结果保存与导出def save_results_to_file( self, results: List[Dict], output_file: str, format: str json ): 将结果保存到文件 Args: results: 重排序结果 output_file: 输出文件路径 format: 输出格式json/csv/txt try: if format.lower() json: with open(output_file, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) elif format.lower() csv: import csv with open(output_file, w, encodingutf-8, newline) as f: writer csv.writer(f) writer.writerow([Rank, Score, Document]) for result in results: writer.writerow([ result.get(rank, ), result.get(score, ), result.get(document, )[:500] # 限制长度 ]) elif format.lower() txt: with open(output_file, w, encodingutf-8) as f: for result in results: f.write( fRank: {result.get(rank, )}, fScore: {result.get(score, ):.4f}\n fDocument: {result.get(document, )[:200]}...\n f{-*80}\n ) logger.info(f结果已保存到: {output_file} ({format.upper()}格式)) except Exception as e: logger.error(f保存结果到文件失败: {e})4. 完整使用示例4.1 基础使用示例# 示例基础批量重排序 def example_basic_usage(): # 初始化客户端 reranker LycheeBatchReranker(http://localhost:8080) # 检查服务状态 if not reranker.health_check(): logger.error(Lychee Rerank服务不可用) return # 准备测试数据 query 人工智能在医疗领域的应用 documents [ 人工智能正在改变医疗诊断的方式..., 机器学习算法可以帮助分析医学影像..., # ...这里可以添加100个文档 深度学习在药物发现中的应用越来越广泛... ] # 执行批量重排序 results reranker.batch_rerank( queryquery, documentsdocuments, batch_size8 ) if results: # 显示前10个结果 for result in results[:10]: print(fRank {result[rank]}: Score {result[score]:.4f}) print(fDocument: {result[document][:100]}...) print() # 保存结果 reranker.save_results_to_file(results, rerank_results.json, json)4.2 处理1000文档的实战示例# 示例处理超大规模文档集 def example_large_scale_processing(): reranker LycheeBatchReranker() # 生成模拟的大规模文档集实际使用时替换为真实数据 large_document_set [ f这是第{i}个文档内容关于人工智能和机器学习... for i in range(1000) ] query 深度学习框架比较 # 使用分块处理大规模文档 results reranker.process_large_document_set( queryquery, all_documentslarge_document_set, chunk_size50, # 每次处理50个文档 batch_size8, max_retries3 ) if results: # 分析结果 top_scores [r[score] for r in results[:10]] avg_score sum(r[score] for r in results) / len(results) print(f处理完成总共处理 {len(results)} 个文档) print(f平均相关性得分: {avg_score:.4f}) print(f前10名得分范围: {min(top_scores):.4f} - {max(top_scores):.4f}) # 保存详细结果 reranker.save_results_to_file(results, large_scale_results.csv, csv)4.3 错误处理与性能监控# 高级功能添加性能监控和详细日志 def example_advanced_usage(): reranker LycheeBatchReranker() # 添加性能监控装饰器 def monitor_performance(func): def wrapper(*args, **kwargs): start_time time.time() result func(*args, **kwargs) end_time time.time() logger.info( f函数 {func.__name__} 执行时间: {end_time - start_time:.2f}秒 ) return result return wrapper # 监控批处理函数 monitored_rerank monitor_performance(reranker.batch_rerank) # 使用监控后的函数 documents [f测试文档{i} for i in range(100)] results monitored_rerank( query测试查询, documentsdocuments ) return results5. 最佳实践与优化建议5.1 性能优化策略根据实际使用经验以下策略可以显著提升批量处理效率合适的批处理大小根据GPU显存调整batch_size参数通常4-16之间连接复用使用Session对象保持HTTP连接减少连接建立开销超时设置根据文档数量合理设置超时时间错误重试实现指数退避重试机制提高鲁棒性5.2 内存管理建议# 内存优化示例 def memory_optimized_processing(reranker, query, document_path): 内存优化的处理方式适用于极大文档集 results [] batch_size 50 # 流式读取和处理文档 with open(document_path, r, encodingutf-8) as f: current_batch [] for line in f: current_batch.append(line.strip()) if len(current_batch) batch_size: batch_results reranker.batch_rerank( queryquery, documentscurrent_batch ) if batch_results: results.extend(batch_results) # 清空当前批次释放内存 current_batch [] # 处理最后一批 if current_batch: batch_results reranker.batch_rerank( queryquery, documentscurrent_batch ) if batch_results: results.extend(batch_results) return sorted(results, keylambda x: x[score], reverseTrue)5.3 生产环境部署建议对于生产环境的使用建议服务高可用部署多个Lychee Rerank实例使用负载均衡监控告警实现服务健康检查和使用量监控速率限制根据硬件能力实施适当的API调用限制日志记录完善的操作日志和错误日志记录6. 总结通过本文提供的完整Python脚本你可以轻松地使用Lychee Rerank MM系统处理100甚至1000文档的批量重排序任务。脚本包含了错误处理、性能优化、内存管理等生产环境需要的功能。主要功能特点完整的批量处理支持大规模文档集的重排序健壮的错误处理包含重试机制和异常处理灵活的配置选项可调整批处理大小、超时设置等参数多种输出格式支持JSON、CSV、TXT等多种结果格式性能监控内置执行时间监控和日志记录在实际使用中建议根据具体的硬件配置和业务需求调整参数设置特别是batch_size和chunk_size参数这些参数会显著影响处理性能和内存使用。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。