AIGlasses_for_navigation自动化运维：利用Python脚本进行模型服务监控与告警

AIGlasses_for_navigation自动化运维利用Python脚本进行模型服务监控与告警最近在负责一个基于AIGlasses_for_navigation模型的在线服务部署上线后最让人头疼的就是服务稳定性问题。半夜三更被电话叫醒说服务挂了或者响应慢如蜗牛这种经历相信不少搞AI应用落地的朋友都深有体会。模型服务不像传统的Web服务它背后依赖GPU资源推理过程也充满不确定性单纯靠人肉盯着监控面板效率低还容易遗漏。为了解决这个问题我花了一些时间用Python写了一套轻量级的监控告警脚本。这套东西不复杂但特别实用能自动检查服务的“心跳”、GPU的“血压”、推理的“速度”一旦发现异常立马通过邮件或者钉钉“喊人”。今天就把这套方案的思路和代码分享出来如果你也在为AI模型服务的运维发愁希望能给你一些直接的参考。1. 为什么需要专门的模型服务监控你可能觉得用现有的APM应用性能监控工具不就行了刚开始我也这么想但实际用下来发现通用监控工具对AI模型服务的“把脉”不够准。模型服务有几个独特的地方第一它严重依赖GPU显存泄露、算力饱和是常事第二推理延迟波动大受输入数据影响显著第三服务进程可能没挂但模型加载失败导致返回错误。这些细枝末节的问题通用监控往往覆盖不到。我们需要的监控得能直接“听懂”模型服务的“语言”。比如不仅要看端口通不通还要看模型能不能正常推理不仅要看服务器负载还要看显存用了多少、CUDA内核有没有异常。这套Python脚本就是干这个的它像是一个贴在模型服务身上的“听诊器”专门收集那些关键的、模型特有的健康指标。2. 监控脚本的核心设计思路整个脚本的设计围绕四个核心监控指标展开我称之为模型服务的“生命体征”。服务进程状态这是最基本的。脚本会检查托管模型的服务进程比如用FastAPI、Flask或者Triton Inference Server启动的是否还在运行。光检查进程存在还不够我们还需要确认其HTTP/GRPC服务端口是否可访问。GPU资源占用模型推理的“主战场”。我们会监控GPU的显存使用率、利用率以及温度。显存使用率缓慢增长可能预示内存泄露利用率持续100%可能意味着请求队列堵塞温度过高则会影响硬件寿命和推理稳定性。推理性能指标这是用户体验的直接体现。主要监控两个值平均推理延迟和尾部延迟如P99。延迟突然飙升往往意味着输入数据异常或模型遇到了“难缠”的case。同时我们也会统计请求成功率过滤掉因模型内部错误导致的失败请求。业务与系统资源除了模型本身还要关注其运行环境。比如服务器的CPU使用率、内存占用、磁盘IO以及网络带宽。这些资源瓶颈同样会间接拖累模型服务。当这些指标超出我们设定的阈值时脚本就需要触发告警。告警策略上我采用了“多级预警”机制。比如GPU温度达到85度发提醒到90度则发严重告警延迟偶尔超标只记录日志连续5次超标则立即告警这样可以有效减少告警噪音。3. 动手编写Python监控脚本理论说完了我们直接看代码。整个脚本主要分为监控器、告警器和调度器三部分。3.1 监控指标收集器首先我们需要收集数据。这里会用到几个有用的库psutil用于获取系统信息pynvml用于读取NVIDIA GPU状态requests用于探测服务端点。import psutil import requests import time import json from pynvml import * def check_service_health(service_url): 检查模型服务健康状态 :param service_url: 模型服务的健康检查端点或推理端点 :return: (is_healthy, latency, status_code) try: start_time time.time() # 假设服务有一个 /health 端点或者直接用轻量级推理端点 resp requests.get(f{service_url}/health, timeout5) latency (time.time() - start_time) * 1000 # 转换为毫秒 is_healthy resp.status_code 200 return is_healthy, latency, resp.status_code except requests.exceptions.RequestException as e: return False, 0, str(e) def get_gpu_metrics(): 获取GPU监控指标 :return: 列表包含每个GPU的字典信息 nvmlInit() device_count nvmlDeviceGetCount() gpu_metrics [] for i in range(device_count): handle nvmlDeviceGetHandleByIndex(i) util nvmlDeviceGetUtilizationRates(handle) memory_info nvmlDeviceGetMemoryInfo(handle) temp nvmlDeviceGetTemperature(handle, NVML_TEMPERATURE_GPU) gpu_info { gpu_id: i, gpu_utilization: util.gpu, memory_utilization: (memory_info.used / memory_info.total) * 100, memory_used_mb: memory_info.used / 1024**2, memory_total_mb: memory_info.total / 1024**2, temperature: temp } gpu_metrics.append(gpu_info) nvmlShutdown() return gpu_metrics def get_system_metrics(): 获取系统资源指标 :return: 系统指标字典 cpu_percent psutil.cpu_percent(interval1) memory psutil.virtual_memory() disk psutil.disk_usage(/) return { cpu_percent: cpu_percent, memory_percent: memory.percent, memory_used_gb: memory.used / 1024**3, disk_percent: disk.percent, disk_free_gb: disk.free / 1024**3 }3.2 告警触发器收集到数据后需要判断是否异常并决定是否告警。这里我实现了邮件和钉钉两种告警方式。import smtplib from email.mime.text import MIMEText from email.header import Header class AlertManager: def __init__(self, config): self.mail_config config.get(mail, {}) self.dingtalk_config config.get(dingtalk, {}) self.alert_history {} # 用于告警降噪记录上次告警时间 def send_mail_alert(self, subject, content, levelWARNING): 发送邮件告警 :param level: 告警级别 INFO/WARNING/ERROR if not self.mail_config.get(enabled, False): return False try: mail_host self.mail_config[host] mail_user self.mail_config[user] mail_pass self.mail_config[pass] sender self.mail_config[sender] receivers self.mail_config[receivers].split(,) message MIMEText(content, plain, utf-8) message[From] Header(fAI模型监控平台 {sender}, utf-8) message[To] Header(,.join(receivers), utf-8) message[Subject] Header(f[{level}] {subject}, utf-8) smtp_obj smtplib.SMTP_SSL(mail_host, 465) smtp_obj.login(mail_user, mail_pass) smtp_obj.sendmail(sender, receivers, message.as_string()) smtp_obj.quit() print(f邮件告警已发送: {subject}) return True except Exception as e: print(f发送邮件告警失败: {e}) return False def send_dingtalk_alert(self, subject, content, levelWARNING): 发送钉钉群机器人告警 if not self.dingtalk_config.get(enabled, False): return False webhook_url self.dingtalk_config[webhook] secret self.dingtalk_config.get(secret, ) # 加签密钥 import hashlib import hmac import base64 import urllib.parse from datetime import datetime timestamp str(round(time.time() * 1000)) sign_string f{timestamp}\n{secret} hmac_code hmac.new(secret.encode(utf-8), sign_string.encode(utf-8), digestmodhashlib.sha256).digest() sign urllib.parse.quote_plus(base64.b64encode(hmac_code)) webhook f{webhook_url}timestamp{timestamp}sign{sign} ding_msg { msgtype: markdown, markdown: { title: f{level}告警, text: f### {subject}\n\n**告警级别**: {level}\n\n**告警内容**: {content}\n\n**告警时间**: {datetime.now().strftime(%Y-%m-%d %H:%M:%S)} } } try: resp requests.post(webhook, jsonding_msg, timeout5) if resp.status_code 200: print(f钉钉告警已发送: {subject}) return True else: print(f钉钉告警发送失败: {resp.status_code}) return False except Exception as e: print(f发送钉钉告警异常: {e}) return False def check_and_alert(self, metric_name, current_value, threshold, alert_window300): 检查指标并触发告警带降噪窗口 :param alert_window: 相同告警的最小间隔时间秒防止告警风暴 is_alert current_value threshold now time.time() if is_alert: alert_key f{metric_name}_{threshold} last_alert self.alert_history.get(alert_key, 0) if now - last_alert alert_window: # 触发告警 subject f模型服务监控告警 - {metric_name} content f指标 [{metric_name}] 当前值: {current_value:.2f}, 超过阈值: {threshold}\n时间: {time.strftime(%Y-%m-%d %H:%M:%S)} # 根据严重程度选择告警方式 if current_value threshold * 1.5: # 严重超标 self.send_dingtalk_alert(subject, content, ERROR) self.send_mail_alert(subject, content, ERROR) else: self.send_dingtalk_alert(subject, content, WARNING) self.alert_history[alert_key] now return True return False3.3 主调度与配置最后我们把所有组件组装起来形成一个可以定时运行的完整脚本。import yaml import schedule from datetime import datetime class ModelServiceMonitor: def __init__(self, config_pathmonitor_config.yaml): with open(config_path, r) as f: self.config yaml.safe_load(f) self.service_url self.config[service][url] self.alert_manager AlertManager(self.config[alert]) self.metrics_history [] # 用于记录历史数据分析趋势 # 监控阈值配置 self.thresholds { service_latency_ms: self.config[thresholds].get(service_latency_ms, 1000), gpu_memory_percent: self.config[thresholds].get(gpu_memory_percent, 90), gpu_temperature: self.config[thresholds].get(gpu_temperature, 85), cpu_percent: self.config[thresholds].get(cpu_percent, 80), memory_percent: self.config[thresholds].get(memory_percent, 85) } def run_one_cycle(self): 执行一次完整的监控检查 print(f\n[{datetime.now().strftime(%Y-%m-%d %H:%M:%S)}] 开始监控检查...) # 1. 检查服务健康 is_healthy, latency, status check_service_health(self.service_url) if not is_healthy: self.alert_manager.send_dingtalk_alert( 服务不可用告警, f服务 {self.service_url} 健康检查失败。状态: {status}, ERROR ) else: # 检查延迟是否超标 self.alert_manager.check_and_alert(service_latency_ms, latency, self.thresholds[service_latency_ms]) # 2. 检查GPU指标 gpu_metrics get_gpu_metrics() for gpu in gpu_metrics: self.alert_manager.check_and_alert( fgpu_{gpu[gpu_id]}_memory_percent, gpu[memory_utilization], self.thresholds[gpu_memory_percent] ) self.alert_manager.check_and_alert( fgpu_{gpu[gpu_id]}_temperature, gpu[temperature], self.thresholds[gpu_temperature] ) # 3. 检查系统指标 sys_metrics get_system_metrics() self.alert_manager.check_and_alert(cpu_percent, sys_metrics[cpu_percent], self.thresholds[cpu_percent]) self.alert_manager.check_and_alert(memory_percent, sys_metrics[memory_percent], self.thresholds[memory_percent]) # 4. 记录本次监控结果可用于后续分析 record { timestamp: datetime.now().isoformat(), service_healthy: is_healthy, service_latency_ms: latency, gpu_metrics: gpu_metrics, system_metrics: sys_metrics } self.metrics_history.append(record) # 保持历史记录不超过1000条 if len(self.metrics_history) 1000: self.metrics_history self.metrics_history[-1000:] print(f监控检查完成。服务状态: {健康 if is_healthy else 异常}, 延迟: {latency:.2f}ms) return record def start(self, interval_minutes5): 启动定时监控 print(f启动模型服务监控每 {interval_minutes} 分钟检查一次) schedule.every(interval_minutes).minutes.do(self.run_one_cycle) # 立即运行一次 self.run_one_cycle() while True: schedule.run_pending() time.sleep(1) if __name__ __main__: monitor ModelServiceMonitor(config.yaml) monitor.start(interval_minutes5)对应的配置文件config.yaml可以这样写service: url: http://localhost:8000 # 你的模型服务地址 alert: mail: enabled: true host: smtp.example.com user: your_emailexample.com pass: your_password sender: monitorexample.com receivers: ops_teamexample.com,dev_teamexample.com dingtalk: enabled: true webhook: https://oapi.dingtalk.com/robot/send?access_tokenyour_token secret: your_secret # 钉钉机器人加签密钥 thresholds: service_latency_ms: 1000 # 服务延迟阈值毫秒 gpu_memory_percent: 90 # GPU显存使用率阈值 gpu_temperature: 85 # GPU温度阈值 cpu_percent: 80 # CPU使用率阈值 memory_percent: 85 # 内存使用率阈值4. 部署与实践建议脚本写好了怎么把它用起来呢这里分享几个实践中的小经验。部署方式建议在模型服务所在的服务器上用systemd或者supervisor把监控脚本作为一个后台服务来运行。这样即使服务器重启监控也能自动恢复。记得给脚本设置合适的日志轮转避免日志文件撑满磁盘。监控频率不是越频繁越好。对于GPU温度和显存每分钟检查一次是合理的。但对于服务延迟和成功率可以放宽到每5分钟一次避免监控本身对服务造成压力。关键业务时段可以调高频率夜间可以调低。阈值设置阈值不是一成不变的。最好能先让监控脚本在正常业务负载下跑几天收集一个基准线。然后根据基准线的平均值和峰值来设置合理的告警阈值。比如平时延迟是200ms峰值可能到500ms那么阈值可以设在800ms-1000ms给系统留出一定的缓冲空间。告警升级重要的线上服务建议设置多级告警。比如第一次异常发钉钉10分钟后未恢复则追加发邮件30分钟后仍未恢复直接打电话。这可以在告警管理器里扩展实现。数据持久化与可视化上面的脚本只是把数据存在内存里重启就没了。对于长期运维建议把metrics_history里的数据定期写入数据库比如InfluxDB然后配合Grafana做个监控大盘。这样不仅能看实时状态还能分析历史趋势提前发现潜在问题。5. 总结这套用Python编写的模型服务监控脚本在我们自己的AIGlasses_for_navigation项目里已经稳定运行了小半年。它最大的好处就是“够用”且“可控”没有引入复杂的重型监控系统所有逻辑都透明可见出了问题也能快速定位和修改。从效果上看它成功地把我们从“被动救火”变成了“主动预警”。好几次GPU显存缓慢泄露、推理延迟因异常输入陡增的情况都是监控脚本提前发现并告警让我们能在用户感知到问题之前就完成处理。如果你觉得这个脚本对你有用可以直接拿去用。代码里留了不少扩展点比如告警渠道可以很方便地增加企业微信、飞书监控指标也可以根据你的模型特点加入自定义的健康检查逻辑。运维工作的自动化就是从这样一个个小工具开始的希望这个分享能帮你省下一些深夜排查问题的时间。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。