文脉定序实战教程：在FastAPI中封装BGE重排序服务并添加JWT鉴权-育师

文脉定序实战教程：在FastAPI中封装BGE重排序服务并添加JWT鉴权

1. 引言：为什么需要智能重排序服务？

在日常的信息检索和知识库查询中，我们经常遇到这样的困扰：搜索引擎能找到大量相关文档，但最准确的答案往往排在不显眼的位置。这就是传统检索系统的"搜得到但排不准"痛点。

文脉定序系统正是为了解决这个问题而生。它基于BGE-Reranker-v2-m3模型，通过深度学习技术对初步检索结果进行智能重排序，让最相关的信息脱颖而出。

本教程将手把手教你如何将这个强大的重排序能力封装成FastAPI服务，并添加JWT鉴权确保服务安全。学完本文，你将能够：

快速部署BGE重排序模型
构建完整的RESTful API服务
实现安全的身份验证机制
掌握生产环境部署的最佳实践

2. 环境准备与依赖安装

2.1 系统要求与Python环境

首先确保你的系统满足以下要求：

Python 3.8或更高版本
至少8GB内存（推荐16GB）
GPU支持（可选，但能显著提升速度）

创建并激活虚拟环境：

python -m venv reranker_env source reranker_env/bin/activate # Linux/Mac # 或者 reranker_env\Scripts\activate # Windows

2.2 安装必要依赖

创建requirements.txt文件：

fastapi==0.104.1 uvicorn==0.24.0 python-jose[cryptography]==3.3.0 passlib[bcrypt]==1.7.4 python-multipart==0.0.6 transformers==4.35.0 torch==2.1.0 sentence-transformers==2.2.2 pydantic==2.5.0

安装依赖：

pip install -r requirements.txt

3. 核心代码实现

3.1 创建FastAPI应用基础结构

首先建立项目的基本结构：

bge-reranker-api/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── models.py │ ├── auth.py │ └── reranker.py ├── requirements.txt └── README.md

3.2 实现JWT鉴权模块

在auth.py中实现身份验证功能：

from datetime import datetime, timedelta from jose import JWTError, jwt from passlib.context import CryptContext from pydantic import BaseModel from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer # 安全配置 SECRET_KEY = "your-secret-key-change-in-production" ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 # 密码哈希 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # 用户模型 class User(BaseModel): username: str disabled: bool = False class UserInDB(User): hashed_password: str # 模拟用户数据库 fake_users_db = { "admin": { "username": "admin", "hashed_password": pwd_context.hash("adminpassword"), "disabled": False } } def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_user(db, username: str): if username in db: user_dict = db[username] return UserInDB(**user_dict) def authenticate_user(fake_db, username: str, password: str): user = get_user(fake_db, username) if not user: return False if not verify_password(password, user.hashed_password): return False return user def create_access_token(data: dict, expires_delta: timedelta = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt async def get_current_user(token: str = Depends(oauth2_scheme)): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) username: str = payload.get("sub") if username is None: raise credentials_exception except JWTError: raise credentials_exception user = get_user(fake_users_db, username) if user is None: raise credentials_exception return user

3.3 实现重排序核心功能

在reranker.py中实现BGE模型封装：

from transformers import AutoModelForSequenceClassification, AutoTokenizer import torch import numpy as np from typing import List, Tuple class BGEReranker: def __init__(self, model_name: str = "BAAI/bge-reranker-v2-m3"): self.device = "cuda" if torch.cuda.is_available() else "cpu" self.tokenizer = AutoTokenizer.from_pretrained(model_name) self.model = AutoModelForSequenceClassification.from_pretrained(model_name) self.model.to(self.device) self.model.eval() def rerank(self, query: str, passages: List[str]) -> List[Tuple[str, float]]: """ 对候选段落进行重排序 """ pairs = [[query, passage] for passage in passages] with torch.no_grad(): 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()} scores = self.model(**inputs).logits.squeeze(dim=1) scores = torch.sigmoid(scores).cpu().numpy() # 按分数排序 ranked_results = sorted( zip(passages, scores), key=lambda x: x[1], reverse=True ) return ranked_results # 全局重排序器实例 reranker = BGEReranker()

3.4 构建完整的FastAPI应用

在main.py中整合所有功能：

from fastapi import FastAPI, Depends, HTTPException, status from fastapi.security import OAuth2PasswordRequestForm from typing import List from pydantic import BaseModel from app.auth import ( authenticate_user, create_access_token, get_current_user, ACCESS_TOKEN_EXPIRE_MINUTES ) from app.reranker import reranker app = FastAPI( title="文脉定序重排序API", description="基于BGE-Reranker-v2-m3的智能语义重排序服务", version="1.0.0" ) # 请求响应模型 class RerankRequest(BaseModel): query: str passages: List[str] class RerankResponse(BaseModel): ranked_passages: List[str] scores: List[float] class Token(BaseModel): access_token: str token_type: str # 认证路由 @app.post("/token", response_model=Token) async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()): user = authenticate_user(fake_users_db, form_data.username, form_data.password) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={"sub": user.username}, expires_delta=access_token_expires ) return {"access_token": access_token, "token_type": "bearer"} # 重排序路由 @app.post("/rerank", response_model=RerankResponse) async def rerank_passages( request: RerankRequest, current_user: User = Depends(get_current_user) ): try: results = reranker.rerank(request.query, request.passages) ranked_passages = [result[0] for result in results] scores = [float(result[1]) for result in results] return RerankResponse( ranked_passages=ranked_passages, scores=scores ) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"重排序处理失败: {str(e)}" ) # 健康检查路由 @app.get("/health") async def health_check(): return {"status": "healthy", "model_loaded": True} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

4. 服务部署与测试

4.1 启动服务

使用以下命令启动FastAPI服务：

uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

服务启动后，访问 http://localhost:8000/docs 可以看到自动生成的API文档。

4.2 获取访问令牌

首先需要获取JWT令牌：

curl -X POST "http://localhost:8000/token" \ -H "Content-Type: application/x-www-form-urlencoded" \ -d "username=admin&password=adminpassword"

响应示例：

{ "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...", "token_type": "bearer" }

4.3 测试重排序功能

使用获取的令牌调用重排序接口：

curl -X POST "http://localhost:8000/rerank" \ -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..." \ -H "Content-Type: application/json" \ -d '{ "query": "人工智能的发展历史", "passages": [ "机器学习是人工智能的一个重要分支", "深度学习在2010年后取得了突破性进展", "人工智能概念最早在1956年达特茅斯会议上提出", "神经网络是模仿人脑结构的计算模型" ] }'

响应示例：

{ "ranked_passages": [ "人工智能概念最早在1956年达特茅斯会议上提出", "机器学习是人工智能的一个重要分支", "深度学习在2010年后取得了突破性进展", "神经网络是模仿人脑结构的计算模型" ], "scores": [0.95, 0.87, 0.82, 0.78] }

5. 生产环境部署建议

5.1 安全性增强

在实际生产环境中，需要加强安全措施：

# 使用环境变量存储敏感信息 import os from dotenv import load_dotenv load_dotenv() SECRET_KEY = os.getenv("SECRET_KEY") ALGORITHM = os.getenv("ALGORITHM", "HS256")

5.2 性能优化配置

创建Dockerfile优化部署：

FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . # 安装GPU支持（可选） RUN pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 EXPOSE 8000 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

5.3 使用Gunicorn部署

对于生产环境，建议使用Gunicorn：

gunicorn -w 4 -k uvicorn.workers.UvicornWorker app.main:app

6. 总结

通过本教程，我们成功构建了一个完整的文脉定序重排序服务：

核心技术实现：

使用BGE-Reranker-v2-m3模型实现智能语义重排序
基于FastAPI构建了RESTful API服务
实现了JWT身份验证保障接口安全

主要功能特点：

支持多段落同时重排序
返回详细的置信度分数
提供完整的API文档界面
具备生产级别的安全措施

实际应用价值：这个服务可以轻松集成到现有的搜索系统、知识库平台或RAG应用中，显著提升检索结果的相关性和准确性。无论是企业内部的文档检索，还是面向用户的智能问答系统，都能从中受益。

下一步，你可以考虑：

添加批量处理接口提升效率
实现异步处理支持大规模请求
添加缓存机制减少重复计算
集成监控和日志系统

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

文脉定序实战教程：在FastAPI中封装BGE重排序服务并添加JWT鉴权