文脉定序实战教程在FastAPI中封装BGE重排序服务并添加JWT鉴权1. 引言为什么需要智能重排序服务在日常的信息检索和知识库查询中我们经常遇到这样的困扰搜索引擎能找到大量相关文档但最准确的答案往往排在不显眼的位置。这就是传统检索系统的搜得到但排不准痛点。文脉定序系统正是为了解决这个问题而生。它基于BGE-Reranker-v2-m3模型通过深度学习技术对初步检索结果进行智能重排序让最相关的信息脱颖而出。本教程将手把手教你如何将这个强大的重排序能力封装成FastAPI服务并添加JWT鉴权确保服务安全。学完本文你将能够快速部署BGE重排序模型构建完整的RESTful API服务实现安全的身份验证机制掌握生产环境部署的最佳实践2. 环境准备与依赖安装2.1 系统要求与Python环境首先确保你的系统满足以下要求Python 3.8或更高版本至少8GB内存推荐16GBGPU支持可选但能显著提升速度创建并激活虚拟环境python -m venv reranker_env source reranker_env/bin/activate # Linux/Mac # 或者 reranker_env\Scripts\activate # Windows2.2 安装必要依赖创建requirements.txt文件fastapi0.104.1 uvicorn0.24.0 python-jose[cryptography]3.3.0 passlib[bcrypt]1.7.4 python-multipart0.0.6 transformers4.35.0 torch2.1.0 sentence-transformers2.2.2 pydantic2.5.0安装依赖pip install -r requirements.txt3. 核心代码实现3.1 创建FastAPI应用基础结构首先建立项目的基本结构bge-reranker-api/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── models.py │ ├── auth.py │ └── reranker.py ├── requirements.txt └── README.md3.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], deprecatedauto) oauth2_scheme OAuth2PasswordBearer(tokenUrltoken) # 用户模型 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(minutes15) to_encode.update({exp: expire}) encoded_jwt jwt.encode(to_encode, SECRET_KEY, algorithmALGORITHM) return encoded_jwt async def get_current_user(token: str Depends(oauth2_scheme)): credentials_exception HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detailCould 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 user3.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, paddingTrue, truncationTrue, return_tensorspt, max_length512 ) inputs {k: v.to(self.device) for k, v in inputs.items()} scores self.model(**inputs).logits.squeeze(dim1) scores torch.sigmoid(scores).cpu().numpy() # 按分数排序 ranked_results sorted( zip(passages, scores), keylambda x: x[1], reverseTrue ) 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的智能语义重排序服务, version1.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_modelToken) 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_codestatus.HTTP_401_UNAUTHORIZED, detailIncorrect username or password, headers{WWW-Authenticate: Bearer}, ) access_token_expires timedelta(minutesACCESS_TOKEN_EXPIRE_MINUTES) access_token create_access_token( data{sub: user.username}, expires_deltaaccess_token_expires ) return {access_token: access_token, token_type: bearer} # 重排序路由 app.post(/rerank, response_modelRerankResponse) 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_passagesranked_passages, scoresscores ) except Exception as e: raise HTTPException( status_codestatus.HTTP_500_INTERNAL_SERVER_ERROR, detailf重排序处理失败: {str(e)} ) # 健康检查路由 app.get(/health) async def health_check(): return {status: healthy, model_loaded: True} if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)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 usernameadminpasswordadminpassword响应示例{ 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部署对于生产环境建议使用Gunicorngunicorn -w 4 -k uvicorn.workers.UvicornWorker app.main:app6. 总结通过本教程我们成功构建了一个完整的文脉定序重排序服务核心技术实现使用BGE-Reranker-v2-m3模型实现智能语义重排序基于FastAPI构建了RESTful API服务实现了JWT身份验证保障接口安全主要功能特点支持多段落同时重排序返回详细的置信度分数提供完整的API文档界面具备生产级别的安全措施实际应用价值 这个服务可以轻松集成到现有的搜索系统、知识库平台或RAG应用中显著提升检索结果的相关性和准确性。无论是企业内部的文档检索还是面向用户的智能问答系统都能从中受益。下一步你可以考虑添加批量处理接口提升效率实现异步处理支持大规模请求添加缓存机制减少重复计算集成监控和日志系统获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。