通义千问3-Reranker-0.6B API服务搭建：FastAPI实战-开发者社区

通义千问3-Reranker-0.6B API服务搭建：FastAPI实战

1. 引言

如果你正在构建一个智能检索系统，可能会遇到这样的问题：初步检索返回的结果很多，但真正相关的却不多。这时候就需要一个"精排"模型来帮你从海量候选中筛选出最相关的内容。通义千问3-Reranker-0.6B就是这样一个专门做文本重排序的轻量级模型。

今天我将手把手教你如何使用FastAPI框架，为这个重排序模型搭建一个完整的RESTful API服务。无论你是想为自己的项目添加智能排序功能，还是希望将模型能力封装成服务供团队使用，这篇教程都能帮到你。

我们将从环境准备开始，一步步实现接口设计、并发处理、性能监控等关键功能，最终打造一个稳定高效的重排序API服务。

2. 环境准备与模型部署

2.1 基础环境配置

首先确保你的Python环境是3.8或更高版本，然后安装必要的依赖包：

pip install fastapi uvicorn transformers torch sentence-transformers pip install python-multipart pydantic-settings

这些包涵盖了Web框架、模型加载、数据处理等核心功能。如果你打算在生产环境部署，还可以考虑安装gunicorn作为进程管理器。

2.2 模型下载与初始化

通义千问3-Reranker-0.6B模型可以从Hugging Face平台获取。创建一个简单的模型加载脚本：

from transformers import AutoModelForCausalLM, AutoTokenizer import torch def load_reranker_model(): """加载重排序模型和分词器""" model_name = "Qwen/Qwen3-Reranker-0.6B" # 加载分词器，设置padding_side为left以确保正确性 tokenizer = AutoTokenizer.from_pretrained( model_name, padding_side='left', trust_remote_code=True ) # 加载模型并设置为评估模式 model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=torch.float16, # 使用半精度减少内存占用 device_map="auto", # 自动选择设备（GPU/CPU） trust_remote_code=True ).eval() return model, tokenizer

这个加载函数会自动处理设备分配，优先使用GPU加速推理。如果你的显存有限，可以考虑使用更低的精度或者模型量化技术。

3. FastAPI服务核心实现

3.1 应用初始化与配置

创建一个完整的FastAPI应用，包含必要的配置和中间件：

from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from typing import List, Optional import time import logging # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI( title="Qwen3-Reranker API", description="通义千问3-Reranker-0.6B的RESTful API服务", version="1.0.0" ) # 配置CORS中间件 app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 定义请求响应模型 class RerankRequest(BaseModel): query: str documents: List[str] instruction: Optional[str] = None top_k: Optional[int] = None class RerankResult(BaseModel): document: str score: float rank: int class RerankResponse(BaseModel): results: List[RerankResult] processing_time: float

这里我们定义了清晰的数据模型，确保API的输入输出结构明确。CORS配置使得前端应用可以方便地调用这个API。

3.2 重排序核心逻辑

实现模型的重排序功能，这是整个服务的核心：

# 全局模型变量 reranker_model = None reranker_tokenizer = None # 重排序配置 token_false_id = None token_true_id = None max_reranker_length = 8192 @app.on_event("startup") async def startup_event(): """应用启动时加载模型""" global reranker_model, reranker_tokenizer, token_false_id, token_true_id logger.info("正在加载重排序模型...") reranker_model, reranker_tokenizer = load_reranker_model() # 获取特殊token的ID token_false_id = reranker_tokenizer.convert_tokens_to_ids("no") token_true_id = reranker_tokenizer.convert_tokens_to_ids("yes") logger.info("模型加载完成，服务准备就绪") def format_instruction(instruction: str, query: str, document: str) -> str: """格式化指令用于重排序输入""" if instruction is None: instruction = 'Given a web search query, retrieve relevant passages that answer the query' return f"<Instruct>: {instruction}\n<Query>: {query}\n<Document>: {document}" def process_inputs(pairs: List[str]): """处理输入用于重排序模型""" inputs = reranker_tokenizer( pairs, padding=False, truncation='longest_first', return_attention_mask=False, max_length=max_reranker_length - len(prefix_tokens) - len(suffix_tokens) ) # 添加特殊token for i, ele in enumerate(inputs['input_ids']): inputs['input_ids'][i] = prefix_tokens + ele + suffix_tokens # 填充并转换为tensor inputs = reranker_tokenizer.pad(inputs, padding=True, return_tensors="pt") for key in inputs: inputs[key] = inputs[key].to(reranker_model.device) return inputs @torch.no_grad() def compute_relevance_scores(inputs): """计算相关性分数""" batch_scores = reranker_model(**inputs).logits[:, -1, :] true_vector = batch_scores[:, token_true_id] false_vector = batch_scores[:, token_false_id] batch_scores = torch.stack([false_vector, true_vector], dim=1) batch_scores = torch.nn.functional.log_softmax(batch_scores, dim=1) scores = batch_scores[:, 1].exp().tolist() return scores

3.3 API端点实现

创建主要的重排序API端点：

# 定义特殊token prefix = "<|im_start|>system\nJudge whether the Document meets the requirements based on the Query and the Instruct provided. Note that the answer can only be \"yes\" or \"no\".<|im_end|>\n<|im_start|>user\n" suffix = "<|im_end|>\n<|im_start|>assistant\n" prefix_tokens = reranker_tokenizer.encode(prefix, add_special_tokens=False) suffix_tokens = reranker_tokenizer.encode(suffix, add_special_tokens=False) @app.post("/rerank", response_model=RerankResponse) async def rerank_documents(request: RerankRequest): """重排序API端点""" start_time = time.time() try: if not request.documents: raise HTTPException(status_code=400, detail="文档列表不能为空") # 格式化输入对 pairs = [format_instruction(request.instruction, request.query, doc) for doc in request.documents] # 处理输入并计算分数 inputs = process_inputs(pairs) scores = compute_relevance_scores(inputs) # 组合文档和分数 doc_scores = list(zip(request.documents, scores)) doc_scores.sort(key=lambda x: x[1], reverse=True) # 应用top_k筛选 if request.top_k is not None and request.top_k > 0: doc_scores = doc_scores[:request.top_k] # 构建响应 results = [ RerankResult(document=doc, score=score, rank=idx + 1) for idx, (doc, score) in enumerate(doc_scores) ] processing_time = time.time() - start_time logger.info(f"重排序完成: 处理了 {len(request.documents)} 个文档, 耗时 {processing_time:.2f}秒") return RerankResponse( results=results, processing_time=processing_time ) except Exception as e: logger.error(f"重排序处理失败: {str(e)}") raise HTTPException(status_code=500, detail=f"处理失败: {str(e)}")

这个端点接收查询语句、文档列表和可选指令，返回按相关性排序的结果。

4. 高级功能与性能优化

4.1 并发处理与批处理

为了提高吞吐量，我们需要实现批处理功能：

from fastapi import BackgroundTasks import asyncio from concurrent.futures import ThreadPoolExecutor # 创建线程池用于CPU密集型操作 executor = ThreadPoolExecutor(max_workers=4) @app.post("/rerank/batch", response_model=List[RerankResponse]) async def batch_rerank(requests: List[RerankRequest], background_tasks: BackgroundTasks): """批量重排序端点""" results = [] # 使用线程池并行处理每个请求 loop = asyncio.get_event_loop() tasks = [ loop.run_in_executor(executor, process_single_request, request) for request in requests ] results = await asyncio.gather(*tasks) return results def process_single_request(request: RerankRequest) -> RerankResponse: """处理单个重排序请求（用于批处理）""" # 这里复用之前实现的逻辑 # 为了简洁省略具体实现 pass

4.2 性能监控与健康检查

添加监控端点来跟踪服务状态：

from prometheus_client import Counter, Histogram, generate_latest from fastapi import Response # 定义监控指标 REQUEST_COUNT = Counter('rerank_requests_total', 'Total rerank requests', ['method', 'endpoint']) REQUEST_LATENCY = Histogram('rerank_request_latency_seconds', 'Request latency', ['endpoint']) @app.get("/metrics") async def metrics(): """Prometheus监控指标端点""" return Response(content=generate_latest(), media_type="text/plain") @app.get("/health") async def health_check(): """健康检查端点""" return { "status": "healthy", "model_loaded": reranker_model is not None, "timestamp": time.time() } # 在重排序端点中添加监控 @app.post("/rerank") @REQUEST_LATENCY.time() async def rerank_documents(request: RerankRequest): REQUEST_COUNT.labels(method='POST', endpoint='/rerank').inc() # 原有逻辑...

5. 部署与测试

5.1 启动服务

创建一个启动脚本main.py：

if __name__ == "__main__": import uvicorn uvicorn.run( app, host="0.0.0.0", port=8000, workers=1, # 对于GPU服务，通常每个进程使用一个GPU timeout_keep_alive=60 )

使用以下命令启动服务：

python main.py

或者使用生产级服务器：

gunicorn main:app -w 1 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000 --timeout 120

5.2 测试API服务

使用curl测试API功能：

# 测试健康检查 curl http://localhost:8000/health # 测试重排序功能 curl -X POST "http://localhost:8000/rerank" \ -H "Content-Type: application/json" \ -d '{ "query": "机器学习的基本概念", "documents": [ "机器学习是人工智能的一个分支，专注于开发能够从数据中学习的系统", "深度学习是机器学习的一个子领域，使用神经网络处理复杂模式识别", "Python是一种流行的编程语言，广泛用于数据科学和机器学习" ], "top_k": 2 }'

你应该会得到类似这样的响应：

{ "results": [ { "document": "机器学习是人工智能的一个分支，专注于开发能够从数据中学习的系统", "score": 0.95, "rank": 1 }, { "document": "深度学习是机器学习的一个子领域，使用神经网络处理复杂模式识别", "score": 0.82, "rank": 2 } ], "processing_time": 0.45 }