基于FastAPI的Lychee Rerank高性能API服务
你是不是也遇到过这样的场景？自己搭建的智能问答系统，明明检索到了一堆相关文档，但最后生成的答案却总是不太对劲，要么答非所问，要么漏掉了关键信息。
问题往往出在“排序”这个环节。传统的向量检索只能找到语义相似的文档，却无法判断哪个文档真正“有用”。这就好比你去图书馆找书，管理员只告诉你“这几本书都跟你的问题有关”，却不告诉你哪本最值得读。
今天，我们就来解决这个问题。我将带你用FastAPI搭建一个高性能的Lychee Rerank API服务，它能像一位经验丰富的图书管理员，从一堆相关文档中精准挑出最有价值的那几份。
1. 为什么需要重排序服务？
在深入代码之前，我们先聊聊为什么重排序如此重要。
想象一下，你问AI：“如何快速学习Python？”向量检索可能会返回：
- Python官方文档（权威但冗长）
- 某博客的“10分钟学会Python”（标题党）
- 大学Python课程大纲（系统但耗时）
- Stack Overflow的某个具体问题解答（针对性强但片面）
如果没有重排序，AI可能随机选一个作为参考，结果自然不尽如人意。而重排序模型（比如Lychee Rerank）会分析每个文档与问题的“实际相关性”，而不仅仅是“语义相似性”。
它考虑的因素包括：
- 文档是否直接回答了问题
- 信息的时效性如何
- 内容的权威性怎样
- 表述是否清晰易懂
我们的目标就是把这个智能的“图书管理员”封装成一个随时可用的API服务。
2. 环境准备与快速部署
2.1 系统要求
- Python 3.8+
- 至少4GB内存（运行模型需要）
- 支持CUDA的GPU（可选，能大幅加速）
2.2 安装依赖
创建一个新的项目目录，然后安装必要的包：
# 创建项目目录 mkdir lychee-rerank-api cd lychee-rerank-api # 创建虚拟环境（推荐） python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心依赖 pip install fastapi uvicorn pydantic # 安装模型相关依赖 pip install torch transformers # 可选：如果需要GPU加速 pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu1182.3 项目结构规划
在开始编码前，我们先规划一下项目结构：
lychee-rerank-api/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI应用入口 │ ├── models.py # 数据模型定义 │ ├── rerank.py # 重排序核心逻辑 │ └── config.py # 配置管理 ├── requirements.txt └── README.md3. 核心代码实现
3.1 定义数据模型
首先，在app/models.py中定义API的请求和响应格式：
from pydantic import BaseModel from typing import List, Optional class RerankRequest(BaseModel): """重排序请求模型""" query: str # 用户查询 documents: List[str] # 待排序的文档列表 top_k: Optional[int] = 5 # 返回前K个结果，默认5个 class DocumentScore(BaseModel): """文档得分模型""" document: str # 文档内容 score: float # 相关性得分 rank: int # 排名位置 class RerankResponse(BaseModel): """重排序响应模型""" query: str # 原始查询 results: List[DocumentScore] # 排序后的结果 total_documents: int # 处理的文档总数 processing_time: float # 处理耗时（秒）3.2 实现重排序核心逻辑
接下来是重头戏，在app/rerank.py中实现Lychee Rerank模型：
import torch from transformers import AutoTokenizer, AutoModelForSequenceClassification from typing import List, Tuple import time class LycheeReranker: """Lychee重排序器""" def __init__(self, model_name: str = "BAAI/bge-reranker-base"): """ 初始化重排序模型 Args: model_name: 模型名称，默认使用BAAI的bge-reranker-base """ print(f"正在加载模型: {model_name}") start_time = time.time() # 加载tokenizer和模型 self.tokenizer = AutoTokenizer.from_pretrained(model_name) self.model = AutoModelForSequenceClassification.from_pretrained(model_name) # 设置为评估模式 self.model.eval() # 如果有GPU，移到GPU上 self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") self.model.to(self.device) load_time = time.time() - start_time print(f"模型加载完成，耗时: {load_time:.2f}秒") print(f"使用设备: {self.device}") def rerank(self, query: str, documents: List[str], top_k: int = 5) -> List[Tuple[str, float]]: """ 对文档进行重排序 Args: query: 查询文本 documents: 文档列表 top_k: 返回前K个结果 Returns: 排序后的(文档, 得分)列表 """ if not documents: return [] scores = [] # 批量处理文档（实际生产环境可以优化为真正的批量处理） with torch.no_grad(): for doc in documents: # 准备输入 inputs = self.tokenizer( query, doc, padding=True, truncation=True, return_tensors="pt", max_length=512 ) # 移到对应设备 inputs = {k: v.to(self.device) for k, v in inputs.items()} # 前向传播 outputs = self.model(**inputs) # 获取相关性得分 score = outputs.logits.squeeze().item() scores.append((doc, score)) # 按得分降序排序 scores.sort(key=lambda x: x[1], reverse=True) # 返回前top_k个结果 return scores[:top_k] async def rerank_async(self, query: str, documents: List[str], top_k: int = 5): """异步版本的重排序（实际还是同步计算，但接口是异步的）""" return self.rerank(query, documents, top_k)3.3 创建FastAPI应用
现在，在app/main.py中创建FastAPI应用：
from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware import time from typing import List from .models import RerankRequest, RerankResponse, DocumentScore from .rerank import LycheeReranker # 创建FastAPI应用 app = FastAPI( title="Lychee Rerank API", description="基于Lychee模型的高性能重排序API服务", version="1.0.0" ) # 添加CORS中间件（允许跨域请求） app.add_middleware( CORSMiddleware, allow_origins=["*"], # 生产环境应该限制具体域名 allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 全局重排序器实例 reranker = None @app.on_event("startup") async def startup_event(): """应用启动时加载模型""" global reranker print("应用启动中，正在加载重排序模型...") reranker = LycheeReranker() print("重排序模型加载完成，API服务准备就绪") @app.get("/") async def root(): """根路径，返回服务信息""" return { "service": "Lychee Rerank API", "status": "running", "version": "1.0.0", "endpoints": { "health": "/health", "rerank": "/api/v1/rerank", "docs": "/docs" } } @app.get("/health") async def health_check(): """健康检查端点""" return { "status": "healthy", "model_loaded": reranker is not None, "device": str(reranker.device) if reranker else "unknown" } @app.post("/api/v1/rerank", response_model=RerankResponse) async def rerank_documents(request: RerankRequest): """ 重排序API端点 - **query**: 用户查询文本 - **documents**: 需要排序的文档列表 - **top_k**: 返回前K个结果（默认5） """ if reranker is None: raise HTTPException(status_code=503, detail="服务正在初始化，请稍后重试") if not request.documents: raise HTTPException(status_code=400, detail="文档列表不能为空") # 记录开始时间 start_time = time.time() try: # 执行重排序 results = reranker.rerank( query=request.query, documents=request.documents, top_k=request.top_k ) # 计算处理时间 processing_time = time.time() - start_time # 构建响应 ranked_results = [] for rank, (doc, score) in enumerate(results, 1): ranked_results.append( DocumentScore( document=doc, score=score, rank=rank ) ) return RerankResponse( query=request.query, results=ranked_results, total_documents=len(request.documents), processing_time=processing_time ) except Exception as e: raise HTTPException(status_code=500, detail=f"重排序失败: {str(e)}") @app.post("/api/v1/rerank/batch") async def rerank_batch(requests: List[RerankRequest]): """ 批量重排序API端点 注意：这个端点目前是顺序处理，实际生产环境需要优化为真正的批量处理 """ if reranker is None: raise HTTPException(status_code=503, detail="服务正在初始化，请稍后重试") responses = [] total_start = time.time() for request in requests: start_time = time.time() results = reranker.rerank( query=request.query, documents=request.documents, top_k=request.top_k ) processing_time = time.time() - start_time ranked_results = [] for rank, (doc, score) in enumerate(results, 1): ranked_results.append({ "document": doc, "score": score, "rank": rank }) responses.append({ "query": request.query, "results": ranked_results, "total_documents": len(request.documents), "processing_time": processing_time }) total_time = time.time() - total_start return { "batch_results": responses, "total_queries": len(requests), "total_processing_time": total_time, "average_time_per_query": total_time / len(requests) if requests else 0 }3.4 配置管理
在app/config.py中添加配置管理：
import os from typing import Optional class Config: """应用配置""" # API配置 API_HOST: str = os.getenv("API_HOST", "0.0.0.0") API_PORT: int = int(os.getenv("API_PORT", "8000")) API_WORKERS: int = int(os.getenv("API_WORKERS", "1")) # 模型配置 MODEL_NAME: str = os.getenv("MODEL_NAME", "BAAI/bge-reranker-base") MAX_SEQUENCE_LENGTH: int = int(os.getenv("MAX_SEQUENCE_LENGTH", "512")) # 性能配置 BATCH_SIZE: int = int(os.getenv("BATCH_SIZE", "16")) USE_GPU: bool = os.getenv("USE_GPU", "true").lower() == "true" # 日志配置 LOG_LEVEL: str = os.getenv("LOG_LEVEL", "info") @classmethod def print_config(cls): """打印当前配置""" print("=== 应用配置 ===") for key, value in cls.__dict__.items(): if not key.startswith("__") and not callable(value): print(f"{key}: {value}") print("================") # 创建配置实例 config = Config()4. 启动和测试服务
4.1 创建启动脚本
在项目根目录创建run.py：
import uvicorn from app.config import config if __name__ == "__main__": # 打印配置信息 config.print_config() # 启动FastAPI应用 uvicorn.run( "app.main:app", host=config.API_HOST, port=config.API_PORT, workers=config.API_WORKERS, log_level=config.LOG_LEVEL, reload=True # 开发环境启用热重载 )4.2 启动服务
# 确保在项目根目录 python run.py如果一切正常，你会看到类似这样的输出：
=== 应用配置 === API_HOST: 0.0.0.0 API_PORT: 8000 API_WORKERS: 1 MODEL_NAME: BAAI/bge-reranker-base ... 应用启动中，正在加载重排序模型... 正在加载模型: BAAI/bge-reranker-base 模型加载完成，耗时: 3.45秒 使用设备: cuda 重排序模型加载完成，API服务准备就绪 INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)4.3 测试API
打开浏览器访问http://localhost:8000/docs，你会看到自动生成的API文档界面。
让我们用Python脚本测试一下：
import requests import json # API地址 url = "http://localhost:8000/api/v1/rerank" # 测试数据 test_data = { "query": "如何学习Python编程？", "documents": [ "Python是一种高级编程语言，适合初学者学习。", "Java是企业级应用开发的首选语言。", "学习Python需要掌握基础语法和常用库。", "C++主要用于系统级编程和游戏开发。", "Python在数据科学和机器学习领域应用广泛。", "JavaScript是Web前端开发的核心语言。", "Python社区活跃，有大量的学习资源。", "Go语言以其并发性能著称。" ], "top_k": 3 } # 发送请求 response = requests.post(url, json=test_data) # 打印结果 print("状态码:", response.status_code) print("响应内容:") print(json.dumps(response.json(), indent=2, ensure_ascii=False))你应该能看到类似这样的输出：
{ "query": "如何学习Python编程？", "results": [ { "document": "学习Python需要掌握基础语法和常用库。", "score": 8.765, "rank": 1 }, { "document": "Python是一种高级编程语言，适合初学者学习。", "score": 8.234, "rank": 2 }, { "document": "Python在数据科学和机器学习领域应用广泛。", "score": 7.891, "rank": 3 } ], "total_documents": 8, "processing_time": 0.345 }5. 性能优化技巧
5.1 启用真正的批量处理
上面的实现是伪批量（逐个处理），我们可以优化为真正的批量处理：
def rerank_batch(self, query: str, documents: List[str], top_k: int = 5): """真正的批量处理版本""" if not documents: return [] # 准备所有输入对 pairs = [(query, doc) for doc in documents] # 批量编码 inputs = self.tokenizer( pairs, padding=True, truncation=True, return_tensors="pt", max_length=512 ) # 移到对应设备 inputs = {k: v.to(self.device) for k, v in inputs.items()} # 批量前向传播 with torch.no_grad(): outputs = self.model(**inputs) scores = outputs.logits.squeeze().tolist() # 如果是单个文档，scores可能是标量 if isinstance(scores, float): scores = [scores] # 组合结果并排序 results = list(zip(documents, scores)) results.sort(key=lambda x: x[1], reverse=True) return results[:top_k]5.2 添加缓存机制
对于重复的查询，我们可以添加缓存来提高性能：
from functools import lru_cache import hashlib class CachedReranker(LycheeReranker): """带缓存的重排序器""" def __init__(self, model_name: str = "BAAI/bge-reranker-base", cache_size: int = 1000): super().__init__(model_name) self.cache_size = cache_size def _get_cache_key(self, query: str, documents: List[str]) -> str: """生成缓存键""" # 使用MD5哈希作为缓存键 content = query + "|||" + "|||".join(documents) return hashlib.md5(content.encode()).hexdigest() @lru_cache(maxsize=1000) def rerank_cached(self, query: str, documents_tuple: tuple, top_k: int = 5): """带缓存的排序（注意：documents需要转为tuple才能被缓存）""" documents = list(documents_tuple) return self.rerank(query, documents, top_k) def rerank(self, query: str, documents: List[str], top_k: int = 5): """重写rerank方法，添加缓存逻辑""" # 将列表转为元组以便缓存 documents_tuple = tuple(documents) return self.rerank_cached(query, documents_tuple, top_k)5.3 异步处理优化
对于高并发场景，我们可以使用异步批处理：
import asyncio from concurrent.futures import ThreadPoolExecutor class AsyncReranker: """异步重排序器""" def __init__(self, model_name: str = "BAAI/bge-reranker-base", max_workers: int = 4): self.reranker = LycheeReranker(model_name) self.executor = ThreadPoolExecutor(max_workers=max_workers) self.loop = asyncio.get_event_loop() async def rerank_async(self, query: str, documents: List[str], top_k: int = 5): """真正的异步处理""" # 将同步的rerank方法放到线程池中执行 return await self.loop.run_in_executor( self.executor, lambda: self.reranker.rerank(query, documents, top_k) ) async def rerank_batch_async(self, requests: List[RerankRequest]): """批量异步处理""" tasks = [] for request in requests: task = self.rerank_async( request.query, request.documents, request.top_k ) tasks.append(task) # 并发执行所有任务 return await asyncio.gather(*tasks)6. 生产环境部署建议
6.1 使用Docker容器化
创建Dockerfile：
FROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update && apt-get install -y \ gcc \ g++ \ && rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . # 安装Python依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 暴露端口 EXPOSE 8000 # 启动命令 CMD ["python", "run.py"]创建docker-compose.yml：
version: '3.8' services: lychee-rerank-api: build: . ports: - "8000:8000" environment: - MODEL_NAME=BAAI/bge-reranker-base - API_HOST=0.0.0.0 - API_PORT=8000 - USE_GPU=true deploy: resources: reservations: devices: - driver: nvidia count: 1 capabilities: [gpu] volumes: - ./models:/app/models # 挂载模型缓存目录 restart: unless-stopped6.2 添加监控和日志
在app/main.py中添加：
import logging from fastapi import Request import time # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) @app.middleware("http") async def log_requests(request: Request, call_next): """请求日志中间件""" start_time = time.time() response = await call_next(request) process_time = time.time() - start_time logger.info( f"{request.method} {request.url.path} - " f"Status: {response.status_code} - " f"Time: {process_time:.3f}s" ) return response @app.get("/metrics") async def get_metrics(): """监控指标端点""" # 这里可以添加各种业务指标 return { "requests_processed": 0, # 实际应该从计数器获取 "average_response_time": 0.0, "error_rate": 0.0, "cache_hit_rate": 0.0 }6.3 配置反向代理（Nginx）
创建nginx.conf：
server { listen 80; server_name rerank.yourdomain.com; location / { proxy_pass http://localhost:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; # 超时设置 proxy_connect_timeout 60s; proxy_send_timeout 60s; proxy_read_timeout 60s; } # 限制请求大小 client_max_body_size 10M; # 启用gzip压缩 gzip on; gzip_types application/json; }7. 实际应用场景
7.1 集成到RAG系统
class RAGSystem: """检索增强生成系统""" def __init__(self, rerank_api_url: str = "http://localhost:8000"): self.rerank_api_url = rerank_api_url # 初始化向量数据库、LLM等... async def answer_question(self, question: str): """回答问题""" # 1. 从向量数据库检索相关文档 retrieved_docs = await self.retrieve_documents(question) # 2. 使用重排序API对文档排序 sorted_docs = await self.rerank_documents(question, retrieved_docs) # 3. 使用LLM生成答案 answer = await self.generate_answer(question, sorted_docs) return answer async def rerank_documents(self, question: str, documents: List[str]): """调用重排序API""" import aiohttp async with aiohttp.ClientSession() as session: payload = { "query": question, "documents": documents, "top_k": 5 } async with session.post( f"{self.rerank_api_url}/api/v1/rerank", json=payload ) as response: result = await response.json() return [doc["document"] for doc in result["results"]]7.2 智能客服系统
class CustomerServiceBot: """智能客服机器人""" def __init__(self, rerank_api_url: str): self.rerank_api_url = rerank_api_url self.knowledge_base = self.load_knowledge_base() async def handle_customer_query(self, query: str): """处理客户查询""" # 从知识库中找到相关FAQ relevant_faqs = self.find_relevant_faqs(query) if len(relevant_faqs) > 1: # 如果有多个相关FAQ，使用重排序找出最相关的 reranked = await self.rerank_faqs(query, relevant_faqs) best_faq = reranked[0] else: best_faq = relevant_faqs[0] if relevant_faqs else None if best_faq and best_faq["confidence"] > 0.8: # 直接返回FAQ答案 return best_faq["answer"] else: # 转人工或使用LLM生成答案 return await self.escalate_to_llm(query, relevant_faqs)8. 总结
通过这篇文章，我们从头构建了一个基于FastAPI的Lychee Rerank高性能API服务。从环境搭建、核心代码实现，到性能优化和生产部署，我们覆盖了实际开发中的关键环节。
这个服务最大的价值在于，它把复杂的重排序模型封装成了简单易用的API接口。无论你是要构建智能问答系统、内容推荐引擎，还是智能客服，都可以通过几行HTTP调用获得专业的文档排序能力。
实际使用下来，部署过程比想象中要简单，FastAPI的异步特性让服务能够轻松应对高并发场景。性能方面，在GPU支持下，单次重排序通常在几百毫秒内完成，完全满足生产环境的要求。
如果你正在构建需要智能检索的系统，不妨试试这个方案。先从简单的版本开始，跑通基本流程，然后根据实际需求逐步添加缓存、批量处理、监控等高级功能。遇到问题也不用担心，FastAPI的自动文档和清晰的错误信息会让调试变得轻松很多。
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
