当前位置：首页 > news >正文

Qwen3-Reranker-0.6B生产环境：Prometheus监控+日志分级+错误码体系

news 2026/3/26 20:14:25

Qwen3-Reranker-0.6B生产环境：Prometheus监控+日志分级+错误码体系

1. 项目概述

通义千问3-Reranker-0.6B是Qwen家族的最新专有模型，专门设计用于文本嵌入和排序任务。这个0.6B参数的模型在保持高性能的同时，提供了出色的多语言能力、长文本理解和推理技能。

在生产环境中部署这样的AI模型，仅仅让服务运行起来是远远不够的。我们需要建立完整的监控、日志和错误处理体系，确保服务的稳定性、可观测性和可维护性。本文将详细介绍如何在生产环境中为Qwen3-Reranker-0.6B构建专业的运维体系。

2. Prometheus监控体系搭建

2.1 监控指标设计

在生产环境中，我们需要监控以下几个关键维度：

服务健康指标：

服务存活状态（up/down）
服务启动时间（uptime）
端口监听状态

性能指标：

请求处理延迟（分位数：P50、P90、P99）
请求吞吐量（QPS）
批处理大小分布
模型推理时间

资源指标：

GPU内存使用率
GPU利用率
系统内存使用量
CPU使用率

2.2 Prometheus配置示例

# prometheus.yml global: scrape_interval: 15s evaluation_interval: 15s scrape_configs: - job_name: 'qwen-reranker' static_configs: - targets: ['localhost:8000'] # 监控暴露端口 metrics_path: '/metrics' scrape_interval: 10s

2.3 监控数据采集实现

from prometheus_client import start_http_server, Counter, Gauge, Histogram import time # 定义监控指标 REQUEST_COUNT = Counter('reranker_requests_total', 'Total requests') REQUEST_LATENCY = Histogram('reranker_request_latency_seconds', 'Request latency') BATCH_SIZE = Gauge('reranker_batch_size', 'Current batch size') GPU_MEMORY = Gauge('reranker_gpu_memory_usage', 'GPU memory usage in MB') ERROR_COUNT = Counter('reranker_errors_total', 'Total errors', ['error_type']) class MonitoringMiddleware: def __init__(self, app): self.app = app # 启动监控服务器 start_http_server(8000) def track_request(self, batch_size): """记录请求指标""" REQUEST_COUNT.inc() BATCH_SIZE.set(batch_size) return time.time() def track_response(self, start_time, success=True): """记录响应指标""" latency = time.time() - start_time REQUEST_LATENCY.observe(latency) if not success: ERROR_COUNT.labels(error_type='request_failed').inc()

3. 日志分级管理系统

3.1 日志级别定义

在生产环境中，合理的日志分级至关重要：

DEBUG：详细的调试信息，包括输入输出、中间结果
INFO：服务运行状态信息，如启动完成、请求处理
WARNING：不影响服务的异常情况，如参数校验失败
ERROR：服务错误，但服务仍可继续运行
CRITICAL：严重错误，可能导致服务不可用

3.2 结构化日志配置

import logging import json from datetime import datetime def setup_logging(): """配置结构化日志""" logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('/var/log/qwen-reranker/app.log'), logging.StreamHandler() ] ) # JSON格式日志用于ELK收集 json_handler = logging.FileHandler('/var/log/qwen-reranker/json.log') json_handler.setFormatter(JsonFormatter()) logging.getLogger().addHandler(json_handler) class JsonFormatter(logging.Formatter): def format(self, record): log_data = { 'timestamp': datetime.now().isoformat(), 'level': record.levelname, 'logger': record.name, 'message': record.getMessage(), 'module': record.module, 'function': record.funcName, 'line': record.lineno } return json.dumps(log_data)

3.3 关键日志点设计

class RerankerService: def __init__(self): self.logger = logging.getLogger(__name__) def process_request(self, query, documents, instruction=None): """处理重排序请求""" start_time = time.time() # 记录请求信息 self.logger.info( "Processing rerank request", extra={ 'query_length': len(query), 'doc_count': len(documents), 'has_instruction': instruction is not None } ) try: # 业务逻辑处理 result = self._rerank(query, documents, instruction) # 记录成功日志 self.logger.info( "Request processed successfully", extra={ 'processing_time': time.time() - start_time, 'result_count': len(result) } ) return result except Exception as e: # 记录错误日志 self.logger.error( "Request processing failed", extra={ 'error_type': type(e).__name__, 'error_message': str(e), 'processing_time': time.time() - start_time }, exc_info=True ) raise

4. 错误码体系设计

4.1 错误码分类

建立统一的错误码体系，便于问题定位和处理：

系统错误（10xx）：

1001：服务未就绪
1002：模型加载失败
1003：内存不足
1004：GPU资源不足

输入错误（20xx）：

2001：查询文本为空
2002：文档列表为空
2003：文档数量超限
2004：输入格式错误

处理错误（30xx）：

3001：模型推理失败
3002：批处理失败
3003：结果解析失败

4.2 错误处理实现

class ErrorCode: # 系统错误 SERVICE_NOT_READY = 1001 MODEL_LOAD_FAILED = 1002 OUT_OF_MEMORY = 1003 GPU_RESOURCE_UNAVAILABLE = 1004 # 输入错误 EMPTY_QUERY = 2001 EMPTY_DOCUMENTS = 2002 DOCUMENT_LIMIT_EXCEEDED = 2003 INVALID_INPUT_FORMAT = 2004 # 处理错误 INFERENCE_FAILED = 3001 BATCH_PROCESSING_FAILED = 3002 RESULT_PARSING_FAILED = 3003 class RerankerError(Exception): """重排序服务异常基类""" def __init__(self, code, message, details=None): self.code = code self.message = message self.details = details or {} super().__init__(f"[{code}] {message}") def validate_input(query, documents): """输入验证""" if not query or not query.strip(): raise RerankerError( ErrorCode.EMPTY_QUERY, "Query text cannot be empty" ) if not documents or not any(doc.strip() for doc in documents): raise RerankerError( ErrorCode.EMPTY_DOCUMENTS, "Documents list cannot be empty" ) if len(documents) > 100: raise RerankerError( ErrorCode.DOCUMENT_LIMIT_EXCEEDED, f"Document count exceeds limit: {len(documents)} > 100", {'document_count': len(documents)} )

5. 生产环境部署方案

5.1 Docker容器化部署

# Dockerfile FROM python:3.10-slim # 安装系统依赖 RUN apt-get update && apt-get install -y \ curl \ && rm -rf /var/lib/apt/lists/* # 创建工作目录 WORKDIR /app # 复制项目文件 COPY requirements.txt . COPY app.py . COPY start.sh . # 安装Python依赖 RUN pip install --no-cache-dir -r requirements.txt # 创建日志目录 RUN mkdir -p /var/log/qwen-reranker # 暴露端口 EXPOSE 7860 8000 # 启动服务 CMD ["./start.sh"]

5.2 Kubernetes部署配置

# deployment.yaml apiVersion: apps/v1 kind: Deployment metadata: name: qwen-reranker spec: replicas: 2 selector: matchLabels: app: qwen-reranker template: metadata: labels: app: qwen-reranker spec: containers: - name: reranker image: qwen-reranker:latest ports: - containerPort: 7860 - containerPort: 8000 resources: requests: memory: "4Gi" cpu: "2" nvidia.com/gpu: 1 limits: memory: "8Gi" cpu: "4" nvidia.com/gpu: 1 volumeMounts: - name: log-volume mountPath: /var/log/qwen-reranker volumes: - name: log-volume emptyDir: {}

5.3 健康检查配置

from flask import Flask, jsonify import psutil import torch app = Flask(__name__) @app.route('/health') def health_check(): """健康检查接口""" health_status = { 'status': 'healthy', 'timestamp': datetime.now().isoformat(), 'components': {} } # 检查GPU状态 if torch.cuda.is_available(): gpu_memory = torch.cuda.memory_allocated() / 1024**2 health_status['components']['gpu'] = { 'available': True, 'memory_used_mb': round(gpu_memory, 2) } else: health_status['components']['gpu'] = {'available': False} # 检查内存状态 memory = psutil.virtual_memory() health_status['components']['memory'] = { 'total_mb': round(memory.total / 1024**2, 2), 'available_mb': round(memory.available / 1024**2, 2), 'used_percent': round(memory.percent, 2) } # 检查服务状态 health_status['components']['service'] = { 'status': 'running', 'uptime': get_uptime() } return jsonify(health_status) def get_uptime(): """获取服务运行时间""" with open('/proc/uptime', 'r') as f: uptime_seconds = float(f.readline().split()[0]) return str(timedelta(seconds=uptime_seconds))