通义千问3-Reranker-0.6B实战指南：与FastAPI集成构建RESTful rerank微服务

1. 项目概述

通义千问3-Reranker-0.6B是Qwen3 Embedding模型系列中的重排序专用模型，专门设计用于文本相关性排序任务。这个6亿参数的模型在保持轻量级的同时，提供了出色的多语言支持和长文本理解能力。

在实际应用中，我们经常需要将这样的AI模型集成到现有的系统中，而Gradio界面虽然友好，但更适合演示和测试。通过FastAPI构建RESTful微服务，我们可以获得更好的性能、更灵活的部署方式，以及更易于集成的API接口。

本文将手把手教你如何将Qwen3-Reranker-0.6B模型与FastAPI框架集成，构建一个生产级的重排序微服务。

2. 环境准备与依赖安装

2.1 基础环境要求

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

# Python版本要求
Python >= 3.8 (推荐Python 3.10)
PyTorch >= 2.0.0
Transformers >= 4.51.0

2.2 安装必要依赖

创建并激活虚拟环境后，安装所需依赖：

# 创建虚拟环境
python -m venv qwen3-reranker-env
source qwen3-reranker-env/bin/activate  # Linux/Mac
# 或者 .\qwen3-reranker-env\Scripts\activate  # Windows

# 安装核心依赖
pip install torch>=2.0.0 --index-url https://download.pytorch.org/whl/cu118
pip install transformers>=4.51.0
pip install fastapi>=0.104.0
pip install uvicorn>=0.24.0
pip install pydantic>=2.0.0
pip install accelerate>=0.24.0
pip install safetensors>=0.4.0

3. FastAPI服务核心代码实现

3.1 创建主应用文件

新建一个app.py文件，包含完整的FastAPI服务代码：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import torch
from transformers import AutoModelForSequenceClassification, AutoTokenizer
import logging
import time

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(
    title="Qwen3-Reranker-0.6B API",
    description="基于FastAPI的文本重排序微服务",
    version="1.0.0"
)

class RerankRequest(BaseModel):
    query: str
    documents: List[str]
    instruction: Optional[str] = None
    batch_size: Optional[int] = 8
    top_k: Optional[int] = None

class RerankResponse(BaseModel):
    results: List[dict]
    processing_time: float

# 全局变量存储模型和tokenizer
model = None
tokenizer = None
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

@app.on_event("startup")
async def load_model():
    """启动时加载模型"""
    global model, tokenizer
    try:
        logger.info("开始加载Qwen3-Reranker-0.6B模型...")
        start_time = time.time()
        
        model_path = "/root/ai-models/Qwen/Qwen3-Reranker-0___6B"
        
        # 加载tokenizer和模型
        tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
        model = AutoModelForSequenceClassification.from_pretrained(
            model_path,
            torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
            device_map="auto",
            trust_remote_code=True
        )
        
        model.eval()
        load_time = time.time() - start_time
        logger.info(f"模型加载完成，耗时: {load_time:.2f}秒")
        logger.info(f"使用设备: {device}")
        
    except Exception as e:
        logger.error(f"模型加载失败: {str(e)}")
        raise

def rerank_documents(query: str, documents: List[str], instruction: Optional[str] = None, batch_size: int = 8):
    """执行重排序的核心函数"""
    if model is None or tokenizer is None:
        raise HTTPException(status_code=503, detail="模型未加载完成")
    
    if not documents:
        return []
    
    # 构建输入对
    pairs = []
    for doc in documents:
        if instruction:
            text_pair = f"{instruction}\nQuery: {query}\nDocument: {doc}"
        else:
            text_pair = f"Query: {query}\nDocument: {doc}"
        pairs.append(text_pair)
    
    # 批量处理
    results = []
    for i in range(0, len(pairs), batch_size):
        batch_pairs = pairs[i:i + batch_size]
        
        # tokenize
        inputs = tokenizer(
            batch_pairs,
            padding=True,
            truncation=True,
            max_length=32768,
            return_tensors="pt"
        ).to(device)
        
        # 推理
        with torch.no_grad():
            outputs = model(**inputs)
            scores = outputs.logits[:, 0].cpu().numpy()
        
        # 收集结果
        for j, score in enumerate(scores):
            doc_index = i + j
            if doc_index < len(documents):
                results.append({
                    "document": documents[doc_index],
                    "score": float(score),
                    "rank": 0  # 后续统一排序
                })
    
    # 按分数排序
    results.sort(key=lambda x: x["score"], reverse=True)
    
    # 添加排名
    for rank, result in enumerate(results, 1):
        result["rank"] = rank
    
    return results

@app.post("/rerank", response_model=RerankResponse)
async def rerank_endpoint(request: RerankRequest):
    """重排序API端点"""
    start_time = time.time()
    
    try:
        # 验证输入
        if not request.query:
            raise HTTPException(status_code=400, detail="查询文本不能为空")
        
        if not request.documents:
            raise HTTPException(status_code=400, detail="文档列表不能为空")
        
        if len(request.documents) > 100:
            raise HTTPException(status_code=400, detail="单次最多处理100个文档")
        
        # 执行重排序
        results = rerank_documents(
            query=request.query,
            documents=request.documents,
            instruction=request.instruction,
            batch_size=request.batch_size
        )
        
        # 应用top_k过滤
        if request.top_k and request.top_k > 0:
            results = results[:request.top_k]
        
        processing_time = time.time() - start_time
        
        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)}")

@app.get("/health")
async def health_check():
    """健康检查端点"""
    return {
        "status": "healthy",
        "model_loaded": model is not None,
        "device": str(device)
    }

@app.get("/")
async def root():
    """根端点，返回API信息"""
    return {
        "message": "Qwen3-Reranker-0.6B API服务",
        "version": "1.0.0",
        "endpoints": {
            "rerank": "POST /rerank - 执行文本重排序",
            "health": "GET /health - 服务健康检查"
        }
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

3.2 创建配置文件

新建config.py文件用于配置管理：

import os
from typing import Optional

class Config:
    # 模型配置
    MODEL_PATH: str = os.getenv("MODEL_PATH", "/root/ai-models/Qwen/Qwen3-Reranker-0___6B")
    DEVICE: str = os.getenv("DEVICE", "auto")
    BATCH_SIZE: int = int(os.getenv("BATCH_SIZE", "8"))
    
    # 服务配置
    HOST: str = os.getenv("HOST", "0.0.0.0")
    PORT: int = int(os.getenv("PORT", "8000"))
    WORKERS: int = int(os.getenv("WORKERS", "1"))
    
    # 性能配置
    MAX_DOCUMENTS: int = int(os.getenv("MAX_DOCUMENTS", "100"))
    TIMEOUT: int = int(os.getenv("TIMEOUT", "300"))
    
    # 日志配置
    LOG_LEVEL: str = os.getenv("LOG_LEVEL", "INFO")

config = Config()

4. 部署与运行指南

4.1 启动FastAPI服务

有多种方式可以启动你的重排序服务：

方式一：直接运行Python脚本

python app.py

方式二：使用Uvicorn启动（推荐）

# 开发模式（带热重载）
uvicorn app:app --reload --host 0.0.0.0 --port 8000

# 生产模式（多worker）
uvicorn app:app --host 0.0.0.0 --port 8000 --workers 2 --timeout-keep-alive 300

4.2 创建启动脚本

创建start_fastapi.sh启动脚本：

#!/bin/bash

# Qwen3-Reranker FastAPI服务启动脚本

# 设置环境变量
export MODEL_PATH="/root/ai-models/Qwen/Qwen3-Reranker-0___6B"
export BATCH_SIZE=8
export PORT=8000
export WORKERS=1

# 激活虚拟环境
source /root/Qwen3-Reranker-0.6B/venv/bin/activate

# 启动服务
echo "启动Qwen3-Reranker FastAPI服务..."
uvicorn app:app --host 0.0.0.0 --port $PORT --workers $WORKERS

# 如果使用GPU，可以添加CUDA相关设置
# export CUDA_VISIBLE_DEVICES=0

给脚本添加执行权限并运行：

chmod +x start_fastapi.sh
./start_fastapi.sh

5. API使用示例

5.1 Python客户端调用示例

创建client_example.py文件：

import requests
import json
from typing import List, Optional

class QwenRerankerClient:
    def __init__(self, base_url: str = "http://localhost:8000"):
        self.base_url = base_url.rstrip('/')
    
    def rerank(self, 
               query: str, 
               documents: List[str], 
               instruction: Optional[str] = None,
               batch_size: int = 8,
               top_k: Optional[int] = None) -> dict:
        """
        调用重排序API
        
        Args:
            query: 查询文本
            documents: 候选文档列表
            instruction: 自定义指令（可选）
            batch_size: 批处理大小
            top_k: 返回前K个结果
            
        Returns:
            API响应结果
        """
        payload = {
            "query": query,
            "documents": documents,
            "instruction": instruction,
            "batch_size": batch_size,
            "top_k": top_k
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/rerank",
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"API调用失败: {e}")
            return None

# 使用示例
if __name__ == "__main__":
    client = QwenRerankerClient()
    
    # 示例1：英文查询
    english_result = client.rerank(
        query="What is the capital of China?",
        documents=[
            "Beijing is the capital of China.",
            "Gravity is a force that attracts two bodies towards each other.",
            "The sky appears blue because of Rayleigh scattering."
        ],
        instruction="Given a web search query, retrieve relevant passages that answer the query"
    )
    
    print("英文查询结果:")
    print(json.dumps(english_result, indent=2, ensure_ascii=False))
    
    # 示例2：中文查询
    chinese_result = client.rerank(
        query="解释量子力学",
        documents=[
            "量子力学是物理学的一个分支，主要研究微观粒子的运动规律。",
            "今天天气很好，适合外出游玩。",
            "苹果是一种常见的水果，富含维生素。"
        ],
        top_k=2  # 只返回前2个结果
    )
    
    print("\n中文查询结果:")
    print(json.dumps(chinese_result, indent=2, ensure_ascii=False))

5.2 命令行调用示例

使用curl命令测试API：

# 健康检查
curl http://localhost:8000/health

# 重排序请求
curl -X POST "http://localhost:8000/rerank" \
  -H "Content-Type: application/json" \
  -d '{
    "query": "What is machine learning?",
    "documents": [
      "Machine learning is a subset of artificial intelligence.",
      "The Eiffel Tower is in Paris, France.",
      "ML algorithms build models based on sample data."
    ],
    "batch_size": 4,
    "top_k": 2
  }'

6. 性能优化与最佳实践

6.1 批处理大小优化

根据你的硬件配置调整批处理大小：

# GPU内存充足（8GB+）
BATCH_SIZE = 16

# GPU内存一般（4-8GB）  
BATCH_SIZE = 8

# GPU内存较小或使用CPU
BATCH_SIZE = 4

6.2 自定义指令模板

为不同场景提供优化指令：

INSTRUCTION_TEMPLATES = {
    "web_search": "Given a web search query, retrieve relevant passages that answer the query",
    "legal": "Given a legal query, retrieve relevant legal documents and provisions",
    "academic": "Given an academic query, retrieve relevant research papers and articles",
    "code": "Given a code-related query, retrieve relevant code snippets and documentation",
    "general": "Given a query, retrieve the most relevant documents from the candidate set"
}

6.3 异步处理支持

对于高并发场景，可以考虑使用异步处理：

@app.post("/async-rerank")
async def async_rerank_endpoint(request: RerankRequest):
    """异步重排序端点（适用于高并发）"""
    # 将任务提交到线程池执行
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(
        None,  # 使用默认线程池
        lambda: rerank_documents(
            request.query, 
            request.documents, 
            request.instruction, 
            request.batch_size
        )
    )
    return {"results": result}

7. 监控与维护

7.1 添加性能监控

集成Prometheus监控：

from prometheus_fastapi_instrumentator import Instrumentator

# 添加性能监控
Instrumentator().instrument(app).expose(app)

7.2 日志配置优化

创建更详细的日志配置：

import logging
from logging.config import dictConfig

log_config = {
    "version": 1,
    "formatters": {
        "default": {
            "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
        }
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "formatter": "default",
            "level": "INFO"
        },
        "file": {
            "class": "logging.handlers.RotatingFileHandler",
            "formatter": "default",
            "filename": "reranker_service.log",
            "maxBytes": 10485760,  # 10MB
            "backupCount": 5,
            "level": "INFO"
        }
    },
    "root": {
        "handlers": ["console", "file"],
        "level": "INFO"
    }
}

dictConfig(log_config)

8. 总结

通过本文的实战指南，你已经学会了如何将Qwen3-Reranker-0.6B模型与FastAPI框架集成，构建一个生产级的重排序微服务。相比原始的Gradio界面，FastAPI版本提供了：

• 更好的性能：支持批处理、异步处理
• 更灵活的API：RESTful接口，易于集成
• 更强的稳定性：错误处理、健康检查、监控
• 更高的可配置性：支持自定义指令、批处理大小调整

这种集成方式特别适合需要将重排序能力嵌入到现有系统的场景，比如搜索引擎、推荐系统、问答系统等。FastAPI的自动文档生成功能（访问 http://localhost:8000/docs）也让API的使用和测试变得更加方便。

记得根据你的实际需求调整配置参数，特别是批处理大小和硬件配置，以达到最佳的性能效果。

---

获取更多AI镜像

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