Qwen3-Reranker-0.6B保姆级教程模型服务健康检查端点与Prometheus监控集成1. 引言为什么你的AI服务需要“体检”想象一下你刚部署了一个强大的语义重排序服务它基于Qwen3-Reranker-0.6B模型能够精准理解用户查询和文档之间的语义关系。这个服务运行在Streamlit上界面友好功能强大。但几天后你发现用户反馈说“服务响应变慢了”或者“有时候会报错”。你登录服务器手忙脚乱地查看日志试图找出问题所在——这个过程是不是很熟悉这就是我们今天要解决的问题。一个健壮的AI服务不仅要功能强大还要“可观测”。你需要知道它是否健康、性能如何、资源消耗怎样。就像给汽车做定期保养我们需要给AI服务建立一套“体检”机制。本文将带你为Qwen3-Reranker-0.6B服务添加两个关键能力健康检查端点让外部系统如Kubernetes、负载均衡器能够快速判断服务是否存活、是否就绪Prometheus监控集成实时收集服务的性能指标让你对服务的运行状态了如指掌无论你是个人开发者还是团队运维这套监控方案都能让你的AI服务更加可靠、更容易维护。2. 准备工作理解我们的起点在开始改造之前我们先回顾一下Qwen3-Reranker-0.6B服务的基本架构。这有助于我们理解在哪里添加监控功能最合适。2.1 现有服务架构分析当前的服务基于Streamlit构建核心流程如下# 简化的核心逻辑示意 import streamlit as st from transformers import AutoModelForCausalLM, AutoTokenizer # 1. 加载模型使用缓存优化 st.cache_resource def load_model(): model AutoModelForCausalLM.from_pretrained(qwen/Qwen3-Reranker-0.6B) tokenizer AutoTokenizer.from_pretrained(qwen/Qwen3-Reranker-0.6B) return model, tokenizer # 2. 处理用户请求 def rerank_documents(query, documents): # 语义相关性计算逻辑 scores calculate_scores(query, documents) return sorted_results这个架构有几个特点单进程运行Streamlit默认以单进程方式运行同步处理用户请求按顺序处理模型缓存模型只加载一次后续推理复用2.2 需要监控的关键指标基于上述架构我们需要关注以下几个方面的指标监控维度具体指标为什么重要服务可用性服务是否存活、是否就绪确保服务能被正常访问性能表现请求处理时间、QPS每秒查询数了解服务处理能力资源使用CPU、内存、GPU使用率防止资源耗尽导致服务崩溃业务质量平均得分、排序准确率可选评估模型效果是否稳定接下来我们将分步实现这些监控能力。3. 第一步添加健康检查端点健康检查是服务监控的基础。它让外部系统能够知道“这个服务还活着吗能处理请求吗”3.1 为什么需要健康检查在实际部署中你的服务可能会遇到各种问题模型加载失败内存泄漏导致服务变慢依赖的服务如数据库不可用网络连接问题健康检查就像服务的“心跳”定期告诉外界“我还好可以工作”。3.2 实现健康检查的三种方式对于Streamlit应用我们有几种添加健康检查的方法方法一使用单独的HTTP服务器推荐这是最灵活的方式我们可以在同一个进程中启动一个简单的HTTP服务器专门处理健康检查请求。# health_check.py from http.server import HTTPServer, BaseHTTPRequestHandler import threading import json class HealthCheckHandler(BaseHTTPRequestHandler): def do_GET(self): if self.path /health: # 检查服务健康状态 is_healthy self.check_service_health() if is_healthy: self.send_response(200) self.send_header(Content-Type, application/json) self.end_headers() response { status: healthy, timestamp: time.time(), service: qwen3-reranker } self.wfile.write(json.dumps(response).encode()) else: self.send_response(503) # 服务不可用 self.send_header(Content-Type, application/json) self.end_headers() response { status: unhealthy, timestamp: time.time(), service: qwen3-reranker, error: Service not ready } self.wfile.write(json.dumps(response).encode()) else: self.send_response(404) self.end_headers() def check_service_health(self): 检查服务是否健康 try: # 检查1模型是否加载成功 if not hasattr(self.server, model_loaded) or not self.server.model_loaded: return False # 检查2内存使用是否正常 import psutil memory_percent psutil.Process().memory_percent() if memory_percent 90: # 内存使用超过90%认为不健康 return False return True except Exception: return False def log_message(self, format, *args): # 减少健康检查的日志输出避免日志污染 pass def start_health_check_server(port8081): 启动健康检查服务器 server HTTPServer((0.0.0.0, port), HealthCheckHandler) server.model_loaded False # 初始状态 # 在单独的线程中运行 thread threading.Thread(targetserver.serve_forever, daemonTrue) thread.start() print(fHealth check server started on port {port}) return server方法二使用Streamlit的页面路由如果你不想启动额外的HTTP服务器可以利用Streamlit的多页面功能# pages/health.py import streamlit as st import time import json st.set_page_config(page_titleHealth Check, layoutwide) # 检查服务状态 def check_health(): health_status { status: healthy, timestamp: time.time(), checks: [] } # 检查模型是否加载 try: if model in st.session_state and st.session_state.model is not None: health_status[checks].append({ name: model_loaded, status: pass }) else: health_status[checks].append({ name: model_loaded, status: fail, error: Model not loaded }) health_status[status] unhealthy except Exception as e: health_status[checks].append({ name: model_loaded, status: fail, error: str(e) }) health_status[status] unhealthy return health_status # 显示健康检查结果 health_data check_health() if st.query_params.get(format) json: st.json(health_data) else: st.title(服务健康状态) status_color if health_data[status] healthy else st.header(f{status_color} 状态: {health_data[status]}) for check in health_data[checks]: check_icon if check[status] pass else st.write(f{check_icon} {check[name]}) if error in check: st.error(f错误: {check[error]})方法三使用FastAPI集成高级方案如果你的服务需要更复杂的功能可以考虑将Streamlit与FastAPI结合# app.py from fastapi import FastAPI import uvicorn import threading import streamlit as st from streamlit.web import bootstrap app FastAPI(titleQwen3-Reranker Service) app.get(/health) async def health_check(): return { status: healthy, service: qwen3-reranker, version: 1.0.0 } app.get(/metrics) async def get_metrics(): # Prometheus格式的指标 metrics [ # HELP requests_total Total number of requests, # TYPE requests_total counter, requests_total 1234 ] return \n.join(metrics) def run_streamlit(): # 启动Streamlit应用 bootstrap.run() if __name__ __main__: # 在后台线程中启动Streamlit streamlit_thread threading.Thread(targetrun_streamlit, daemonTrue) streamlit_thread.start() # 启动FastAPI服务器 uvicorn.run(app, host0.0.0.0, port8080)3.3 集成到现有服务现在让我们把健康检查功能集成到Qwen3-Reranker服务中。我们将使用方法一因为它最简单、最通用。修改你的主应用文件比如app.py# app.py - 修改后的版本 import streamlit as st import threading import time from health_check import start_health_check_server # 启动健康检查服务器在8081端口 health_server start_health_check_server(port8081) # 设置页面配置 st.set_page_config( page_titleQwen3-Reranker Semantic Refiner, page_icon, layoutwide ) # 标记模型已加载在实际加载模型后设置 def on_model_loaded(): health_server.model_loaded True # 原有的模型加载和业务逻辑... st.cache_resource def load_model(): from transformers import AutoModelForCausalLM, AutoTokenizer model AutoModelForCausalLM.from_pretrained(qwen/Qwen3-Reranker-0.6B) tokenizer AutoTokenizer.from_pretrained(qwen/Qwen3-Reranker-0.6B) # 模型加载成功后更新健康检查状态 on_model_loaded() return model, tokenizer # 页面标题 st.title( Qwen3-Reranker Semantic Refiner) st.markdown(基于 **Qwen3-Reranker-0.6B** 大模型的语义重排序 Web 工具) # 原有的界面和业务逻辑继续...3.4 测试健康检查部署后你可以通过以下方式测试健康检查# 使用curl测试 curl http://localhost:8081/health # 预期响应 { status: healthy, timestamp: 1677654321.123, service: qwen3-reranker } # 如果服务不健康 { status: unhealthy, timestamp: 1677654321.123, service: qwen3-reranker, error: Service not ready }4. 第二步集成Prometheus监控健康检查告诉我们服务是否存活但我们需要更多细节服务处理了多少请求平均响应时间是多少内存使用情况如何这就是Prometheus的用武之地。4.1 Prometheus基础概念Prometheus是一个开源的监控和告警工具包。它的工作流程如下指标暴露你的应用通过HTTP端点暴露指标通常是/metrics指标抓取Prometheus服务器定期抓取这些指标指标存储抓取的指标存储在时间序列数据库中查询展示通过PromQL查询语言查询指标在Grafana中展示4.2 为Python应用添加Prometheus指标我们将使用prometheus_client库来暴露指标。首先安装它pip install prometheus-client然后在我们的服务中添加指标收集# metrics.py from prometheus_client import Counter, Histogram, Gauge, generate_latest, REGISTRY import time import psutil import threading # 定义指标 REQUEST_COUNT Counter( reranker_requests_total, Total number of requests, [method, endpoint, status] ) REQUEST_LATENCY Histogram( reranker_request_duration_seconds, Request latency in seconds, [method, endpoint] ) MODEL_INFERENCE_TIME Histogram( reranker_model_inference_seconds, Model inference time in seconds, [model_name] ) MEMORY_USAGE Gauge( reranker_memory_usage_bytes, Memory usage in bytes ) CPU_USAGE Gauge( reranker_cpu_usage_percent, CPU usage percentage ) ACTIVE_REQUESTS Gauge( reranker_active_requests, Number of active requests ) # 请求跟踪装饰器 def track_request(methodPOST, endpoint/rerank): def decorator(func): def wrapper(*args, **kwargs): start_time time.time() ACTIVE_REQUESTS.inc() try: result func(*args, **kwargs) status success return result except Exception as e: status error raise e finally: duration time.time() - start_time REQUEST_LATENCY.labels(methodmethod, endpointendpoint).observe(duration) REQUEST_COUNT.labels(methodmethod, endpointendpoint, statusstatus).inc() ACTIVE_REQUESTS.dec() return wrapper return decorator # 模型推理跟踪 def track_model_inference(model_nameqwen3-reranker-0.6b): def decorator(func): def wrapper(*args, **kwargs): start_time time.time() try: result func(*args, **kwargs) return result finally: duration time.time() - start_time MODEL_INFERENCE_TIME.labels(model_namemodel_name).observe(duration) return wrapper return decorator # 系统资源监控 def start_system_metrics_collection(interval10): 定期收集系统指标 def collect_metrics(): while True: # 内存使用 process psutil.Process() memory_info process.memory_info() MEMORY_USAGE.set(memory_info.rss) # RSS: 实际使用的物理内存 # CPU使用率 cpu_percent process.cpu_percent(interval1) CPU_USAGE.set(cpu_percent) time.sleep(interval) thread threading.Thread(targetcollect_metrics, daemonTrue) thread.start() # Prometheus指标端点处理器 class MetricsHandler: staticmethod def handle_request(handler): handler.send_response(200) handler.send_header(Content-Type, text/plain; version0.0.4) handler.end_headers() handler.wfile.write(generate_latest(REGISTRY))4.3 集成到主应用现在让我们把指标收集集成到主应用中# app.py - 完整版本 import streamlit as st import threading import time from http.server import HTTPServer, BaseHTTPRequestHandler import json # 导入指标模块 from metrics import ( track_request, track_model_inference, start_system_metrics_collection, MetricsHandler ) # 组合的健康检查和指标处理器 class CombinedHandler(BaseHTTPRequestHandler): def do_GET(self): if self.path /health: self.handle_health_check() elif self.path /metrics: MetricsHandler.handle_request(self) else: self.send_response(404) self.end_headers() def handle_health_check(self): 处理健康检查请求 try: # 检查服务状态 checks [] # 检查1模型是否加载 if hasattr(self, model_loaded) and self.model_loaded: checks.append({name: model_loaded, status: pass}) else: checks.append({ name: model_loaded, status: fail, error: Model not loaded }) # 检查2内存使用 import psutil memory_percent psutil.Process().memory_percent() if memory_percent 90: checks.append({name: memory_usage, status: pass}) else: checks.append({ name: memory_usage, status: fail, error: fMemory usage too high: {memory_percent}% }) # 确定整体状态 all_passed all(check[status] pass for check in checks) status healthy if all_passed else unhealthy self.send_response(200 if all_passed else 503) self.send_header(Content-Type, application/json) self.end_headers() response { status: status, timestamp: time.time(), service: qwen3-reranker, checks: checks } self.wfile.write(json.dumps(response, indent2).encode()) except Exception as e: self.send_response(500) self.send_header(Content-Type, application/json) self.end_headers() response { status: unhealthy, timestamp: time.time(), error: str(e) } self.wfile.write(json.dumps(response).encode()) def log_message(self, format, *args): # 减少日志输出 pass def start_monitoring_server(port8081): 启动监控服务器健康检查 Prometheus指标 server HTTPServer((0.0.0.0, port), CombinedHandler) server.model_loaded False # 启动系统指标收集 start_system_metrics_collection(interval10) # 在单独的线程中运行 thread threading.Thread(targetserver.serve_forever, daemonTrue) thread.start() print(fMonitoring server started on port {port}) return server # 启动监控服务器 monitoring_server start_monitoring_server(port8081) # Streamlit应用主逻辑 st.set_page_config( page_titleQwen3-Reranker Semantic Refiner, page_icon, layoutwide ) # 标记模型加载状态 def on_model_loaded(): monitoring_server.model_loaded True # 加载模型使用缓存 st.cache_resource def load_model(): from transformers import AutoModelForCausalLM, AutoTokenizer import torch st.info(正在加载Qwen3-Reranker-0.6B模型...) start_time time.time() model AutoModelForCausalLM.from_pretrained( qwen/Qwen3-Reranker-0.6B, torch_dtypetorch.float16 if torch.cuda.is_available() else torch.float32, device_mapauto if torch.cuda.is_available() else None ) tokenizer AutoTokenizer.from_pretrained(qwen/Qwen3-Reranker-0.6B) load_time time.time() - start_time st.success(f模型加载完成耗时 {load_time:.2f} 秒) # 更新监控状态 on_model_loaded() return model, tokenizer # 使用装饰器跟踪模型推理 track_model_inference(model_nameqwen3-reranker-0.6b) def calculate_scores(model, tokenizer, query, documents): 计算查询与文档的相关性分数 # 这里是你的评分逻辑 # 为了示例我们返回模拟分数 scores [] for doc in documents: # 实际应用中这里会调用模型进行推理 score len(query) / (len(doc) 1) # 模拟分数 scores.append(score) return scores # 使用装饰器跟踪请求 track_request(methodPOST, endpoint/rerank) def process_rerank_request(query, documents): 处理重排序请求 model, tokenizer load_model() scores calculate_scores(model, tokenizer, query, documents) # 排序结果 sorted_indices sorted(range(len(scores)), keylambda i: scores[i], reverseTrue) sorted_docs [documents[i] for i in sorted_indices] sorted_scores [scores[i] for i in sorted_indices] return sorted_docs, sorted_scores # 页面标题和描述 st.title( Qwen3-Reranker Semantic Refiner) st.markdown(基于 **Qwen3-Reranker-0.6B** 大模型的语义重排序 Web 工具) # 侧边栏监控信息 with st.sidebar: st.header( 服务监控) # 显示基本监控信息 if monitoring_server.model_loaded: st.success( 服务状态健康) else: st.warning( 服务状态加载中...) # 显示监控端点信息 st.info(监控端点) st.code(健康检查: http://localhost:8081/health) st.code(Prometheus指标: http://localhost:8081/metrics) # 主界面重排序功能 st.header( 语义重排序) # 输入查询 query st.text_input(输入查询 (Query):, placeholder例如人工智能的发展趋势) # 输入文档 st.subheader(输入文档 (Documents)) st.caption(每行输入一个文档至少输入2个文档进行比较) documents_text st.text_area( 文档列表:, height200, placeholder文档1人工智能是模拟人类智能的技术...\n文档2机器学习是人工智能的一个子领域...\n文档3深度学习使用神经网络进行特征学习... ) if st.button( 开始重排序, typeprimary): if not query: st.error(请输入查询内容) elif not documents_text: st.error(请输入至少一个文档) else: # 解析文档 documents [doc.strip() for doc in documents_text.split(\n) if doc.strip()] if len(documents) 2: st.error(请至少输入2个文档进行比较) else: with st.spinner(正在计算语义相关性...): # 处理请求会被track_request装饰器跟踪 sorted_docs, sorted_scores process_rerank_request(query, documents) # 显示结果 st.subheader( 排序结果) # 创建结果表格 result_data [] for i, (doc, score) in enumerate(zip(sorted_docs, sorted_scores), 1): result_data.append({ 排名: i, 得分: f{score:.4f}, 文档预览: doc[:100] ... if len(doc) 100 else doc }) st.table(result_data) # 显示详细信息 st.subheader( 文档详情) for i, (doc, score) in enumerate(zip(sorted_docs, sorted_scores), 1): with st.expander(f第{i}名 | 得分: {score:.4f}): st.text(doc)4.4 配置Prometheus抓取现在我们的服务已经暴露了指标需要配置Prometheus来抓取这些指标。创建Prometheus配置文件prometheus.yml# prometheus.yml global: scrape_interval: 15s # 每15秒抓取一次 evaluation_interval: 15s scrape_configs: - job_name: qwen3-reranker static_configs: - targets: [localhost:8081] # 你的服务地址 metrics_path: /metrics scrape_interval: 10s # 对这个job使用更频繁的抓取间隔 - job_name: prometheus static_configs: - targets: [localhost:9090]启动Prometheus# 下载Prometheus如果还没安装 wget https://github.com/prometheus/prometheus/releases/download/v2.45.0/prometheus-2.45.0.linux-amd64.tar.gz tar xvfz prometheus-*.tar.gz cd prometheus-* # 启动Prometheus ./prometheus --config.fileprometheus.yml访问http://localhost:9090你应该能看到Prometheus的Web界面。4.5 创建Grafana仪表板Prometheus擅长收集和查询指标但Grafana更适合可视化。让我们创建一个简单的仪表板。首先安装并启动Grafana# 使用Docker运行Grafana docker run -d -p 3000:3000 --name grafana grafana/grafana访问http://localhost:3000默认用户名密码是admin/admin。然后添加数据源点击Configuration → Data Sources选择Prometheus设置URL为http://localhost:9090如果你的Prometheus运行在其他地方修改相应地址点击Save Test现在创建仪表板添加以下面板面板1请求速率查询rate(reranker_requests_total[5m])图例{{method}} {{endpoint}}标题请求速率请求数/秒面板2请求延迟查询histogram_quantile(0.95, rate(reranker_request_duration_seconds_bucket[5m]))图例{{method}} {{endpoint}}标题95%请求延迟秒面板3内存使用查询reranker_memory_usage_bytes单位bytes → GB标题内存使用面板4CPU使用查询reranker_cpu_usage_percent标题CPU使用率%面板5活跃请求数查询reranker_active_requests标题当前活跃请求数面板6模型推理时间查询histogram_quantile(0.95, rate(reranker_model_inference_seconds_bucket[5m]))标题模型推理时间95%分位5. 第三步高级监控与告警基本的监控已经搭建好了但我们可以做得更好。让我们添加一些高级功能。5.1 自定义业务指标除了系统指标我们还可以添加业务相关的指标# business_metrics.py from prometheus_client import Counter, Histogram, Gauge # 文档长度分布 DOCUMENT_LENGTH Histogram( reranker_document_length_chars, Length of processed documents in characters, buckets[0, 100, 500, 1000, 5000, 10000] ) # 查询长度分布 QUERY_LENGTH Histogram( reranker_query_length_chars, Length of queries in characters, buckets[0, 10, 50, 100, 200, 500] ) # 得分分布 SCORE_DISTRIBUTION Histogram( reranker_score_values, Distribution of relevance scores, buckets[0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0] ) # 处理文档数量 DOCUMENTS_PROCESSED Counter( reranker_documents_processed_total, Total number of documents processed ) def track_document_metrics(documents): 跟踪文档相关指标 for doc in documents: doc_length len(doc) DOCUMENT_LENGTH.observe(doc_length) DOCUMENTS_PROCESSED.inc() def track_query_metrics(query): 跟踪查询相关指标 query_length len(query) QUERY_LENGTH.observe(query_length) def track_score_metrics(scores): 跟踪分数相关指标 for score in scores: SCORE_DISTRIBUTION.observe(score)在主应用中集成这些业务指标# 在process_rerank_request函数中添加 track_request(methodPOST, endpoint/rerank) def process_rerank_request(query, documents): 处理重排序请求 # 跟踪业务指标 from business_metrics import track_document_metrics, track_query_metrics, track_score_metrics track_query_metrics(query) track_document_metrics(documents) # 原有的处理逻辑... model, tokenizer load_model() scores calculate_scores(model, tokenizer, query, documents) # 跟踪分数指标 track_score_metrics(scores) # 排序和返回结果...5.2 设置告警规则当指标出现异常时我们希望收到通知。在Prometheus中配置告警规则创建alerts.yml# alerts.yml groups: - name: qwen3-reranker-alerts rules: # 高错误率告警 - alert: HighErrorRate expr: rate(reranker_requests_total{statuserror}[5m]) / rate(reranker_requests_total[5m]) 0.05 for: 2m labels: severity: warning annotations: summary: 高错误率检测到 description: 错误率超过5% (当前值: {{ $value }}) # 高延迟告警 - alert: HighLatency expr: histogram_quantile(0.95, rate(reranker_request_duration_seconds_bucket[5m])) 5 for: 2m labels: severity: warning annotations: summary: 高延迟检测到 description: 95%请求延迟超过5秒 (当前值: {{ $value }}秒) # 高内存使用告警 - alert: HighMemoryUsage expr: reranker_memory_usage_bytes 4e9 # 4GB for: 5m labels: severity: warning annotations: summary: 高内存使用检测到 description: 内存使用超过4GB (当前值: {{ $value | humanize1024 }}B) # 服务宕机告警 - alert: ServiceDown expr: up{jobqwen3-reranker} 0 for: 1m labels: severity: critical annotations: summary: 服务宕机 description: Qwen3-Reranker服务已宕机超过1分钟在prometheus.yml中引用告警规则# 在prometheus.yml中添加 rule_files: - alerts.yml alerting: alertmanagers: - static_configs: - targets: # - alertmanager:9093 # 如果你有Alertmanager5.3 使用Alertmanager发送通知如果你想要通过邮件、Slack等方式接收告警可以设置Alertmanager# alertmanager.yml global: smtp_smarthost: smtp.gmail.com:587 smtp_from: your-emailgmail.com smtp_auth_username: your-emailgmail.com smtp_auth_password: your-app-password route: group_by: [alertname] group_wait: 10s group_interval: 10s repeat_interval: 1h receiver: email-notifications receivers: - name: email-notifications email_configs: - to: your-emailgmail.com send_resolved: true启动Alertmanager./alertmanager --config.filealertmanager.yml6. 第四步部署与运维实践监控系统搭建好了现在让我们看看如何在实际环境中部署和运维。6.1 Docker化部署为了便于部署我们可以将整个服务Docker化# Dockerfile FROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y \ curl \ rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . # 安装Python依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 暴露端口 EXPOSE 8080 # Streamlit应用 EXPOSE 8081 # 监控端点 # 健康检查 HEALTHCHECK --interval30s --timeout3s --start-period5s --retries3 \ CMD curl -f http://localhost:8081/health || exit 1 # 启动命令 CMD [sh, -c, python app.py streamlit run app.py --server.port8080 --server.address0.0.0.0]创建requirements.txtstreamlit1.28.0 transformers4.35.0 torch2.0.0 prometheus-client0.19.0 psutil5.9.0构建和运行Docker镜像# 构建镜像 docker build -t qwen3-reranker:latest . # 运行容器 docker run -d \ -p 8080:8080 \ -p 8081:8081 \ --name qwen3-reranker \ qwen3-reranker:latest6.2 Kubernetes部署配置如果你在Kubernetes中运行可以创建以下配置# deployment.yaml apiVersion: apps/v1 kind: Deployment metadata: name: qwen3-reranker spec: replicas: 2 selector: matchLabels: app: qwen3-reranker template: metadata: labels: app: qwen3-reranker spec: containers: - name: reranker image: qwen3-reranker:latest ports: - containerPort: 8080 name: web - containerPort: 8081 name: metrics resources: requests: memory: 4Gi cpu: 1000m limits: memory: 8Gi cpu: 2000m livenessProbe: httpGet: path: /health port: 8081 initialDelaySeconds: 30 periodSeconds: 10 readinessProbe: httpGet: path: /health port: 8081 initialDelaySeconds: 5 periodSeconds: 5 --- # service.yaml apiVersion: v1 kind: Service metadata: name: qwen3-reranker spec: selector: app: qwen3-reranker ports: - port: 80 targetPort: 8080 name: web - port: 8081 targetPort: 8081 name: metrics type: ClusterIP --- # service-monitor.yaml (用于Prometheus Operator) apiVersion: monitoring.coreos.com/v1 kind: ServiceMonitor metadata: name: qwen3-reranker spec: selector: matchLabels: app: qwen3-reranker endpoints: - port: metrics interval: 15s path: /metrics6.3 性能优化建议基于监控数据你可以优化服务性能批处理请求如果同时有多个请求可以考虑批处理模型量化使用8位或4位量化减少内存使用缓存结果对相同的查询和文档缓存计算结果异步处理对于长文档使用异步处理避免阻塞# 示例添加结果缓存 from functools import lru_cache import hashlib lru_cache(maxsize1000) def get_cached_scores(query, documents_tuple): 缓存计算结果 # documents_tuple是文档元组因为列表不可哈希 documents list(documents_tuple) model, tokenizer load_model() return calculate_scores(model, tokenizer, query, documents) def process_with_cache(query, documents): 使用缓存处理请求 # 创建缓存键 documents_tuple tuple(documents) cache_key f{query}_{hashlib.md5(str(documents_tuple).encode()).hexdigest()} # 检查缓存 if hasattr(process_with_cache, cache): if cache_key in process_with_cache.cache: return process_with_cache.cache[cache_key] # 计算并缓存结果 result get_cached_scores(query, documents_tuple) if not hasattr(process_with_cache, cache): process_with_cache.cache {} process_with_cache.cache[cache_key] result return result7. 总结通过本教程我们为Qwen3-Reranker-0.6B语义重排序服务构建了一套完整的监控系统。让我们回顾一下关键收获7.1 核心成果健康检查端点让外部系统能够判断服务是否健康支持Kubernetes的存活性和就绪性探针Prometheus监控集成实时收集服务性能指标包括请求速率、延迟、资源使用等业务指标跟踪不仅监控系统状态还跟踪业务相关的指标如文档长度、分数分布等告警系统当服务出现异常时能够及时通知运维人员可视化仪表板通过Grafana直观展示服务运行状态7.2 实际价值这套监控系统为你带来了以下实际价值快速故障排查当服务出现问题时你可以立即查看相关指标快速定位问题原因容量规划通过监控资源使用趋势你可以合理规划服务器资源避免资源浪费或不足性能优化基于延迟和吞吐量数据你可以识别性能瓶颈并进行优化服务质量保证确保用户获得稳定、可靠的服务体验运维自动化减少人工检查的工作量让运维更加高效7.3 下一步建议如果你想要进一步深化监控能力可以考虑分布式追踪集成Jaeger或Zipkin跟踪请求在微服务间的流转日志聚合使用ELK栈Elasticsearch, Logstash, Kibana集中管理日志自定义指标根据你的业务需求添加更多有意义的业务指标自动化扩缩容基于监控指标自动调整服务实例数量A/B测试监控如果你有多个模型版本可以监控不同版本的表现监控不是一次性的工作而是一个持续的过程。随着业务的发展你需要不断调整和优化监控策略。但有了今天搭建的基础你已经迈出了重要的一步。记住好的监控系统就像汽车的仪表盘——它不能防止事故但能让你在问题发生前及时察觉在问题发生时快速响应。现在你的Qwen3-Reranker服务有了这样的仪表盘可以更加自信地服务用户了。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
Qwen3-Reranker-0.6B保姆级教程：模型服务健康检查端点与Prometheus监控集成
Qwen3-Reranker-0.6B保姆级教程模型服务健康检查端点与Prometheus监控集成1. 引言为什么你的AI服务需要“体检”想象一下你刚部署了一个强大的语义重排序服务它基于Qwen3-Reranker-0.6B模型能够精准理解用户查询和文档之间的语义关系。这个服务运行在Streamlit上界面友好功能强大。但几天后你发现用户反馈说“服务响应变慢了”或者“有时候会报错”。你登录服务器手忙脚乱地查看日志试图找出问题所在——这个过程是不是很熟悉这就是我们今天要解决的问题。一个健壮的AI服务不仅要功能强大还要“可观测”。你需要知道它是否健康、性能如何、资源消耗怎样。就像给汽车做定期保养我们需要给AI服务建立一套“体检”机制。本文将带你为Qwen3-Reranker-0.6B服务添加两个关键能力健康检查端点让外部系统如Kubernetes、负载均衡器能够快速判断服务是否存活、是否就绪Prometheus监控集成实时收集服务的性能指标让你对服务的运行状态了如指掌无论你是个人开发者还是团队运维这套监控方案都能让你的AI服务更加可靠、更容易维护。2. 准备工作理解我们的起点在开始改造之前我们先回顾一下Qwen3-Reranker-0.6B服务的基本架构。这有助于我们理解在哪里添加监控功能最合适。2.1 现有服务架构分析当前的服务基于Streamlit构建核心流程如下# 简化的核心逻辑示意 import streamlit as st from transformers import AutoModelForCausalLM, AutoTokenizer # 1. 加载模型使用缓存优化 st.cache_resource def load_model(): model AutoModelForCausalLM.from_pretrained(qwen/Qwen3-Reranker-0.6B) tokenizer AutoTokenizer.from_pretrained(qwen/Qwen3-Reranker-0.6B) return model, tokenizer # 2. 处理用户请求 def rerank_documents(query, documents): # 语义相关性计算逻辑 scores calculate_scores(query, documents) return sorted_results这个架构有几个特点单进程运行Streamlit默认以单进程方式运行同步处理用户请求按顺序处理模型缓存模型只加载一次后续推理复用2.2 需要监控的关键指标基于上述架构我们需要关注以下几个方面的指标监控维度具体指标为什么重要服务可用性服务是否存活、是否就绪确保服务能被正常访问性能表现请求处理时间、QPS每秒查询数了解服务处理能力资源使用CPU、内存、GPU使用率防止资源耗尽导致服务崩溃业务质量平均得分、排序准确率可选评估模型效果是否稳定接下来我们将分步实现这些监控能力。3. 第一步添加健康检查端点健康检查是服务监控的基础。它让外部系统能够知道“这个服务还活着吗能处理请求吗”3.1 为什么需要健康检查在实际部署中你的服务可能会遇到各种问题模型加载失败内存泄漏导致服务变慢依赖的服务如数据库不可用网络连接问题健康检查就像服务的“心跳”定期告诉外界“我还好可以工作”。3.2 实现健康检查的三种方式对于Streamlit应用我们有几种添加健康检查的方法方法一使用单独的HTTP服务器推荐这是最灵活的方式我们可以在同一个进程中启动一个简单的HTTP服务器专门处理健康检查请求。# health_check.py from http.server import HTTPServer, BaseHTTPRequestHandler import threading import json class HealthCheckHandler(BaseHTTPRequestHandler): def do_GET(self): if self.path /health: # 检查服务健康状态 is_healthy self.check_service_health() if is_healthy: self.send_response(200) self.send_header(Content-Type, application/json) self.end_headers() response { status: healthy, timestamp: time.time(), service: qwen3-reranker } self.wfile.write(json.dumps(response).encode()) else: self.send_response(503) # 服务不可用 self.send_header(Content-Type, application/json) self.end_headers() response { status: unhealthy, timestamp: time.time(), service: qwen3-reranker, error: Service not ready } self.wfile.write(json.dumps(response).encode()) else: self.send_response(404) self.end_headers() def check_service_health(self): 检查服务是否健康 try: # 检查1模型是否加载成功 if not hasattr(self.server, model_loaded) or not self.server.model_loaded: return False # 检查2内存使用是否正常 import psutil memory_percent psutil.Process().memory_percent() if memory_percent 90: # 内存使用超过90%认为不健康 return False return True except Exception: return False def log_message(self, format, *args): # 减少健康检查的日志输出避免日志污染 pass def start_health_check_server(port8081): 启动健康检查服务器 server HTTPServer((0.0.0.0, port), HealthCheckHandler) server.model_loaded False # 初始状态 # 在单独的线程中运行 thread threading.Thread(targetserver.serve_forever, daemonTrue) thread.start() print(fHealth check server started on port {port}) return server方法二使用Streamlit的页面路由如果你不想启动额外的HTTP服务器可以利用Streamlit的多页面功能# pages/health.py import streamlit as st import time import json st.set_page_config(page_titleHealth Check, layoutwide) # 检查服务状态 def check_health(): health_status { status: healthy, timestamp: time.time(), checks: [] } # 检查模型是否加载 try: if model in st.session_state and st.session_state.model is not None: health_status[checks].append({ name: model_loaded, status: pass }) else: health_status[checks].append({ name: model_loaded, status: fail, error: Model not loaded }) health_status[status] unhealthy except Exception as e: health_status[checks].append({ name: model_loaded, status: fail, error: str(e) }) health_status[status] unhealthy return health_status # 显示健康检查结果 health_data check_health() if st.query_params.get(format) json: st.json(health_data) else: st.title(服务健康状态) status_color if health_data[status] healthy else st.header(f{status_color} 状态: {health_data[status]}) for check in health_data[checks]: check_icon if check[status] pass else st.write(f{check_icon} {check[name]}) if error in check: st.error(f错误: {check[error]})方法三使用FastAPI集成高级方案如果你的服务需要更复杂的功能可以考虑将Streamlit与FastAPI结合# app.py from fastapi import FastAPI import uvicorn import threading import streamlit as st from streamlit.web import bootstrap app FastAPI(titleQwen3-Reranker Service) app.get(/health) async def health_check(): return { status: healthy, service: qwen3-reranker, version: 1.0.0 } app.get(/metrics) async def get_metrics(): # Prometheus格式的指标 metrics [ # HELP requests_total Total number of requests, # TYPE requests_total counter, requests_total 1234 ] return \n.join(metrics) def run_streamlit(): # 启动Streamlit应用 bootstrap.run() if __name__ __main__: # 在后台线程中启动Streamlit streamlit_thread threading.Thread(targetrun_streamlit, daemonTrue) streamlit_thread.start() # 启动FastAPI服务器 uvicorn.run(app, host0.0.0.0, port8080)3.3 集成到现有服务现在让我们把健康检查功能集成到Qwen3-Reranker服务中。我们将使用方法一因为它最简单、最通用。修改你的主应用文件比如app.py# app.py - 修改后的版本 import streamlit as st import threading import time from health_check import start_health_check_server # 启动健康检查服务器在8081端口 health_server start_health_check_server(port8081) # 设置页面配置 st.set_page_config( page_titleQwen3-Reranker Semantic Refiner, page_icon, layoutwide ) # 标记模型已加载在实际加载模型后设置 def on_model_loaded(): health_server.model_loaded True # 原有的模型加载和业务逻辑... st.cache_resource def load_model(): from transformers import AutoModelForCausalLM, AutoTokenizer model AutoModelForCausalLM.from_pretrained(qwen/Qwen3-Reranker-0.6B) tokenizer AutoTokenizer.from_pretrained(qwen/Qwen3-Reranker-0.6B) # 模型加载成功后更新健康检查状态 on_model_loaded() return model, tokenizer # 页面标题 st.title( Qwen3-Reranker Semantic Refiner) st.markdown(基于 **Qwen3-Reranker-0.6B** 大模型的语义重排序 Web 工具) # 原有的界面和业务逻辑继续...3.4 测试健康检查部署后你可以通过以下方式测试健康检查# 使用curl测试 curl http://localhost:8081/health # 预期响应 { status: healthy, timestamp: 1677654321.123, service: qwen3-reranker } # 如果服务不健康 { status: unhealthy, timestamp: 1677654321.123, service: qwen3-reranker, error: Service not ready }4. 第二步集成Prometheus监控健康检查告诉我们服务是否存活但我们需要更多细节服务处理了多少请求平均响应时间是多少内存使用情况如何这就是Prometheus的用武之地。4.1 Prometheus基础概念Prometheus是一个开源的监控和告警工具包。它的工作流程如下指标暴露你的应用通过HTTP端点暴露指标通常是/metrics指标抓取Prometheus服务器定期抓取这些指标指标存储抓取的指标存储在时间序列数据库中查询展示通过PromQL查询语言查询指标在Grafana中展示4.2 为Python应用添加Prometheus指标我们将使用prometheus_client库来暴露指标。首先安装它pip install prometheus-client然后在我们的服务中添加指标收集# metrics.py from prometheus_client import Counter, Histogram, Gauge, generate_latest, REGISTRY import time import psutil import threading # 定义指标 REQUEST_COUNT Counter( reranker_requests_total, Total number of requests, [method, endpoint, status] ) REQUEST_LATENCY Histogram( reranker_request_duration_seconds, Request latency in seconds, [method, endpoint] ) MODEL_INFERENCE_TIME Histogram( reranker_model_inference_seconds, Model inference time in seconds, [model_name] ) MEMORY_USAGE Gauge( reranker_memory_usage_bytes, Memory usage in bytes ) CPU_USAGE Gauge( reranker_cpu_usage_percent, CPU usage percentage ) ACTIVE_REQUESTS Gauge( reranker_active_requests, Number of active requests ) # 请求跟踪装饰器 def track_request(methodPOST, endpoint/rerank): def decorator(func): def wrapper(*args, **kwargs): start_time time.time() ACTIVE_REQUESTS.inc() try: result func(*args, **kwargs) status success return result except Exception as e: status error raise e finally: duration time.time() - start_time REQUEST_LATENCY.labels(methodmethod, endpointendpoint).observe(duration) REQUEST_COUNT.labels(methodmethod, endpointendpoint, statusstatus).inc() ACTIVE_REQUESTS.dec() return wrapper return decorator # 模型推理跟踪 def track_model_inference(model_nameqwen3-reranker-0.6b): def decorator(func): def wrapper(*args, **kwargs): start_time time.time() try: result func(*args, **kwargs) return result finally: duration time.time() - start_time MODEL_INFERENCE_TIME.labels(model_namemodel_name).observe(duration) return wrapper return decorator # 系统资源监控 def start_system_metrics_collection(interval10): 定期收集系统指标 def collect_metrics(): while True: # 内存使用 process psutil.Process() memory_info process.memory_info() MEMORY_USAGE.set(memory_info.rss) # RSS: 实际使用的物理内存 # CPU使用率 cpu_percent process.cpu_percent(interval1) CPU_USAGE.set(cpu_percent) time.sleep(interval) thread threading.Thread(targetcollect_metrics, daemonTrue) thread.start() # Prometheus指标端点处理器 class MetricsHandler: staticmethod def handle_request(handler): handler.send_response(200) handler.send_header(Content-Type, text/plain; version0.0.4) handler.end_headers() handler.wfile.write(generate_latest(REGISTRY))4.3 集成到主应用现在让我们把指标收集集成到主应用中# app.py - 完整版本 import streamlit as st import threading import time from http.server import HTTPServer, BaseHTTPRequestHandler import json # 导入指标模块 from metrics import ( track_request, track_model_inference, start_system_metrics_collection, MetricsHandler ) # 组合的健康检查和指标处理器 class CombinedHandler(BaseHTTPRequestHandler): def do_GET(self): if self.path /health: self.handle_health_check() elif self.path /metrics: MetricsHandler.handle_request(self) else: self.send_response(404) self.end_headers() def handle_health_check(self): 处理健康检查请求 try: # 检查服务状态 checks [] # 检查1模型是否加载 if hasattr(self, model_loaded) and self.model_loaded: checks.append({name: model_loaded, status: pass}) else: checks.append({ name: model_loaded, status: fail, error: Model not loaded }) # 检查2内存使用 import psutil memory_percent psutil.Process().memory_percent() if memory_percent 90: checks.append({name: memory_usage, status: pass}) else: checks.append({ name: memory_usage, status: fail, error: fMemory usage too high: {memory_percent}% }) # 确定整体状态 all_passed all(check[status] pass for check in checks) status healthy if all_passed else unhealthy self.send_response(200 if all_passed else 503) self.send_header(Content-Type, application/json) self.end_headers() response { status: status, timestamp: time.time(), service: qwen3-reranker, checks: checks } self.wfile.write(json.dumps(response, indent2).encode()) except Exception as e: self.send_response(500) self.send_header(Content-Type, application/json) self.end_headers() response { status: unhealthy, timestamp: time.time(), error: str(e) } self.wfile.write(json.dumps(response).encode()) def log_message(self, format, *args): # 减少日志输出 pass def start_monitoring_server(port8081): 启动监控服务器健康检查 Prometheus指标 server HTTPServer((0.0.0.0, port), CombinedHandler) server.model_loaded False # 启动系统指标收集 start_system_metrics_collection(interval10) # 在单独的线程中运行 thread threading.Thread(targetserver.serve_forever, daemonTrue) thread.start() print(fMonitoring server started on port {port}) return server # 启动监控服务器 monitoring_server start_monitoring_server(port8081) # Streamlit应用主逻辑 st.set_page_config( page_titleQwen3-Reranker Semantic Refiner, page_icon, layoutwide ) # 标记模型加载状态 def on_model_loaded(): monitoring_server.model_loaded True # 加载模型使用缓存 st.cache_resource def load_model(): from transformers import AutoModelForCausalLM, AutoTokenizer import torch st.info(正在加载Qwen3-Reranker-0.6B模型...) start_time time.time() model AutoModelForCausalLM.from_pretrained( qwen/Qwen3-Reranker-0.6B, torch_dtypetorch.float16 if torch.cuda.is_available() else torch.float32, device_mapauto if torch.cuda.is_available() else None ) tokenizer AutoTokenizer.from_pretrained(qwen/Qwen3-Reranker-0.6B) load_time time.time() - start_time st.success(f模型加载完成耗时 {load_time:.2f} 秒) # 更新监控状态 on_model_loaded() return model, tokenizer # 使用装饰器跟踪模型推理 track_model_inference(model_nameqwen3-reranker-0.6b) def calculate_scores(model, tokenizer, query, documents): 计算查询与文档的相关性分数 # 这里是你的评分逻辑 # 为了示例我们返回模拟分数 scores [] for doc in documents: # 实际应用中这里会调用模型进行推理 score len(query) / (len(doc) 1) # 模拟分数 scores.append(score) return scores # 使用装饰器跟踪请求 track_request(methodPOST, endpoint/rerank) def process_rerank_request(query, documents): 处理重排序请求 model, tokenizer load_model() scores calculate_scores(model, tokenizer, query, documents) # 排序结果 sorted_indices sorted(range(len(scores)), keylambda i: scores[i], reverseTrue) sorted_docs [documents[i] for i in sorted_indices] sorted_scores [scores[i] for i in sorted_indices] return sorted_docs, sorted_scores # 页面标题和描述 st.title( Qwen3-Reranker Semantic Refiner) st.markdown(基于 **Qwen3-Reranker-0.6B** 大模型的语义重排序 Web 工具) # 侧边栏监控信息 with st.sidebar: st.header( 服务监控) # 显示基本监控信息 if monitoring_server.model_loaded: st.success( 服务状态健康) else: st.warning( 服务状态加载中...) # 显示监控端点信息 st.info(监控端点) st.code(健康检查: http://localhost:8081/health) st.code(Prometheus指标: http://localhost:8081/metrics) # 主界面重排序功能 st.header( 语义重排序) # 输入查询 query st.text_input(输入查询 (Query):, placeholder例如人工智能的发展趋势) # 输入文档 st.subheader(输入文档 (Documents)) st.caption(每行输入一个文档至少输入2个文档进行比较) documents_text st.text_area( 文档列表:, height200, placeholder文档1人工智能是模拟人类智能的技术...\n文档2机器学习是人工智能的一个子领域...\n文档3深度学习使用神经网络进行特征学习... ) if st.button( 开始重排序, typeprimary): if not query: st.error(请输入查询内容) elif not documents_text: st.error(请输入至少一个文档) else: # 解析文档 documents [doc.strip() for doc in documents_text.split(\n) if doc.strip()] if len(documents) 2: st.error(请至少输入2个文档进行比较) else: with st.spinner(正在计算语义相关性...): # 处理请求会被track_request装饰器跟踪 sorted_docs, sorted_scores process_rerank_request(query, documents) # 显示结果 st.subheader( 排序结果) # 创建结果表格 result_data [] for i, (doc, score) in enumerate(zip(sorted_docs, sorted_scores), 1): result_data.append({ 排名: i, 得分: f{score:.4f}, 文档预览: doc[:100] ... if len(doc) 100 else doc }) st.table(result_data) # 显示详细信息 st.subheader( 文档详情) for i, (doc, score) in enumerate(zip(sorted_docs, sorted_scores), 1): with st.expander(f第{i}名 | 得分: {score:.4f}): st.text(doc)4.4 配置Prometheus抓取现在我们的服务已经暴露了指标需要配置Prometheus来抓取这些指标。创建Prometheus配置文件prometheus.yml# prometheus.yml global: scrape_interval: 15s # 每15秒抓取一次 evaluation_interval: 15s scrape_configs: - job_name: qwen3-reranker static_configs: - targets: [localhost:8081] # 你的服务地址 metrics_path: /metrics scrape_interval: 10s # 对这个job使用更频繁的抓取间隔 - job_name: prometheus static_configs: - targets: [localhost:9090]启动Prometheus# 下载Prometheus如果还没安装 wget https://github.com/prometheus/prometheus/releases/download/v2.45.0/prometheus-2.45.0.linux-amd64.tar.gz tar xvfz prometheus-*.tar.gz cd prometheus-* # 启动Prometheus ./prometheus --config.fileprometheus.yml访问http://localhost:9090你应该能看到Prometheus的Web界面。4.5 创建Grafana仪表板Prometheus擅长收集和查询指标但Grafana更适合可视化。让我们创建一个简单的仪表板。首先安装并启动Grafana# 使用Docker运行Grafana docker run -d -p 3000:3000 --name grafana grafana/grafana访问http://localhost:3000默认用户名密码是admin/admin。然后添加数据源点击Configuration → Data Sources选择Prometheus设置URL为http://localhost:9090如果你的Prometheus运行在其他地方修改相应地址点击Save Test现在创建仪表板添加以下面板面板1请求速率查询rate(reranker_requests_total[5m])图例{{method}} {{endpoint}}标题请求速率请求数/秒面板2请求延迟查询histogram_quantile(0.95, rate(reranker_request_duration_seconds_bucket[5m]))图例{{method}} {{endpoint}}标题95%请求延迟秒面板3内存使用查询reranker_memory_usage_bytes单位bytes → GB标题内存使用面板4CPU使用查询reranker_cpu_usage_percent标题CPU使用率%面板5活跃请求数查询reranker_active_requests标题当前活跃请求数面板6模型推理时间查询histogram_quantile(0.95, rate(reranker_model_inference_seconds_bucket[5m]))标题模型推理时间95%分位5. 第三步高级监控与告警基本的监控已经搭建好了但我们可以做得更好。让我们添加一些高级功能。5.1 自定义业务指标除了系统指标我们还可以添加业务相关的指标# business_metrics.py from prometheus_client import Counter, Histogram, Gauge # 文档长度分布 DOCUMENT_LENGTH Histogram( reranker_document_length_chars, Length of processed documents in characters, buckets[0, 100, 500, 1000, 5000, 10000] ) # 查询长度分布 QUERY_LENGTH Histogram( reranker_query_length_chars, Length of queries in characters, buckets[0, 10, 50, 100, 200, 500] ) # 得分分布 SCORE_DISTRIBUTION Histogram( reranker_score_values, Distribution of relevance scores, buckets[0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0] ) # 处理文档数量 DOCUMENTS_PROCESSED Counter( reranker_documents_processed_total, Total number of documents processed ) def track_document_metrics(documents): 跟踪文档相关指标 for doc in documents: doc_length len(doc) DOCUMENT_LENGTH.observe(doc_length) DOCUMENTS_PROCESSED.inc() def track_query_metrics(query): 跟踪查询相关指标 query_length len(query) QUERY_LENGTH.observe(query_length) def track_score_metrics(scores): 跟踪分数相关指标 for score in scores: SCORE_DISTRIBUTION.observe(score)在主应用中集成这些业务指标# 在process_rerank_request函数中添加 track_request(methodPOST, endpoint/rerank) def process_rerank_request(query, documents): 处理重排序请求 # 跟踪业务指标 from business_metrics import track_document_metrics, track_query_metrics, track_score_metrics track_query_metrics(query) track_document_metrics(documents) # 原有的处理逻辑... model, tokenizer load_model() scores calculate_scores(model, tokenizer, query, documents) # 跟踪分数指标 track_score_metrics(scores) # 排序和返回结果...5.2 设置告警规则当指标出现异常时我们希望收到通知。在Prometheus中配置告警规则创建alerts.yml# alerts.yml groups: - name: qwen3-reranker-alerts rules: # 高错误率告警 - alert: HighErrorRate expr: rate(reranker_requests_total{statuserror}[5m]) / rate(reranker_requests_total[5m]) 0.05 for: 2m labels: severity: warning annotations: summary: 高错误率检测到 description: 错误率超过5% (当前值: {{ $value }}) # 高延迟告警 - alert: HighLatency expr: histogram_quantile(0.95, rate(reranker_request_duration_seconds_bucket[5m])) 5 for: 2m labels: severity: warning annotations: summary: 高延迟检测到 description: 95%请求延迟超过5秒 (当前值: {{ $value }}秒) # 高内存使用告警 - alert: HighMemoryUsage expr: reranker_memory_usage_bytes 4e9 # 4GB for: 5m labels: severity: warning annotations: summary: 高内存使用检测到 description: 内存使用超过4GB (当前值: {{ $value | humanize1024 }}B) # 服务宕机告警 - alert: ServiceDown expr: up{jobqwen3-reranker} 0 for: 1m labels: severity: critical annotations: summary: 服务宕机 description: Qwen3-Reranker服务已宕机超过1分钟在prometheus.yml中引用告警规则# 在prometheus.yml中添加 rule_files: - alerts.yml alerting: alertmanagers: - static_configs: - targets: # - alertmanager:9093 # 如果你有Alertmanager5.3 使用Alertmanager发送通知如果你想要通过邮件、Slack等方式接收告警可以设置Alertmanager# alertmanager.yml global: smtp_smarthost: smtp.gmail.com:587 smtp_from: your-emailgmail.com smtp_auth_username: your-emailgmail.com smtp_auth_password: your-app-password route: group_by: [alertname] group_wait: 10s group_interval: 10s repeat_interval: 1h receiver: email-notifications receivers: - name: email-notifications email_configs: - to: your-emailgmail.com send_resolved: true启动Alertmanager./alertmanager --config.filealertmanager.yml6. 第四步部署与运维实践监控系统搭建好了现在让我们看看如何在实际环境中部署和运维。6.1 Docker化部署为了便于部署我们可以将整个服务Docker化# Dockerfile FROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y \ curl \ rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . # 安装Python依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 暴露端口 EXPOSE 8080 # Streamlit应用 EXPOSE 8081 # 监控端点 # 健康检查 HEALTHCHECK --interval30s --timeout3s --start-period5s --retries3 \ CMD curl -f http://localhost:8081/health || exit 1 # 启动命令 CMD [sh, -c, python app.py streamlit run app.py --server.port8080 --server.address0.0.0.0]创建requirements.txtstreamlit1.28.0 transformers4.35.0 torch2.0.0 prometheus-client0.19.0 psutil5.9.0构建和运行Docker镜像# 构建镜像 docker build -t qwen3-reranker:latest . # 运行容器 docker run -d \ -p 8080:8080 \ -p 8081:8081 \ --name qwen3-reranker \ qwen3-reranker:latest6.2 Kubernetes部署配置如果你在Kubernetes中运行可以创建以下配置# deployment.yaml apiVersion: apps/v1 kind: Deployment metadata: name: qwen3-reranker spec: replicas: 2 selector: matchLabels: app: qwen3-reranker template: metadata: labels: app: qwen3-reranker spec: containers: - name: reranker image: qwen3-reranker:latest ports: - containerPort: 8080 name: web - containerPort: 8081 name: metrics resources: requests: memory: 4Gi cpu: 1000m limits: memory: 8Gi cpu: 2000m livenessProbe: httpGet: path: /health port: 8081 initialDelaySeconds: 30 periodSeconds: 10 readinessProbe: httpGet: path: /health port: 8081 initialDelaySeconds: 5 periodSeconds: 5 --- # service.yaml apiVersion: v1 kind: Service metadata: name: qwen3-reranker spec: selector: app: qwen3-reranker ports: - port: 80 targetPort: 8080 name: web - port: 8081 targetPort: 8081 name: metrics type: ClusterIP --- # service-monitor.yaml (用于Prometheus Operator) apiVersion: monitoring.coreos.com/v1 kind: ServiceMonitor metadata: name: qwen3-reranker spec: selector: matchLabels: app: qwen3-reranker endpoints: - port: metrics interval: 15s path: /metrics6.3 性能优化建议基于监控数据你可以优化服务性能批处理请求如果同时有多个请求可以考虑批处理模型量化使用8位或4位量化减少内存使用缓存结果对相同的查询和文档缓存计算结果异步处理对于长文档使用异步处理避免阻塞# 示例添加结果缓存 from functools import lru_cache import hashlib lru_cache(maxsize1000) def get_cached_scores(query, documents_tuple): 缓存计算结果 # documents_tuple是文档元组因为列表不可哈希 documents list(documents_tuple) model, tokenizer load_model() return calculate_scores(model, tokenizer, query, documents) def process_with_cache(query, documents): 使用缓存处理请求 # 创建缓存键 documents_tuple tuple(documents) cache_key f{query}_{hashlib.md5(str(documents_tuple).encode()).hexdigest()} # 检查缓存 if hasattr(process_with_cache, cache): if cache_key in process_with_cache.cache: return process_with_cache.cache[cache_key] # 计算并缓存结果 result get_cached_scores(query, documents_tuple) if not hasattr(process_with_cache, cache): process_with_cache.cache {} process_with_cache.cache[cache_key] result return result7. 总结通过本教程我们为Qwen3-Reranker-0.6B语义重排序服务构建了一套完整的监控系统。让我们回顾一下关键收获7.1 核心成果健康检查端点让外部系统能够判断服务是否健康支持Kubernetes的存活性和就绪性探针Prometheus监控集成实时收集服务性能指标包括请求速率、延迟、资源使用等业务指标跟踪不仅监控系统状态还跟踪业务相关的指标如文档长度、分数分布等告警系统当服务出现异常时能够及时通知运维人员可视化仪表板通过Grafana直观展示服务运行状态7.2 实际价值这套监控系统为你带来了以下实际价值快速故障排查当服务出现问题时你可以立即查看相关指标快速定位问题原因容量规划通过监控资源使用趋势你可以合理规划服务器资源避免资源浪费或不足性能优化基于延迟和吞吐量数据你可以识别性能瓶颈并进行优化服务质量保证确保用户获得稳定、可靠的服务体验运维自动化减少人工检查的工作量让运维更加高效7.3 下一步建议如果你想要进一步深化监控能力可以考虑分布式追踪集成Jaeger或Zipkin跟踪请求在微服务间的流转日志聚合使用ELK栈Elasticsearch, Logstash, Kibana集中管理日志自定义指标根据你的业务需求添加更多有意义的业务指标自动化扩缩容基于监控指标自动调整服务实例数量A/B测试监控如果你有多个模型版本可以监控不同版本的表现监控不是一次性的工作而是一个持续的过程。随着业务的发展你需要不断调整和优化监控策略。但有了今天搭建的基础你已经迈出了重要的一步。记住好的监控系统就像汽车的仪表盘——它不能防止事故但能让你在问题发生前及时察觉在问题发生时快速响应。现在你的Qwen3-Reranker服务有了这样的仪表盘可以更加自信地服务用户了。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
