# 通义千问3-Reranker-0.6B实操手册：批量文档重排序脚本（Python+requests） ## 1. 引言：为什么需要文档重排序？ 想象一下这样的场景：你有一个包含数百条文档的数据库，用户输入一个问题，系统返回了50个可能相关的文档。但用户真正需要的是最相关的前5个结果，而不是花时间在大量结果中筛选。 这就是文档重排序的价值所在。通义千问3-Reranker-0.6B就像一个智能的文档筛选器，它能从一堆候选文档中精准找出与查询最相关的那些，并按相关性从高到低排序。 本文将手把手教你如何使用Python和requests库，编写一个批量文档重排序的脚本，让你能够高效处理大量文档的排序需求。 ## 2. 环境准备与快速部署 ### 2.1 安装必要依赖 在开始编写脚本前，确保你的Python环境已经安装了必要的库： ```bash pip install requests tqdm numpy pandas ``` 这些库的作用分别是： - `requests`：用于发送HTTP请求到重排序服务 - `tqdm`：显示进度条，让批量处理更直观 - `numpy`和`pandas`：处理数据和结果分析 ### 2.2 启动重排序服务 确保重排序服务已经启动并运行在7860端口： ```bash cd /root/Qwen3-Reranker-0.6B ./start.sh ``` 服务启动后，你可以在浏览器访问 `http://localhost:7860` 来验证服务是否正常。 ## 3. 基础API调用理解 ### 3.1 单个请求的格式 让我们先看一个最简单的API调用示例： ```python import requests import json def single_rerank(query, documents, instruction=None, batch_size=8): """ 单次重排序请求 """ url = "http://localhost:7860/api/predict" payload = { "data": [ query, # 查询文本 "\n".join(documents), # 文档列表，用换行符分隔 instruction or "", # 可选的任务指令 batch_size # 批处理大小 ] } try: response = requests.post(url, json=payload) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f"请求失败: {e}") return None # 使用示例 query = "什么是机器学习？" documents = [ "机器学习是人工智能的一个分支，让计算机通过数据学习规律。", "Python是一种流行的编程语言，广泛用于数据科学。", "深度学习是机器学习的一个子领域，使用神经网络模型。" ] result = single_rerank(query, documents) print("排序结果:", result) ``` 这个基础函数展示了如何向重排序服务发送单个请求，并获取排序结果。 ## 4. 批量处理脚本实战 ### 4.1 完整的批量重排序脚本 现在让我们编写一个完整的批量处理脚本： ```python import requests import json import time from tqdm import tqdm import pandas as pd from typing import List, Dict, Any class BatchReranker: def __init__(self, api_url="http://localhost:7860/api/predict", max_retries=3): self.api_url = api_url self.max_retries = max_retries def process_batch(self, query: str, documents: List[str], instruction: str = None, batch_size: int = 8) -> List[Dict]: """ 处理单个批次的文档重排序 """ payload = { "data": [ query, "\n".join(documents), instruction or "", batch_size ] } for attempt in range(self.max_retries): try: response = requests.post(self.api_url, json=payload, timeout=30) response.raise_for_status() result = response.json() # 解析结果，返回带排序的文档列表 sorted_docs = [] for i, doc_text in enumerate(documents): sorted_docs.append({ "document": doc_text, "original_index": i, "score": result["data"][0][i] if result and "data" in result else 0 }) # 按分数降序排序 sorted_docs.sort(key=lambda x: x["score"], reverse=True) return sorted_docs except Exception as e: print(f"第{attempt + 1}次尝试失败: {e}") if attempt == self.max_retries - 1: raise time.sleep(2 ** attempt) # 指数退避 def process_large_dataset(self, query: str, all_documents: List[str], batch_size: int = 20, max_docs_per_batch: int = 50) -> List[Dict]: """ 处理大量文档的批量重排序 """ if len(all_documents) <= max_docs_per_batch: # 文档数量少，直接处理 return self.process_batch(query, all_documents, batch_size=batch_size) # 文档数量多，分批处理 all_results = [] # 分批处理文档 for i in tqdm(range(0, len(all_documents), max_docs_per_batch), desc="处理文档批次"): batch_docs = all_documents[i:i + max_docs_per_batch] batch_results = self.process_batch(query, batch_docs, batch_size=batch_size) all_results.extend(batch_results) time.sleep(0.1) # 避免请求过于频繁 # 对所有结果进行最终排序 all_results.sort(key=lambda x: x["score"], reverse=True) return all_results # 使用示例 def main(): # 初始化重排序器 reranker = BatchReranker() # 示例查询和文档 query = "解释神经网络的工作原理" # 模拟大量文档（实际应用中可以从文件或数据库读取） documents = [ "神经网络是受人脑启发的计算模型。", "深度学习使用多层神经网络进行特征学习。", "Python是数据科学中常用的编程语言。", "神经网络由输入层、隐藏层和输出层组成。", "反向传播算法用于训练神经网络。", # ... 可以添加更多文档 ] * 10 # 复制10份模拟大量文档 print(f"开始处理 {len(documents)} 个文档...") # 执行批量重排序 results = reranker.process_large_dataset(query, documents, max_docs_per_batch=30) # 输出前10个最相关的结果 print("\nTop 10 最相关文档:") for i, result in enumerate(results[:10]): print(f"{i+1}. [分数: {result['score']:.4f}] {result['document'][:100]}...") # 保存结果到CSV文件 df = pd.DataFrame(results) df.to_csv("rerank_results.csv", index=False, encoding='utf-8-sig') print("\n结果已保存到 rerank_results.csv") if __name__ == "__main__": main() ``` ### 4.2 脚本功能详解 这个批量处理脚本提供了以下核心功能： 1. **分批处理**：支持处理大量文档，自动分成合适的批次 2. **错误重试**：内置重试机制，提高请求的可靠性 3. **进度显示**：使用tqdm显示处理进度，直观了解处理状态 4. **结果保存**：将排序结果保存到CSV文件，方便后续分析 5. **灵活配置**：可以调整批处理大小、每批文档数量等参数 ## 5. 实际应用场景示例 ### 5.1 搜索引擎结果优化 假设你正在构建一个站内搜索引擎，可以使用重排序来优化搜索结果： ```python def optimize_search_results(search_query, raw_results): """ 优化搜索引擎结果 """ reranker = BatchReranker() # 提取文档内容 documents = [result["content"] for result in raw_results] # 重排序 sorted_docs = reranker.process_large_dataset( search_query, documents, max_docs_per_batch=30 ) # 按新顺序重新组织结果 optimized_results = [] for doc_info in sorted_docs: original_index = doc_info["original_index"] optimized_results.append({ **raw_results[original_index], "relevance_score": doc_info["score"] }) return optimized_results ``` ### 5.2 文档库智能检索 对于企业知识库或文档管理系统： ```python class DocumentRetrievalSystem: def __init__(self): self.reranker = BatchReranker() def retrieve_relevant_docs(self, user_query, all_documents, top_k=10): """ 从文档库中检索最相关的文档 """ # 第一步：使用简单的关键词匹配筛选候选文档 candidate_docs = self._prefilter_documents(user_query, all_documents) # 第二步：使用重排序精排 if candidate_docs: sorted_docs = self.reranker.process_large_dataset( user_query, candidate_docs, instruction="Given a user query, retrieve the most relevant documents from the knowledge base", max_docs_per_batch=40 ) return sorted_docs[:top_k] return [] def _prefilter_documents(self, query, documents): """ 初步筛选文档，减少需要重排序的数量 """ # 简单的关键词匹配逻辑 query_keywords = set(query.lower().split()) candidate_docs = [] for doc in documents: doc_keywords = set(doc.lower().split()) if query_keywords & doc_keywords: # 有共同关键词 candidate_docs.append(doc) return candidate_docs[:100] # 最多100个候选文档 ``` ## 6. 性能优化与实用技巧 ### 6.1 调整批处理大小 根据你的硬件配置调整批处理大小： ```python # 根据可用内存调整批处理大小 def get_optimal_batch_size(available_memory_gb): """ 根据可用内存推荐批处理大小 """ if available_memory_gb >= 16: return 32 # 大内存机器 elif available_memory_gb >= 8: return 16 # 中等内存 elif available_memory_gb >= 4: return 8 # 小内存 else: return 4 # 内存紧张 ``` ### 6.2 使用合适的任务指令 针对不同场景使用特定的任务指令： ```python INSTRUCTION_TEMPLATES = { "general": "Given a query, retrieve relevant passages that answer the query", "technical": "Given a technical question, retrieve the most relevant technical documentation", "customer_support": "Given a customer inquiry, retrieve the most relevant support articles", "academic": "Given a research question, retrieve relevant academic papers or excerpts", "code_search": "Given a code-related query, retrieve relevant code snippets or documentation" } def get_instruction(context_type="general"): """ 获取适合场景的任务指令 """ return INSTRUCTION_TEMPLATES.get(context_type, INSTRUCTION_TEMPLATES["general"]) ``` ## 7. 错误处理与日志记录 ### 7.1 增强的错误处理 ```python import logging from datetime import datetime # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(f"reranker_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"), logging.StreamHandler() ] ) class RobustBatchReranker(BatchReranker): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.logger = logging.getLogger(__name__) def process_batch(self, *args, **kwargs): try: start_time = time.time() result = super().process_batch(*args, **kwargs) elapsed = time.time() - start_time self.logger.info(f"成功处理批次: {len(kwargs.get('documents', []))}个文档, " f"耗时: {elapsed:.2f}秒") return result except Exception as e: self.logger.error(f"处理批次失败: {e}", exc_info=True) raise ``` ## 8. 总结与下一步建议 通过本文的批量文档重排序脚本，你现在可以： 1. **高效处理大量文档**：使用分批处理机制，即使面对成千上万的文档也能高效处理 2. **提升检索质量**：通过重排序获得更相关的结果排序 3. **灵活适应不同场景**：支持自定义任务指令，适应各种应用场景 4. **可靠运行**：内置错误处理和日志记录，保证系统稳定性 ### 8.1 下一步学习建议 想要进一步提升技能，可以考虑： 1. **集成到现有系统**：将重排序功能集成到你的搜索引擎或文档管理系统中 2. **性能监控**：添加更详细的性能监控和指标收集 3. **缓存优化**：对常见查询结果进行缓存，提高响应速度 4. **多模型对比**：尝试不同的重排序模型，比较它们的效果 ### 8.2 实际应用提示 在实际项目中使用时，记得： - **调整参数**：根据你的具体需求调整批处理大小和超时设置 - **监控性能**：密切关注内存使用和响应时间 - **测试不同指令**：尝试不同的任务指令，找到最适合你场景的配置 - **处理异常**：确保有完善的错误处理机制，避免单点故障影响整个系统 --- > **获取更多AI镜像** > > 想探索更多AI镜像和应用场景？访问 [CSDN星图镜像广场](https://ai.csdn.net/?utm_source=mirror_blog_end)，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。