Qwen3-4B Instruct-2507实操手册：模型响应延迟监控+SLA告警配置

1. 引言：为什么需要监控大模型服务？

想象一下，你刚部署好一个基于Qwen3-4B的智能对话服务，用户开始涌入，一切看起来都很美好。但突然，有用户反馈：“你们的AI助手怎么变慢了？” 或者更糟，服务间歇性卡顿，你却毫不知情。等到大量用户投诉时，问题可能已经发酵了很久。

这就是为什么我们需要给AI服务装上“眼睛”和“耳朵”——监控系统。特别是对于Qwen3-4B Instruct-2507这样的纯文本大模型，虽然它已经通过移除视觉模块优化了速度，但在实际生产环境中，响应延迟依然会受到多种因素影响：

• 硬件负载波动：GPU显存占用、CPU使用率变化
• 请求并发压力：同时处理的对话数量
• 生成长度差异：用户请求的文本长度千差万别
• 网络与系统因素：容器调度、内存交换等

没有监控，你就等于在“盲飞”。本文将手把手带你为Qwen3-4B服务搭建一套完整的响应延迟监控与SLA（服务等级协议）告警系统。学完本文，你将能够：

• 实时掌握服务的健康状态和性能表现
• 提前预警潜在的性能瓶颈和故障风险
• 量化评估服务是否满足既定的SLA标准
• 数据驱动地进行容量规划和性能优化

2. 监控体系设计：我们要监控什么？

在开始敲代码之前，我们先要明确监控的目标。一个好的监控体系应该像汽车的仪表盘，能一目了然地告诉你当前的状态和潜在问题。

2.1 核心监控指标

对于Qwen3-4B这样的文本生成服务，我们需要重点关注以下几类指标：

响应延迟相关指标

• 请求处理时间：从收到用户请求到返回完整响应的总时间
• 首Token时间：从请求开始到第一个Token生成的时间（影响“流式输出”的初始感知）
• Token生成速率：每秒生成的Token数量（Tokens per second）

资源使用指标

• GPU显存使用率：模型推理的核心资源
• GPU利用率：显卡计算单元的活跃程度
• 系统内存使用率：防止因内存不足导致的交换
• 请求并发数：当前正在处理的请求数量

服务质量指标

• 请求成功率：成功处理的请求比例
• 错误类型分布：各种错误（超时、显存不足等）的统计
• SLA达标率：响应时间在约定阈值内的请求比例

2.2 SLA定义示例

SLA是服务提供方对用户的质量承诺。对于AI对话服务，常见的SLA指标包括：

# 示例SLA定义
sla_targets:
  p95_response_time:  # 95%的请求响应时间
    threshold: 3000ms  # 不超过3秒
    description: "95%的用户请求应在3秒内完成"
  
  p99_response_time:  # 99%的请求响应时间  
    threshold: 5000ms  # 不超过5秒
    description: "99%的用户请求应在5秒内完成"
  
  availability:  # 服务可用性
    threshold: 99.5%  # 每月可用时间不低于99.5%
    description: "服务月度可用性不低于99.5%"
  
  error_rate:  # 错误率
    threshold: 0.1%  # 错误请求比例不超过0.1%
    description: "请求错误率不高于0.1%"

3. 实战部署：为Qwen3-4B添加监控埋点

现在，让我们进入实战环节。我们将基于现有的Qwen3-4B Streamlit应用，添加监控功能。

3.1 环境准备与依赖安装

首先，确保你的环境已经安装了必要的监控库：

# 安装监控相关依赖
pip install prometheus-client==0.20.0
pip install psutil==5.9.8
pip install gpustat==1.1.0  # GPU监控
pip install pynvml==11.5.0  # NVIDIA管理库

3.2 创建监控模块

我们创建一个独立的监控模块 monitor.py：

# monitor.py - Qwen3-4B服务监控模块
import time
import psutil
import threading
from typing import Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime
import logging

try:
    import pynvml
    HAS_NVIDIA = True
except ImportError:
    HAS_NVIDIA = False

try:
    from prometheus_client import (
        Counter, Histogram, Gauge, 
        generate_latest, REGISTRY,
        start_http_server
    )
    HAS_PROMETHEUS = True
except ImportError:
    HAS_PROMETHEUS = False

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

@dataclass
class RequestMetrics:
    """单次请求的监控指标"""
    request_id: str
    start_time: float
    end_time: Optional[float] = None
    first_token_time: Optional[float] = None
    token_count: int = 0
    success: bool = True
    error_type: Optional[str] = None
    
    @property
    def total_duration(self) -> Optional[float]:
        """计算总处理时间（秒）"""
        if self.end_time:
            return self.end_time - self.start_time
        return None
    
    @property
    def time_to_first_token(self) -> Optional[float]:
        """计算首Token时间（秒）"""
        if self.first_token_time:
            return self.first_token_time - self.start_time
        return None
    
    @property
    def tokens_per_second(self) -> Optional[float]:
        """计算Token生成速率"""
        if self.end_time and self.first_token_time and self.token_count > 0:
            generation_time = self.end_time - self.first_token_time
            if generation_time > 0:
                return self.token_count / generation_time
        return None


class QwenMonitor:
    """Qwen3-4B服务监控器"""
    
    def __init__(self, prometheus_port: int = 8000):
        """
        初始化监控器
        
        Args:
            prometheus_port: Prometheus metrics暴露端口
        """
        self.metrics = {}
        self._init_prometheus_metrics()
        self._start_metrics_server(prometheus_port)
        self._start_system_monitor()
        
        logger.info(f"Qwen3-4B监控器初始化完成，Prometheus metrics端口: {prometheus_port}")
    
    def _init_prometheus_metrics(self):
        """初始化Prometheus指标"""
        if not HAS_PROMETHEUS:
            logger.warning("未安装prometheus-client，监控指标将不会暴露")
            return
        
        # 请求相关指标
        self.request_counter = Counter(
            'qwen_requests_total',
            '总请求数',
            ['status']  # 按状态（success, error）标签
        )
        
        self.request_duration = Histogram(
            'qwen_request_duration_seconds',
            '请求处理时间分布',
            buckets=(0.1, 0.5, 1.0, 2.0, 3.0, 5.0, 10.0, 30.0)
        )
        
        self.first_token_duration = Histogram(
            'qwen_first_token_duration_seconds',
            '首Token生成时间分布',
            buckets=(0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0)
        )
        
        self.tokens_per_second_gauge = Gauge(
            'qwen_tokens_per_second',
            'Token生成速率（Tokens/s）'
        )
        
        # 系统资源指标
        self.gpu_memory_gauge = Gauge(
            'qwen_gpu_memory_usage_percent',
            'GPU显存使用率（百分比）',
            ['gpu_id']
        )
        
        self.gpu_utilization_gauge = Gauge(
            'qwen_gpu_utilization_percent',
            'GPU利用率（百分比）',
            ['gpu_id']
        )
        
        self.system_memory_gauge = Gauge(
            'qwen_system_memory_usage_percent',
            '系统内存使用率（百分比）'
        )
        
        self.concurrent_requests_gauge = Gauge(
            'qwen_concurrent_requests',
            '当前并发请求数'
        )
    
    def _start_metrics_server(self, port: int):
        """启动Prometheus metrics服务器"""
        if HAS_PROMETHEUS:
            try:
                start_http_server(port)
                logger.info(f"Prometheus metrics服务器已启动，端口: {port}")
            except Exception as e:
                logger.error(f"启动Prometheus服务器失败: {e}")
    
    def _start_system_monitor(self):
        """启动系统资源监控线程"""
        if HAS_PROMETHEUS:
            monitor_thread = threading.Thread(
                target=self._system_monitor_loop,
                daemon=True
            )
            monitor_thread.start()
            logger.info("系统资源监控线程已启动")
    
    def _system_monitor_loop(self):
        """系统资源监控循环"""
        import time as time_module
        
        while True:
            try:
                # 监控系统内存
                memory = psutil.virtual_memory()
                self.system_memory_gauge.set(memory.percent)
                
                # 监控GPU资源
                if HAS_NVIDIA:
                    pynvml.nvmlInit()
                    device_count = pynvml.nvmlDeviceGetCount()
                    
                    for i in range(device_count):
                        handle = pynvml.nvmlDeviceGetHandleByIndex(i)
                        
                        # GPU显存使用率
                        mem_info = pynvml.nvmlDeviceGetMemoryInfo(handle)
                        mem_usage = (mem_info.used / mem_info.total) * 100
                        self.gpu_memory_gauge.labels(gpu_id=str(i)).set(mem_usage)
                        
                        # GPU利用率
                        util = pynvml.nvmlDeviceGetUtilizationRates(handle)
                        self.gpu_utilization_gauge.labels(gpu_id=str(i)).set(util.gpu)
                    
                    pynvml.nvmlShutdown()
                
            except Exception as e:
                logger.error(f"系统资源监控出错: {e}")
            
            # 每5秒采集一次
            time_module.sleep(5)
    
    def start_request(self, request_id: str) -> RequestMetrics:
        """
        开始记录一个请求
        
        Args:
            request_id: 请求唯一标识
            
        Returns:
            RequestMetrics对象
        """
        metrics = RequestMetrics(
            request_id=request_id,
            start_time=time.time()
        )
        self.metrics[request_id] = metrics
        
        # 更新并发请求数
        if HAS_PROMETHEUS:
            self.concurrent_requests_gauge.inc()
        
        logger.debug(f"开始监控请求: {request_id}")
        return metrics
    
    def record_first_token(self, request_id: str):
        """记录首Token生成时间"""
        if request_id in self.metrics:
            self.metrics[request_id].first_token_time = time.time()
            
            if HAS_PROMETHEUS:
                duration = self.metrics[request_id].time_to_first_token
                if duration:
                    self.first_token_duration.observe(duration)
    
    def complete_request(
        self, 
        request_id: str, 
        token_count: int = 0,
        success: bool = True,
        error_type: Optional[str] = None
    ):
        """
        完成请求记录
        
        Args:
            request_id: 请求唯一标识
            token_count: 生成的Token数量
            success: 是否成功
            error_type: 错误类型（如果失败）
        """
        if request_id not in self.metrics:
            logger.warning(f"未找到请求记录: {request_id}")
            return
        
        metrics = self.metrics[request_id]
        metrics.end_time = time.time()
        metrics.token_count = token_count
        metrics.success = success
        metrics.error_type = error_type
        
        # 更新Prometheus指标
        if HAS_PROMETHEUS:
            # 请求计数
            status = "success" if success else "error"
            self.request_counter.labels(status=status).inc()
            
            # 请求耗时
            duration = metrics.total_duration
            if duration:
                self.request_duration.observe(duration)
            
            # Token生成速率
            tps = metrics.tokens_per_second
            if tps:
                self.tokens_per_second_gauge.set(tps)
            
            # 减少并发请求数
            self.concurrent_requests_gauge.dec()
        
        # 记录日志
        log_level = logging.INFO if success else logging.ERROR
        logger.log(
            log_level,
            f"请求完成: {request_id}, "
            f"耗时: {duration:.2f}s, "
            f"Token数: {token_count}, "
            f"速率: {tps:.1f} tokens/s" if tps else ""
        )
        
        # 清理过期的记录（保留最近1000条）
        if len(self.metrics) > 1000:
            # 删除最旧的记录
            oldest_id = min(self.metrics.keys(), 
                          key=lambda k: self.metrics[k].start_time)
            del self.metrics[oldest_id]
    
    def get_metrics_summary(self) -> Dict[str, Any]:
        """获取监控指标摘要"""
        if not self.metrics:
            return {"message": "暂无监控数据"}
        
        completed = [m for m in self.metrics.values() 
                    if m.end_time is not None]
        
        if not completed:
            return {"message": "暂无完成的请求"}
        
        durations = [m.total_duration for m in completed if m.total_duration]
        first_token_times = [m.time_to_first_token for m in completed 
                           if m.time_to_first_token]
        
        summary = {
            "total_requests": len(completed),
            "success_rate": sum(1 for m in completed if m.success) / len(completed),
            "avg_duration": sum(durations) / len(durations) if durations else 0,
            "p95_duration": sorted(durations)[int(len(durations) * 0.95)] if durations else 0,
            "p99_duration": sorted(durations)[int(len(durations) * 0.99)] if durations else 0,
            "avg_first_token_time": sum(first_token_times) / len(first_token_times) if first_token_times else 0,
            "avg_tokens_per_second": sum(m.tokens_per_second for m in completed 
                                        if m.tokens_per_second) / len(completed),
            "current_concurrent": len([m for m in self.metrics.values() 
                                     if m.end_time is None])
        }
        
        return summary

3.3 集成监控到Streamlit应用

接下来，我们将监控模块集成到现有的Qwen3-4B Streamlit应用中：

# app_with_monitor.py - 集成监控的Streamlit应用
import streamlit as st
import uuid
import time
from datetime import datetime
from monitor import QwenMonitor, RequestMetrics

# 初始化监控器（单例模式）
@st.cache_resource
def get_monitor():
    """获取监控器实例"""
    return QwenMonitor(prometheus_port=8000)

# 在应用初始化时创建监控器
monitor = get_monitor()

# 修改原有的聊天处理函数，添加监控埋点
def process_message_with_monitoring(user_input, max_length, temperature):
    """
    处理用户消息（带监控版本）
    
    Args:
        user_input: 用户输入
        max_length: 最大生成长度
        temperature: 温度参数
        
    Returns:
        模型生成的回复
    """
    # 生成请求ID
    request_id = str(uuid.uuid4())[:8]
    
    # 开始监控
    metrics = monitor.start_request(request_id)
    
    try:
        # 记录请求开始时间
        st.session_state.metrics = metrics
        
        # 调用原有的模型生成函数
        # 这里假设原有的生成函数是 generate_response
        response_generator = generate_response(
            user_input, 
            max_length=max_length,
            temperature=temperature
        )
        
        # 流式输出处理
        full_response = ""
        first_token_received = False
        
        for chunk in response_generator:
            if not first_token_received:
                # 记录首Token时间
                monitor.record_first_token(request_id)
                first_token_received = True
            
            full_response += chunk
            yield chunk
        
        # 计算Token数量（简化示例，实际应从tokenizer获取）
        token_count = len(full_response.split()) * 1.3  # 近似估算
        
        # 请求成功完成
        monitor.complete_request(
            request_id=request_id,
            token_count=int(token_count),
            success=True
        )
        
    except Exception as e:
        # 请求失败
        error_type = type(e).__name__
        logger.error(f"请求 {request_id} 失败: {error_type} - {str(e)}")
        
        monitor.complete_request(
            request_id=request_id,
            token_count=0,
            success=False,
            error_type=error_type
        )
        
        # 重新抛出异常
        raise e

# 在Streamlit侧边栏添加监控面板
def add_monitoring_sidebar():
    """在侧边栏添加监控信息"""
    with st.sidebar:
        st.markdown("---")
        st.subheader(" 服务监控")
        
        # 显示实时指标
        if st.button("刷新监控数据", key="refresh_metrics"):
            st.rerun()
        
        # 获取监控摘要
        summary = monitor.get_metrics_summary()
        
        if "message" in summary:
            st.info(summary["message"])
        else:
            # 显示关键指标
            col1, col2 = st.columns(2)
            
            with col1:
                st.metric(
                    label="总请求数",
                    value=summary["total_requests"]
                )
                st.metric(
                    label="成功率",
                    value=f"{summary['success_rate']:.1%}"
                )
            
            with col2:
                st.metric(
                    label="平均响应时间",
                    value=f"{summary['avg_duration']:.2f}s"
                )
                st.metric(
                    label="当前并发",
                    value=summary["current_concurrent"]
                )
            
            # 显示详细指标
            with st.expander("查看详细指标"):
                st.write(f"**P95响应时间**: {summary['p95_duration']:.2f}s")
                st.write(f"**P99响应时间**: {summary['p99_duration']:.2f}s")
                st.write(f"**平均首Token时间**: {summary['avg_first_token_time']:.3f}s")
                st.write(f"**平均生成速率**: {summary['avg_tokens_per_second']:.1f} tokens/s")
                
                # Prometheus metrics链接
                st.markdown("---")
                st.caption(" 高级监控")
                st.write("访问 [http://localhost:8000](http://localhost:8000) 查看完整的Prometheus指标")
                st.code("curl http://localhost:8000/metrics", language="bash")

# 在主应用中调用
def main():
    """主应用函数"""
    st.title("⚡ Qwen3-4B Instruct-2507 - 带监控的对话服务")
    
    # 添加监控侧边栏
    add_monitoring_sidebar()
    
    # 原有的聊天界面代码...
    # ... [保留原有的聊天界面代码]
    
    # 修改消息处理部分，使用带监控的函数
    if user_input:
        # 使用带监控的消息处理
        response_generator = process_message_with_monitoring(
            user_input, 
            max_length, 
            temperature
        )
        
        # 流式显示回复
        with st.chat_message("assistant"):
            message_placeholder = st.empty()
            full_response = ""
            
            for chunk in response_generator:
                full_response += chunk
                message_placeholder.markdown(full_response + "▌")
            
            message_placeholder.markdown(full_response)
        
        # 保存到聊天历史
        st.session_state.messages.append({
            "role": "assistant", 
            "content": full_response
        })

if __name__ == "__main__":
    main()

4. SLA告警配置：当服务出现问题时自动通知

监控数据收集好了，接下来我们需要设置告警，当服务不符合SLA时及时通知我们。

4.1 使用Prometheus + Alertmanager配置告警

首先，我们需要配置Prometheus的告警规则。创建一个 alerts.yml 文件：

# alerts.yml - Prometheus告警规则
groups:
  - name: qwen_sla_alerts
    rules:
      # 响应时间告警
      - alert: HighResponseTime
        expr: histogram_quantile(0.95, rate(qwen_request_duration_seconds_bucket[5m])) > 3
        for: 2m
        labels:
          severity: warning
          service: qwen3-4b
        annotations:
          summary: "Qwen服务响应时间过高"
          description: |
            Qwen3-4B服务的P95响应时间超过3秒已达2分钟。
            当前值: {{ $value }}秒
            实例: {{ $labels.instance }}
      
      - alert: CriticalResponseTime
        expr: histogram_quantile(0.95, rate(qwen_request_duration_seconds_bucket[5m])) > 5
        for: 1m
        labels:
          severity: critical
          service: qwen3-4b
        annotations:
          summary: "Qwen服务响应时间严重超标"
          description: |
            Qwen3-4B服务的P95响应时间超过5秒已达1分钟。
            当前值: {{ $value }}秒
            实例: {{ $labels.instance }}
      
      # 错误率告警
      - alert: HighErrorRate
        expr: rate(qwen_requests_total{status="error"}[5m]) / rate(qwen_requests_total[5m]) > 0.01
        for: 3m
        labels:
          severity: warning
          service: qwen3-4b
        annotations:
          summary: "Qwen服务错误率过高"
          description: |
            Qwen3-4B服务的错误率超过1%已达3分钟。
            当前错误率: {{ $value | humanizePercentage }}
            实例: {{ $labels.instance }}
      
      # GPU资源告警
      - alert: HighGPUMemoryUsage
        expr: qwen_gpu_memory_usage_percent > 90
        for: 5m
        labels:
          severity: warning
          service: qwen3-4b
        annotations:
          summary: "GPU显存使用率过高"
          description: |
            GPU显存使用率超过90%已达5分钟。
            当前使用率: {{ $value }}%
            GPU: {{ $labels.gpu_id }}
      
      - alert: HighGPUUtilization
        expr: qwen_gpu_utilization_percent > 95
        for: 5m
        labels:
          severity: warning
          service: qwen3-4b
        annotations:
          summary: "GPU利用率过高"
          description: |
            GPU利用率超过95%已达5分钟。
            当前利用率: {{ $value }}%
            GPU: {{ $labels.gpu_id }}
      
      # 系统内存告警
      - alert: HighSystemMemoryUsage
        expr: qwen_system_memory_usage_percent > 85
        for: 5m
        labels:
          severity: warning
          service: qwen3-4b
        annotations:
          summary: "系统内存使用率过高"
          description: |
            系统内存使用率超过85%已达5分钟。
            当前使用率: {{ $value }}%
      
      # 服务可用性告警（基于请求成功率）
      - alert: LowRequestSuccessRate
        expr: rate(qwen_requests_total{status="success"}[10m]) / rate(qwen_requests_total[10m]) < 0.995
        for: 5m
        labels:
          severity: critical
          service: qwen3-4b
        annotations:
          summary: "Qwen服务成功率过低"
          description: |
            Qwen3-4B服务成功率低于99.5%已达5分钟。
            当前成功率: {{ $value | humanizePercentage }}

4.2 配置Alertmanager发送通知

接下来，配置Alertmanager来发送告警通知。创建 alertmanager.yml：

# alertmanager.yml - Alertmanager配置
global:
  smtp_smarthost: 'smtp.gmail.com:587'  # 邮件服务器
  smtp_from: 'alerts@yourcompany.com'    # 发件人
  smtp_auth_username: 'your-email@gmail.com'
  smtp_auth_password: 'your-app-password'

route:
  group_by: ['alertname', 'service']
  group_wait: 10s
  group_interval: 10s
  repeat_interval: 1h
  receiver: 'default-receiver'
  
  routes:
    - match:
        severity: critical
      receiver: 'critical-alerts'
      group_wait: 5s
      repeat_interval: 5m
    
    - match:
        severity: warning
      receiver: 'warning-alerts'

receivers:
  - name: 'default-receiver'
    email_configs:
      - to: 'team@yourcompany.com'
        send_resolved: true
  
  - name: 'critical-alerts'
    email_configs:
      - to: 'oncall-engineer@yourcompany.com'
        send_resolved: true
    webhook_configs:
      - url: 'https://hooks.slack.com/services/your/slack/webhook'
        send_resolved: true
  
  - name: 'warning-alerts'
    email_configs:
      - to: 'team@yourcompany.com'
        send_resolved: true

4.3 使用Docker Compose一键部署监控栈

为了方便部署，我们可以使用Docker Compose来一键启动完整的监控栈：

# docker-compose-monitor.yml
version: '3.8'

services:
  # Qwen3-4B应用（带监控）
  qwen-app:
    build: .
    ports:
      - "8501:8501"  # Streamlit端口
      - "8000:8000"  # Prometheus metrics端口
    environment:
      - MODEL_NAME=Qwen/Qwen3-4B-Instruct-2507
      - DEVICE=cuda
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    volumes:
      - ./models:/app/models
    restart: unless-stopped
    networks:
      - monitoring

  # Prometheus（指标收集）
  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - ./alerts.yml:/etc/prometheus/alerts.yml
      - prometheus_data:/prometheus
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.path=/prometheus'
      - '--web.console.libraries=/etc/prometheus/console_libraries'
      - '--web.console.templates=/etc/prometheus/consoles'
      - '--storage.tsdb.retention.time=200h'
      - '--web.enable-lifecycle'
    restart: unless-stopped
    networks:
      - monitoring

  # Alertmanager（告警管理）
  alertmanager:
    image: prom/alertmanager:latest
    ports:
      - "9093:9093"
    volumes:
      - ./alertmanager.yml:/etc/alertmanager/alertmanager.yml
      - alertmanager_data:/alertmanager
    command:
      - '--config.file=/etc/alertmanager/alertmanager.yml'
      - '--storage.path=/alertmanager'
    restart: unless-stopped
    networks:
      - monitoring

  # Grafana（数据可视化）
  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin123
    volumes:
      - grafana_data:/var/lib/grafana
      - ./grafana/provisioning:/etc/grafana/provisioning
    restart: unless-stopped
    networks:
      - monitoring

networks:
  monitoring:
    driver: bridge

volumes:
  prometheus_data:
  alertmanager_data:
  grafana_data:

对应的Prometheus配置文件 prometheus.yml：

# prometheus.yml
global:
  scrape_interval: 15s
  evaluation_interval: 15s

rule_files:
  - "alerts.yml"

scrape_configs:
  - job_name: 'qwen3-4b'
    static_configs:
      - targets: ['qwen-app:8000']
        labels:
          service: 'qwen3-4b'
          instance: 'qwen-instance-1'

  - job_name: 'prometheus'
    static_configs:
      - targets: ['localhost:9090']

alerting:
  alertmanagers:
    - static_configs:
        - targets:
            - 'alertmanager:9093'

4.4 启动监控栈

使用以下命令启动完整的监控系统：

# 启动所有服务
docker-compose -f docker-compose-monitor.yml up -d

# 查看服务状态
docker-compose -f docker-compose-monitor.yml ps

# 查看日志
docker-compose -f docker-compose-monitor.yml logs -f qwen-app

服务启动后，你可以访问：

• Qwen应用: http://localhost:8501
• Prometheus: http://localhost:9090
• Alertmanager: http://localhost:9093
• Grafana: http://localhost:3000 (用户名: admin, 密码: admin123)

5. 监控看板与可视化

有了数据，我们还需要一个直观的看板来展示监控指标。Grafana是一个强大的可视化工具，我们可以用它创建Qwen3-4B服务的监控看板。

5.1 创建Grafana数据源

首先，在Grafana中添加Prometheus作为数据源：

1. 访问 http://localhost:3000
2. 使用 admin/admin123 登录
3. 进入 Configuration → Data Sources → Add data source
4. 选择 Prometheus
5. URL填写: http://prometheus:9090
6. 点击 Save & Test

5.2 导入预制的监控看板

你可以导入一个预制的Qwen3-4B监控看板。创建一个JSON文件 qwen-dashboard.json：

{
  "dashboard": {
    "title": "Qwen3-4B服务监控",
    "panels": [
      {
        "title": "请求响应时间（P95）",
        "type": "graph",
        "targets": [{
          "expr": "histogram_quantile(0.95, rate(qwen_request_duration_seconds_bucket[5m]))",
          "legendFormat": "P95响应时间"
        }],
        "yaxes": [{"format": "s"}]
      },
      {
        "title": "请求成功率",
        "type": "stat",
        "targets": [{
          "expr": "rate(qwen_requests_total{status=\"success\"}[5m]) / rate(qwen_requests_total[5m]) * 100",
          "legendFormat": "成功率"
        }],
        "fieldConfig": {
          "defaults": {
            "unit": "percent"
          }
        }
      },
      {
        "title": "Token生成速率",
        "type": "graph",
        "targets": [{
          "expr": "qwen_tokens_per_second",
          "legendFormat": "Tokens/s"
        }]
      },
      {
        "title": "GPU显存使用率",
        "type": "gauge",
        "targets": [{
          "expr": "qwen_gpu_memory_usage_percent",
          "legendFormat": "GPU {{gpu_id}}"
        }],
        "fieldConfig": {
          "defaults": {
            "unit": "percent",
            "thresholds": {
              "steps": [
                {"color": "green", "value": null},
                {"color": "yellow", "value": 70},
                {"color": "red", "value": 90}
              ]
            }
          }
        }
      },
      {
        "title": "当前并发请求数",
        "type": "stat",
        "targets": [{
          "expr": "qwen_concurrent_requests",
          "legendFormat": "并发数"
        }]
      },
      {
        "title": "错误请求分布",
        "type": "piechart",
        "targets": [{
          "expr": "sum by (error_type) (rate(qwen_requests_total{status=\"error\"}[5m]))",
          "legendFormat": "{{error_type}}"
        }]
      }
    ]
  }
}

然后在Grafana中导入这个看板：

1. 进入 Dashboards → Import
2. 上传JSON文件或粘贴JSON内容
3. 选择Prometheus数据源
4. 点击 Import

5.3 创建自定义告警面板

除了系统告警，你还可以在Grafana中创建自定义的告警面板：

# 创建一个简单的告警检查脚本
import requests
import json
from datetime import datetime

class SLAAlertChecker:
    """SLA告警检查器"""
    
    def __init__(self, prometheus_url="http://localhost:9090"):
        self.prometheus_url = prometheus_url
    
    def check_sla_violations(self):
        """检查SLA违规情况"""
        violations = []
        
        # 检查P95响应时间是否超过3秒
        p95_response = self.query_prometheus(
            'histogram_quantile(0.95, rate(qwen_request_duration_seconds_bucket[5m]))'
        )
        
        if p95_response and p95_response[0]['value'][1] > '3':
            violations.append({
                'metric': 'P95响应时间',
                'current': float(p95_response[0]['value'][1]),
                'threshold': 3.0,
                'severity': 'warning'
            })
        
        # 检查错误率是否超过0.1%
        error_rate = self.query_prometheus(
            'rate(qwen_requests_total{status="error"}[5m]) / rate(qwen_requests_total[5m])'
        )
        
        if error_rate and float(error_rate[0]['value'][1]) > 0.001:
            violations.append({
                'metric': '错误率',
                'current': float(error_rate[0]['value'][1]) * 100,
                'threshold': 0.1,
                'severity': 'critical'
            })
        
        # 检查GPU显存使用率是否超过90%
        gpu_memory = self.query_prometheus(
            'qwen_gpu_memory_usage_percent'
        )
        
        for gpu in gpu_memory or []:
            if float(gpu['value'][1]) > 90:
                violations.append({
                    'metric': f"GPU{gpu['metric'].get('gpu_id', '0')}显存使用率",
                    'current': float(gpu['value'][1]),
                    'threshold': 90.0,
                    'severity': 'warning'
                })
        
        return violations
    
    def query_prometheus(self, query):
        """查询Prometheus"""
        try:
            response = requests.get(
                f"{self.prometheus_url}/api/v1/query",
                params={'query': query}
            )
            response.raise_for_status()
            return response.json()['data']['result']
        except Exception as e:
            print(f"查询Prometheus失败: {e}")
            return None
    
    def generate_alert_report(self):
        """生成告警报告"""
        violations = self.check_sla_violations()
        
        if not violations:
            return {
                'status': 'healthy',
                'timestamp': datetime.now().isoformat(),
                'message': '所有SLA指标正常'
            }
        
        report = {
            'status': 'violation',
            'timestamp': datetime.now().isoformat(),
            'violations': violations,
            'summary': f"发现 {len(violations)} 个SLA违规"
        }
        
        # 发送告警通知
        self.send_alerts(violations)
        
        return report
    
    def send_alerts(self, violations):
        """发送告警通知"""
        # 这里可以集成邮件、Slack、钉钉等通知方式
        for violation in violations:
            print(f"[{violation['severity'].upper()}] {violation['metric']} 超标: "
                  f"{violation['current']:.2f} > {violation['threshold']}")

# 使用示例
if __name__ == "__main__":
    checker = SLAAlertChecker()
    
    # 定时检查（例如每分钟一次）
    import time
    while True:
        report = checker.generate_alert_report()
        print(json.dumps(report, indent=2, ensure_ascii=False))
        time.sleep(60)  # 每分钟检查一次

6. 总结：构建完整的AI服务监控体系

通过本文的实践，我们为Qwen3-4B Instruct-2507服务构建了一个完整的监控告警体系。让我们回顾一下关键要点：

6.1 监控体系的核心价值

1. 实时可见性：通过Prometheus + Grafana，我们能够实时查看服务的各项关键指标，从响应时间到资源使用率，一目了然。

2. 主动告警：当服务出现异常或即将违反SLA时，Alertmanager会通过邮件、Slack等方式及时通知我们，让我们能够在用户发现问题之前就采取行动。

3. 数据驱动优化：监控数据不仅用于告警，更是性能优化的依据。通过分析历史数据，我们可以：

   • 识别性能瓶颈（是GPU不足还是代码问题？）
   • 预测容量需求（什么时候需要扩容？）
   • 验证优化效果（修改配置后性能提升多少？）

4. SLA合规证明：详细的监控数据可以作为服务质量的客观证明，无论是内部汇报还是对客户承诺，都有数据支撑。

6.2 实际部署建议

在实际生产环境中部署时，建议考虑以下几点：

监控粒度调整

• 根据业务需求调整监控指标的采集频率
• 对于关键业务指标，可以设置更频繁的采集（如每5秒）
• 对于资源监控，可以适当降低频率（如每30秒）

告警策略优化

• 避免告警疲劳：合理设置告警阈值和静默期
• 分级告警：区分警告（warning）和严重（critical）级别
• 告警聚合：相似告警合并发送，减少干扰

数据保留策略

• 原始数据保留7-30天，用于问题排查
• 聚合数据（如小时/天级统计）长期保留，用于趋势分析
• 定期清理过期数据，控制存储成本

扩展性考虑

• 随着服务规模扩大，考虑使用集群化的Prometheus（如Thanos或Cortex）
• 对于多实例部署，使用服务发现自动添加监控目标
• 考虑使用云原生的监控方案（如AWS CloudWatch、Azure Monitor等）

6.3 下一步行动建议

如果你已经按照本文完成了基础监控部署，接下来可以考虑：

1. 性能基准测试：在不同硬件配置下运行基准测试，建立性能基线
2. 压力测试监控：在压力测试时观察监控指标，确定服务的极限容量
3. 成本监控：结合GPU使用率和响应时间，优化资源利用率
4. 用户体验监控：从前端角度监控页面加载时间、交互响应等
5. 业务指标监控：监控用户活跃度、对话质量等业务相关指标

记住，监控不是一次性的工作，而是一个持续的过程。随着业务发展和用户增长，你需要不断调整监控策略，确保它始终能够反映服务的真实状态。

---

获取更多AI镜像

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