#内容由AI生成

导读：监控系统的“最后一公里”往往是告警通知。传统的告警只告诉你是“什么错了”，却没告诉你“怎么修”。深夜收到一堆告警却不知道哪个最关键？本文将手把手教你在容器化 Zabbix 环境中配置钉钉告警，并集成阿里云百炼千问 AI，实现告警智能分析与建议，让运维效率翻倍。

---

🌟 为什么我们要做这件事？

在企业级运维中，告警风暴已经成了常态。Zabbix 报警不断，消息刷屏，但往往信息不全、上下文缺失，一线运维人员只能“凭感觉”响应。

我们想要的是：

🔥 不只是“告诉你有事”，
💡 更要“告诉你该怎么干”。

我将带大家实现一套，把 Zabbix + 钉钉机器人 + 阿里云百炼千问 AI 三者打通，实现了一个 智能告警通知系统 ——
它能：

• ✅ 自动提取关键信息（IP、主机名、监控项）
• ✅ 调用 AI 智能分析问题原因 & 提供处理建议
• ✅ 格式化为视觉友好、可读性强的 Markdown 消息
• ✅ 支持 @ 指定人员，直达责任人

---

🏗️ 整体架构流程

在开始之前，我们先理清数据流向。整个告警过程如下所示：

流程说明：

1. Zabbix 触发告警，调用本地 Python 脚本。
2. 脚本接收参数，优先调用阿里云 AI 进行分析（若失败则自动降级）。
3. 脚本组装 Markdown 消息，通过钉钉机器人发送。
4. 全过程记录日志，便于审计和排查。

---

📋 前置准备

请确保你拥有以下环境和权限：

1. Zabbix Server：已安装（本文以 Docker 容器/Alpine Linux 环境为例）。
2. 钉钉群机器人：已创建，并获取了 Webhook 地址。
3. 阿里云账号：已开通 百炼/ DashScope 服务，并获取了 API Key。
4. 服务器权限：拥有 root​ 权限用于安装依赖，脚本运行权限属于 zabbix​ 用户。

⚠️ 安全警告：下文代码中的 Webhook Token​ 和 API Key​ 仅为示例，请务必替换为你自己的真实信息，不要直接使用示例中的地址！

---

🚀 第一步：环境准备（解决 Python 依赖）

由于很多 Zabbix 容器基于 Alpine Linux，其 Python 环境遵循 PEP 668 规范，直接使用 pip install​ 会报错。我们需要使用虚拟环境来隔离依赖。

1. 安装 Python 及编译工具

进入 Zabbix 容器或服务器终端，执行以下命令：

# 1. 替换 APK 源为阿里云（加速下载）
sed -i 's/dl-cdn.alpinelinux.org/mirrors.aliyun.com/g' /etc/apk/repositories

# 2. 更新包列表
apk update

# 3. 安装 Python3 及编译依赖
# 解释：libffi-dev, gcc, musl-dev 是为了确保 requests 库能正常编译安装
apk add --no-cache python3 py3-pip libffi-dev gcc musl-dev python3-dev

2. 创建 Python 虚拟环境

为了避免污染系统环境，我们创建一个独立的虚拟环境，并安装必要的 requests​ 库。

# 1. 创建虚拟环境目录
mkdir -p /usr/lib/zabbix/venv

# 2. 初始化虚拟环境
python3 -m venv /usr/lib/zabbix/venv

# 3. 激活环境并安装 requests 库
source /usr/lib/zabbix/venv/bin/activate
pip install requests -i https://mirrors.aliyun.com/pypi/simple/
deactivate  # 安装完成后退出激活状态

# 4. 授权给 zabbix 用户
# 解释：Zabbix 服务通常以 zabbix 用户运行，必须确保它有权限读取虚拟环境
chown -R zabbix:zabbix /usr/lib/zabbix/venv
chmod -R 755 /usr/lib/zabbix/venv

🛠️ 第二步：编写智能告警脚本

我们将脚本放在 Zabbix 默认的告警脚本目录下。该脚本集成了 告警解析、AI 智能分析 和 钉钉消息发送 三大功能。

1. 创建目录与日志文件

# 创建脚本目录
mkdir -p /usr/lib/zabbix/alertscripts/log

# 创建日志文件并授权
touch /usr/lib/zabbix/alertscripts/log/dingding_ai.log
chown -R zabbix:zabbix /usr/lib/zabbix/alertscripts/log
chmod 755 /usr/lib/zabbix/alertscripts/log

2. 完整代码实现

使用 vim​ 创建脚本文件：vim /usr/lib/zabbix/alertscripts/dingding_ai.py​

💡 复制提示：以下为完整可运行代码，请直接复制。记得替换顶部的 WEBHOOK​ 和 API_KEY​，或配置环境变量。

#!/usr/lib/zabbix/venv/bin/python
# -*- coding: utf-8 -*- 
"""
Zabbix 钉钉告警脚本 (Markdown 版 + AI 智能分析)
=====================================================================
功能说明：
1. 接收 Zabbix 告警信息
2. 调用阿里云百炼千问 AI 进行智能分析
3. 格式化告警内容为 Markdown 格式
4. 发送钉钉机器人消息（支持 @ 特定用户）
5. 完善的日志记录和错误处理机制
"""

# ==================== 导入模块 ====================
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import json
import sys
import datetime
import os
import re

# ================= 配置区域 =================
# ⚠️ 优先读取环境变量，如果没有则使用默认值（生产环境建议使用环境变量）
DINGTALK_WEBHOOK = os.getenv("DINGTALK_WEBHOOK", "https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN")
DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY", "YOUR_API_KEY")

DASHSCOPE_API_URL = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation"
AI_MODEL = "qwen-plus"  # 推荐使用 qwen-plus，性价比高

# 网络优化配置
AI_TIMEOUT = 30
AI_RETRY_TIMES = 3
AI_RETRY_BACKOFF = 2
DINGTALK_TIMEOUT = 5
LOG_FILE = "/usr/lib/zabbix/alertscripts/log/dingding_ai.log"
# ===========================================

def write_log(message):
    """安全写入日志，避免因日志失败导致脚本中断"""
    try:
        os.makedirs(os.path.dirname(LOG_FILE), exist_ok=True)
        with open(LOG_FILE, "a+", encoding="utf-8") as f:
            timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            f.write(f"[{timestamp}] {message}\n")
    except Exception as e:
        print(f"Log write error: {e}", file=sys.stderr)

def create_session_with_retry():
    """创建支持自动重试的 requests.Session"""
    session = requests.Session()
    retry_strategy = Retry(
        total=AI_RETRY_TIMES,
        backoff_factor=AI_RETRY_BACKOFF,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["POST"]
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session

def parse_alert_content(raw_content):
    """兼容多种格式的告警内容提取"""
    patterns = {
        '主机': r'主机:\s*([^ ]+)',
        'IP 地址': r'IP 地址:\s*([^ ]+)',
        '监控项': r'监控项:\s*([^:]+):',
        '持续时间': r'持续时间:\s*([^ ]+)',
        '事件 ID': r'事件 ID:\s*(\d+)'
    }

    extracted = {}
    for key, pattern in patterns.items():
        match = re.search(pattern, raw_content)
        if match:
            extracted[key] = match.group(1).strip()

    # 组装 Markdown 内容
    formatted = []
    if '主机' in extracted or 'IP 地址' in extracted:
        formatted.append("**📍 监控对象**")
        if '主机' in extracted:
            formatted.append(f"- **主机名**: {extracted['主机']}")
        if 'IP 地址' in extracted:
            formatted.append(f"- **IP 地址**: `{extracted['IP 地址']}`")
        formatted.append("")

    if '监控项' in extracted:
        formatted.append("**📊 监控信息**")
        formatted.append(f"- **监控项**: {extracted['监控项']}")
        formatted.append("")
  
    if not formatted:
        formatted = [f"> {raw_content}"]

    return "\n".join(formatted)

def call_ai_analyze(alert_title, alert_content):
    """调用阿里云原生 DashScope API"""
    # 检查 API Key 是否配置
    if not DASHSCOPE_API_KEY or "YOUR_API" in DASHSCOPE_API_KEY:
        return None

    system_prompt = """你是一名资深网络运维工程师，监控对象均为网络设备。
请根据告警信息，输出：
1. 告警严重程度评估（紧急/重要/警告/提示）
2. 可能原因分析
3. 建议的处理步骤
4. 预防措施建议
请简洁明了，使用条理化格式，便于快速阅读，不超过 150 字。"""

    user_prompt = f"""告警标题：{alert_title}
告警内容：{alert_content}
告警时间：{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
请分析此告警并给出专业建议。"""

    headers = {
        "Authorization": f"Bearer {DASHSCOPE_API_KEY}",
        "Content-Type": "application/json"
    }

    data = {
        "model": AI_MODEL,
        "input": {
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ]
        },
        "parameters": {
            "result_format": "message",
            "temperature": 0.7,
            "max_tokens": 1000
        }
    }

    session = create_session_with_retry()

    try:
        response = session.post(
            url=DASHSCOPE_API_URL,
            headers=headers,
            data=json.dumps(data),
            timeout=AI_TIMEOUT
        )

        # 检查响应是否为空或 HTML 错误页面
        if not response.text or response.text.strip().startswith('<'):
            return None

        try:
            result = response.json()
        except json.JSONDecodeError:
            return None

        if response.status_code == 200 and result.get("output", {}).get("choices"):
            return result["output"]["choices"][0]["message"]["content"]
        else:
            return None

    except Exception:
        return None
    finally:
        session.close()

def send_dingtalk_alert():
    """主函数：处理 Zabbix 告警并发送钉钉消息"""
    if len(sys.argv) < 4:
        write_log("参数错误：需要 (手机号，标题，内容)")
        sys.exit(1)

    user_phone = sys.argv[1]
    subject = sys.argv[2]
    content = sys.argv[3]

    write_log(f"开始处理告警 | Title: {subject}")

    # 1. 调用 AI
    ai_suggestion = call_ai_analyze(subject, content)
    # 2. 解析内容
    formatted_content = parse_alert_content(content)

    # 3. 构建消息体
    if ai_suggestion:
        markdown_text = (
            f"## 🚨 {subject}\n\n"
            f"**⏰ 时间**: `{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}`\n\n"
            f"---\n\n"
            f"## 📋 告警详情\n\n{formatted_content}\n\n"
            f"---\n\n"
            f"## 🤖 AI 智能分析\n\n{ai_suggestion}\n\n"
            f"---\n"
            f"@{user_phone}"
        )
    else:
        # AI 失败降级处理
        markdown_text = (
            f"## 🚨 {subject}\n\n"
            f"**⏰ 时间**: `{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}`\n\n"
            f"---\n\n"
            f"## 📋 告警详情\n\n{formatted_content}\n\n"
            f"---\n\n"
            f"## ⚠️ AI 分析暂时不可用\n\n"
            f"建议检查：\n1. API Key 是否正确\n2. 账户余额是否充足\n\n"
            f"---\n"
            f"@{user_phone}"
        )

    data = {
        "msgtype": "markdown",
        "markdown": { "title": subject, "text": markdown_text },
        "at": { "atMobiles": [user_phone], "isAtAll": False }
    }

    headers = {'Content-Type': 'application/json'}

    try:
        response = requests.post(
            url=DINGTALK_WEBHOOK,
            data=json.dumps(data),
            headers=headers,
            timeout=DINGTALK_TIMEOUT
        )
        result = response.json()

        if result.get("errcode") == 0:
            write_log(f"SUCCESS | {subject} | AI: {'Yes' if ai_suggestion else 'No'}")
            print("OK")
        else:
            write_log(f"FAILED | {result.get('errcode')}: {result.get('errmsg')}")
            print("FAILED")

    except Exception as e:
        write_log(f"发送异常：{str(e)}")
        print("ERROR")
        sys.exit(1)

if __name__ == "__main__":
    send_dingtalk_alert()

3. 赋予执行权限

关键步骤：脚本必须可执行，且所有者必须是 zabbix​ 用户。

chmod +x /usr/lib/zabbix/alertscripts/dingding_ai.py
chown -R zabbix:zabbix /usr/lib/zabbix/alertscripts/

---

4. 配置 Zabbix 告警媒介

在 Zabbix Web UI 中：

• 告警 → 媒介 → 创建媒介类型
• 类型：脚本​
• 脚本名称：dingding_ai.py​
• 参数：{ALERT.SENDTO} {ALERT.SUBJECT} {ALERT.MESSAGE}​

5. 配置 Zabbix 告警动作

在 Zabbix Web UI 中：

• 告警 → 动作 → 触发器动作→ 创建动作
• 名称：钉钉告警​
• 条件：触发器警示度大于等于告警​
• 操作：自定义消息内容​

主题：{TRIGGER.STATUS}: {TRIGGER.NAME} 消息：主机: {HOST.NAME} IP地址: {HOST.IP} 监控项: {ITEM.NAME}: {ITEM.VALUE} 恢复时间: {EVENT.DATE} {EVENT.TIME} 持续时间: {EVENT.AGE} 事件ID: {EVENT.ID} ##注意！！！消息内容不要分行，否则钉钉告警时模版会乱

6. 配置 Zabbix 告警用户

在 Zabbix Web UI 中：

• 用户 → Admin → 报警媒介→ 添加
• 类型：选择告警媒介→ A钉钉告警​
• 收件人：手机号不用带@​
• 操作：自定义告警通知时间，我设置的是上班时间​

7. 测试脚本告警通知

🧩假设 Zabbix 生成一条告警：

[严重] 主机: web01 IP地址: 192.168.1.108 监控项: CPU使用率 > 80% 持续时间: 5分钟 事件 ID: 12345

系统处理后，在钉钉中收到如下消息：

🚨 CPU 使用率持续超标

⏰ 时间: 2026-04-02 14:30:25​

---

📍 监控对象

• 主机名: web01
• IP 地址: 192.168.1.108​

📊 监控信息

• 监控项: CPU使用率 > 80%

⏱️ 时间信息

• 中断时长: 5分钟
• 事件 ID: #12345

🤖 AI 智能分析
🔴 严重程度：紧急
🔍 可能原因：进程异常占用、定时任务突增、硬件故障征兆
✅ 建议步骤：

1. 登录 192.168.1.108​，运行 top -c​ 查看高负载进程
2. 检查是否有突然启动的脚本或容器
3. 观察过去 10 分钟的 CPU 波动曲线
   🛡️ 预防措施：设置 CPU 模板自动告警阈值，并限制关键进程资源使用

@运维小李

测试命令：

su -s /bin/bash zabbix -c "/usr/lib/zabbix/alertscripts/dingding_ai.py 123456 测试 内容"

查看日志：

验证告警触发：

🧪 使用指南（如何让你的 AI 更“懂行”？）

✅ 优化 AI 提示词（Prompt Engineering）

修改脚本中的 system_prompt​ 内容，能让 AI 输出更符合你的运维风格。

🎯 推荐模板：运维专家风格

system_prompt = """你是一名资深网络与系统工程师，负责保障线上服务的稳定。
请根据以下告警信息，分析：
1. 严重程度（紧急/重要/警告/提示）
2. 可能原因（列出 3 个最可能的）
3. 快速排查步骤（3 步以内，高效优先）
4. 预防建议（1 条可落地的）

要求：语言简洁、条理清晰，控制在 150 字以内，避免术语堆砌。"""

🔧 技巧：你可以定期收集一线工程师的反馈，持续优化提示词，让 AI 越来越“懂你”。

---

🔚 结语：AI 不是替代，而是赋能

“我们不需要更多的报警，
我们需要的是更快、更准、更有方向的响应。”

🧠 记住：
AI 的价值不在于“替代人”，而在于“放大人的能力”。

---

📎 附录：常见问题排查

问题	排查方法
钉钉没收到消息	检查 Webhook 是否正确、是否启用“加签”、是否被防火墙拦截
AI 返回"API Key 无效”	登录阿里云控制台检查 API Key 是否可用、是否过期
日志为空	检查 /usr/lib/zabbix/alertscripts/log/​ 目录权限，是否可写
脚本执行失败，Zabbix 显示 "Failed"	运行脚本命令，查看输出 print​ 内容，或查日志
钉钉消息显示"@ 无效”	检查手机号是否正确，是否在钉钉群内

---

📌 最后提醒（必读）

• 👉 请务必替换 DASHSCOPE_API_KEY​ 和 DINGTALK_WEBHOOK​
• 👉 Docker 用户务必挂载卷，否则容器重启脚本会消失
• 👉 使用虚拟环境避免依赖冲突（推荐 venv​）
• 👉 建议配合 logrotate​ 管理日志文件

---

🎉 你已掌握一个能直接部署的智能告警系统。
从今天起，让每一次报警，都成为一次“提升”的契机。

#内容由AI生成