大模型运维实战：DeepSeek-R1服务健康检查脚本编写

你有没有遇到过这样的情况：线上部署的AI模型服务突然“失联”，前端请求全部超时，但没人第一时间发现？等用户反馈了才去排查，结果发现是服务进程意外退出、GPU显存溢出或者端口被占用。这类问题在大模型运维中并不少见，尤其像 DeepSeek-R1-Distill-Qwen-1.5B 这种依赖GPU资源的推理服务，稳定性至关重要。

本文将带你从零构建一个实用的服务健康检查脚本，专为 DeepSeek-R1 系列模型设计。它不仅能自动检测服务是否存活，还能监控关键资源状态，发现问题时及时告警或尝试自愈。无论你是刚部署完模型的新手，还是负责维护多个AI服务的运维工程师，这套方案都能帮你把被动“救火”变成主动防御。

1. 为什么需要健康检查脚本？

1.1 模型服务的脆弱性

虽然我们通过 Gradio 快速搭建了 Web 接口，但这种服务本质上是一个长期运行的 Python 进程。它可能因为以下原因中断：

• GPU 显存不足导致 OOM（Out of Memory）崩溃
• 模型加载失败或路径错误
• 端口被其他进程占用
• 服务器重启后服务未自动启动
• 长时间运行出现内存泄漏

一旦服务挂掉，除非有人主动访问或调用接口，否则很难第一时间察觉。而等到业务受影响再处理，已经晚了。

1.2 健康检查的核心目标

一个合格的健康检查脚本应该具备以下能力：

• 可达性检测：服务端口是否监听？HTTP 接口能否响应？
• 功能可用性验证：不只是端口通，还要确认模型能正常推理
• 资源监控：GPU 使用率、显存占用、CPU 和内存状态
• 自动恢复机制：发现异常时尝试重启服务
• 告警通知：通过日志、邮件或消息推送告知管理员

接下来，我们就一步步实现这样一个脚本。

2. 环境准备与依赖安装

2.1 基础环境确认

确保你的系统满足以下条件：

# 检查 Python 版本 python3 --version # 输出应为 Python 3.11 或更高 # 检查 CUDA 是否可用 nvidia-smi # 应能看到 GPU 信息和驱动版本

2.2 安装健康检查所需依赖

除了模型本身的torch、transformers和gradio，我们还需要一些额外工具来辅助监控：

pip install requests psutil GPUtil

• requests：用于发送 HTTP 请求测试接口连通性
• psutil：获取系统 CPU、内存、进程信息
• GPUtil：专门用于查询 GPU 状态

如果你使用 Docker 部署，记得把这些包也加入Dockerfile。

3. 编写核心健康检查逻辑

3.1 定义服务配置参数

我们先创建一个配置文件config.py，集中管理所有可变参数：

# config.py SERVICE_URL = "http://localhost:7860" HEALTH_CHECK_INTERVAL = 60 # 检查间隔（秒） MAX_RESTART_ATTEMPTS = 3 # 最大重试次数 LOG_FILE = "/var/log/deepseek_health.log" APP_SCRIPT = "/root/DeepSeek-R1-Distill-Qwen-1.5B/app.py" PORT = 7860

这样后续修改更方便，避免硬编码。

3.2 实现基础连通性检测

创建health_check.py，第一步先测试服务端口是否开放：

import socket def is_port_open(host, port): """检查指定端口是否可连接""" try: with socket.create_connection((host, port), timeout=5): return True except (socket.timeout, ConnectionRefusedError): return False # 使用示例 if not is_port_open("localhost", 7860): print("❌ 端口 7860 无法访问，请检查服务是否启动") else: print(" 端口 7860 正常")

这个方法轻量快速，适合高频检测。

3.3 验证接口功能可用性

光有端口不够，还得确认服务能真正处理请求。我们可以模拟一次简单的推理调用：

import requests def is_service_healthy(url): """通过 API 调用验证服务功能是否正常""" try: # 发送一个简单的推理请求 response = requests.post( f"{url}/predict", json={ "data": [ "你好，请介绍一下你自己。", None, None ] }, timeout=30 ) if response.status_code == 200: result = response.json() if "data" in result and len(result["data"]) > 0: return True except Exception as e: print(f"请求失败: {e}") return False # 测试 if is_service_healthy("http://localhost:7860"): print(" 服务功能正常") else: print("❌ 服务无响应或返回异常")

注意：Gradio 的/predict接口结构依赖前端组件顺序，这里假设输入框是第一个字段。实际需根据你的app.py中的gr.Interface定义调整。

3.4 监控 GPU 与系统资源

使用GPUtil和psutil获取关键指标：

import GPUtil import psutil def get_system_status(): """获取当前系统资源使用情况""" # GPU 状态 gpus = GPUtil.getGPUs() gpu_info = [] for gpu in gpus: gpu_info.append({ "id": gpu.id, "name": gpu.name, "load": f"{gpu.load*100:.1f}%", "memory_used": f"{gpu.memoryUsed}MB", "memory_total": f"{gpu.memoryTotal}MB", "temperature": f"{gpu.temperature}°C" }) # CPU 和内存 cpu_percent = psutil.cpu_percent() memory = psutil.virtual_memory() mem_percent = memory.percent return { "gpu": gpu_info, "cpu_usage": f"{cpu_percent}%", "memory_usage": f"{mem_percent}%", "timestamp": __import__('datetime').datetime.now().strftime("%Y-%m-%d %H:%M:%S") } # 打印状态 status = get_system_status() for gpu in status["gpu"]: print(f"GPU {gpu['id']}: {gpu['name']} | " f"显存 {gpu['memory_used']}/{gpu['memory_total']} | " f"温度 {gpu['temperature']}°C")

这部分信息对排查性能瓶颈非常有用。

4. 构建完整健康检查脚本

现在我们将所有功能整合成一个完整的守护脚本：

# health_check.py import time import logging import subprocess from config import * # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(LOG_FILE), logging.StreamHandler() ] ) def start_service(): """启动模型服务""" cmd = f"nohup python3 {APP_SCRIPT} > /tmp/deepseek_web.log 2>&1 &" result = subprocess.run(cmd, shell=True) if result.returncode == 0: logging.info(" 服务已成功启动") time.sleep(10) # 等待服务初始化 return True else: logging.error("❌ 启动服务失败") return False def stop_service(): """停止现有服务进程""" cmd = f"ps aux | grep 'python3 {APP_SCRIPT}' | grep -v grep | awk '{{print $2}}' | xargs kill" try: subprocess.run(cmd, shell=True, check=True) logging.info("🛑 服务已停止") except subprocess.CalledProcessError: pass # 可能没有进程在运行 def main(): restart_count = 0 logging.info(" 开始健康检查...") while True: # 1. 检查端口 if not is_port_open("localhost", PORT): logging.warning(" 端口未开放，服务可能已中断") if restart_count < MAX_RESTART_ATTEMPTS: logging.info(" 尝试重启服务...") stop_service() if start_service(): restart_count += 1 else: logging.error("💥 重启失败，等待下次检查") else: logging.critical("🚨 达到最大重试次数，请手动排查问题") continue # 2. 检查接口功能 if not is_service_healthy(SERVICE_URL): logging.warning(" 服务无响应，尝试重启") if restart_count < MAX_RESTART_ATTEMPTS: stop_service() if start_service(): restart_count += 1 continue # 3. 记录资源状态（仅首次或每分钟） status = get_system_status() if restart_count == 0: logging.info(f" 系统状态: CPU {status['cpu_usage']}, " f"内存 {status['memory_usage']}, " f"GPU 显存 {status['gpu'][0]['memory_used'] if status['gpu'] else 'N/A'}") # 重置重试计数 restart_count = 0 logging.info(" 服务健康，一切正常") # 等待下一次检查 time.sleep(HEALTH_CHECK_INTERVAL) if __name__ == "__main__": main()

5. 自动化部署与后台运行

5.1 设置脚本权限并后台运行

# 添加执行权限 chmod +x health_check.py # 后台运行健康检查脚本 nohup python3 health_check.py > /var/log/health_check.log 2>&1 &

5.2 使用 systemd 实现开机自启（推荐）

创建系统服务文件/etc/systemd/system/deepseek-health.service：

[Unit] Description=DeepSeek-R1 Health Check Service After=network.target [Service] Type=simple User=root WorkingDirectory=/root/DeepSeek-R1-Distill-Qwen-1.5B ExecStart=/usr/bin/python3 /root/DeepSeek-R1-Distill-Qwen-1.5B/health_check.py Restart=always RestartSec=10 [Install] WantedBy=multi-user.target

启用服务：

# 重新加载配置 sudo systemctl daemon-reexec # 启用开机自启 sudo systemctl enable deepseek-health.service # 启动服务 sudo systemctl start deepseek-health.service # 查看状态 sudo systemctl status deepseek-health.service

这样即使服务器重启，健康检查也会自动运行。

6. 增强建议与最佳实践

6.1 添加告警通知

你可以扩展脚本，在发现问题时发送通知：

• 邮件告警：使用smtplib
• 企业微信/钉钉机器人：通过 Webhook 发送消息
• 短信通知：集成第三方短信平台 API

例如发送钉钉告警：

def send_dingtalk_alert(message): webhook = "https://oapi.dingtalk.com/robot/send?access_token=xxx" data = {"msgtype": "text", "text": {"content": f"[DeepSeek健康检查] {message}"}} requests.post(webhook, json=data)

6.2 结合 Prometheus + Grafana 监控

如果你想做可视化监控，可以将指标暴露为 Prometheus 格式：

from prometheus_client import start_http_server, Gauge # 定义指标 GPU_MEMORY_USED = Gauge('gpu_memory_used_mb', 'GPU memory used in MB') HTTP_REQUEST_SUCCESS = Gauge('http_request_success', 'Last request success') # 在循环中更新 GPU_MEMORY_USED.set(gpu.memoryUsed) HTTP_REQUEST_SUCCESS.set(1 if is_service_healthy() else 0) # 启动 metrics 服务 start_http_server(8000)

然后用 Grafana 展示趋势图。

6.3 日志轮转防止磁盘占满

使用logrotate管理日志文件大小：

# /etc/logrotate.d/deepseek /var/log/deepseek*.log { daily missingok rotate 7 compress delaycompress notifempty create 644 root root }

7. 总结

7.1 我们完成了什么

本文从实际运维需求出发，为你打造了一套完整的DeepSeek-R1 模型服务健康检查方案：

• 实现了端口检测、接口功能验证、资源监控三位一体的检查机制
• 编写了可落地的 Python 脚本，支持自动重启和日志记录
• 提供了 systemd 集成方案，确保检查服务持久运行
• 给出了告警、监控、日志管理等增强建议

这套方案不仅适用于 DeepSeek-R1-Distill-Qwen-1.5B，稍作修改即可用于其他基于 Gradio 或 FastAPI 的 AI 服务。

7.2 下一步可以做什么

• 将脚本打包为 Docker 镜像，统一部署环境
• 集成到 CI/CD 流程中，实现发布后自动健康检查
• 为多个模型服务编写统一的监控面板
• 增加性能基准测试，定期评估推理延迟变化

记住，一个好的 AI 系统不仅是“能用”，更要“稳用”。希望这个脚本能帮你少熬几次夜，多睡几个安稳觉。

获取更多AI镜像

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