【AIOPS】Dify+Zabbix：对话式告警查询优化方案

摘要： Dify与Zabbix-MCP集成方案通过自然语言交互实现Zabbix告警智能查询。用户可通过中文/英文描述需求（如“查询最近2小时严重告警”），系统自动识别意图并调用对应API（/alerts/top、/alerts/query或/alerts/nl），返回结构化运维报告。方案支持时间范围过滤、TOP排名统计及开放式分析，内置时间戳转换、严重程度映射等逻辑，输出包含告警概览、资源影响、根

AIOps打工人

1747人浏览 · 2025-12-01 15:37:30

AIOps打工人 · 2025-12-01 15:37:30 发布

dify+zabbix-mcp 对话式Zabbix告警数据查询方案

一、方案概述

1. 核心目标

通过 Dify 平台集成 Zabbix-MCP（工具介绍）接口，实现自然语言对话式查询 Zabbix 告警数据，支持 TOP 排名、时间范围过滤、开放式分析等场景，输出结构化运维分析报告，提升 SRE 团队告警查询效率与数据解读能力。

2. 适用场景

快速查询指定时间范围、主机/主机组、严重程度的告警数据
统计 TOP N 高频/高严重级告警
开放式告警趋势分析、根因排查需求
运维团队日常告警巡检、问题定位场景

3. 核心优势

自然语言交互：支持中文/英文混合自由表达，无需记忆固定查询格式
智能意图识别：自动解析查询意图，匹配最优接口与参数
结构化报告输出：告警统计、资源影响、时间特征等多维度分析
容错机制完善：参数兜底、时区统一、失败重试，确保查询稳定性

二、系统架构与工作流

1. 整体架构

在这里插入图片描述

用户自然语言查询 → Dify 平台 → 意图识别（DeepSeek-V3 CHAT）→ 时间处理模块 → 
Zabbix-MCP API 调用（GET/POST）→ 告警数据返回 → Zabbix 分析（DeepSeek-V3 CHAT）→ 
结构化运维报告 → 反馈给用户

2. 详细工作流步骤

步骤	操作内容	核心说明
1	初始化配置	获取当前时间（格式：%Y-%m-%d %H:%M:%S，时区：Asia/Shanghai），转换为时间戳
2	意图识别	调用 DeepSeek-V3 CHAT，解析用户查询意图，匹配目标接口（/alerts/top /alerts/query /alerts/nl）
3	条件分支判断	- 含 TOP 语义（无时间范围）→ 执行 TOP 查询（GET 接口） - 含时间范围/过滤条件 → 执行时间范围查询（POST 接口） - 开放式分析需求 → 执行 NL 分析（POST 接口） ⚠️ 优先级：时间范围 > TOP 语义
4	时间处理与参数格式化	基于当前时间戳，解析用户查询中的时间范围、数量限制、严重程度等参数，补全默认值并格式校验
5	API 调用	按接口类型发起 HTTP 请求，失败时自动重试 3 次
6	告警数据分析	调用 DeepSeek-V3 CHAT，将原始告警数据转化为结构化运维分析报告
7	结果反馈	直接返回分析报告给用户

三、核心配置说明

3.1 意图识别提示词（优化版）

角色定位

你是 SRE 监控专家，负责解析用户的告警查询意图，在三类 Zabbix 告警接口中选择最优项，并生成可直接调用的标准化参数。

输入变量

当前时间戳：{{current_ts}}（单位：秒，时区：Asia/Shanghai）
用户查询：中文/英文混合自由表达文本

决策规则（优先级：时间范围 > TOP 语义 > 开放式分析）

意图类型	关键词特征	目标接口	请求方式
TOP 排名查询	含 “top”、“前N”、“排名”、“最多”、“最严重”（无时间范围）	/alerts/top	GET
条件过滤查询	含 “最近N分钟/小时/天”、“今天/昨天”、指定主机/主机组、严重程度	/alerts/query	POST
开放式分析	含 “分析”、“趋势”、“根因”、“状态如何”	/alerts/nl	POST

参数约束与默认值

接口	支持参数	默认值	取值规则
/alerts/query	from_ts（时间戳，秒）	current_ts - 3600	时间范围往前推算，需 ≤ to_ts
	to_ts（时间戳，秒）	current_ts	与 from_ts 时区一致（UTC 秒）
	severities（整数数组）	[3,4,5]	0-5 对应不同严重程度，见下表
	limit（整数）	100	1 ≤ limit ≤ 100
	host（字符串）	无	从 “主机:” 后提取有效值
	host_group（字符串）	无	从 “主机组:” 后提取有效值
/alerts/top	by（字符串）	“severity”	仅支持 “severity”（按严重程度）或 “frequency”（按频率）
	limit（整数）	50	1 ≤ limit ≤ 100
/alerts/nl	text（字符串）	“分析当前告警状态”	直接复用用户查询文本（需非空）

关键解析规则（强制遵守）

1. 时间范围解析（统一转换为 UTC 秒）

用户表述	转换规则	示例
最近 N 分钟	from_ts = current_ts - N×60	最近 10 分钟 → current_ts - 600
最近 N 小时	from_ts = current_ts - N×3600	最近 2 小时 → current_ts - 7200
半小时	from_ts = current_ts - 1800	-
今天	from_ts = 当天 00:00:00 时间戳，to_ts = current_ts	今天 → 00:00:00 至当前时间
显式日期范围	完整时间直接转换，省略时间则补 “00:00:00”（开始）/“23:59:59”（结束）	2025-12-01 至 2025-12-02 → 2025-12-01 00:00:00 至 2025-12-02 23:59:59

2. 严重程度映射

用户表述	对应数组	说明
灾难告警	[5]	服务完全不可用
严重告警	[4,5]	核心功能受损
一般告警	[3]	性能下降或功能受限
所有告警	[0,1,2,3,4,5]	包含未分类告警
无指定	[3,4,5]	默认关注中高危告警
带括号数字	直接取数字	WARNING(3) → [3]，CRITICAL(5) → [5]

3. 数量限制解析

用户表述	limit 值	默认值
前 N 条 / top N	N	/alerts/top → 50；/alerts/query → 100

输出规范（仅输出 JSON，无额外解释）

{
  "endpoint": "/alerts/query" | "/alerts/top" | "/alerts/nl",
  "method": "POST" | "GET",
  "params": {
    // 对应接口的标准化参数
  }
}

输出示例（时间范围查询）

{
  "endpoint": "/alerts/query",
  "method": "POST",
  "params": {
    "from_ts": 1732994400,
    "to_ts": 1733080800,
    "severities": [3,4,5],
    "limit": 100,
    "host": "web-01",
    "host_group": "prod"
  }
}

3.2 Zabbix 分析提示词（优化版）

角色定位

作为 SRE 专家，将 Zabbix 告警原始数据转化为清晰、专业、可行动的运维分析报告，为团队提供数据支撑与操作指导。

输入数据源

查询上下文：用户查询文本、分析时间（当前时间）、查询参数（from_ts/to_ts/severities 等）
告警原始数据：Zabbix-MCP API 返回的 JSON 数据（含 items 告警项数组、total 告警总数等）

数据处理规则

时间格式统一：时区固定为中国时区（UTC+8），时间戳转换为 “MM-DD HH:mm” 格式
时间跨度计算：(to_ts - from_ts) ÷ 60 → 分钟数（超过 60 则转换为小时+分钟）
告警密度计算：总告警数 ÷ 时间跨度（小时）→ 保留 1 位小数
状态判断：告警发生时间距当前时间 < 10 分钟 → “活跃”，否则 → “已恢复”

报告固定结构（需完整填充）

1. 基础信息

项目	内容
查询时间范围	[MM-DD HH:mm] 至 [MM-DD HH:mm]（UTC+8）
数据更新时间	[YYYY-MM-DD HH:mm:ss]（UTC+8）
查询条件	严重程度：[对应中文描述]；主机/主机组：[名称]；结果限制：[N] 条

2. 告警统计概览

总体指标：总告警数 [X] 条；时间跨度 [X] 小时 [X] 分钟；告警密度 [X.X] 条/小时
严重程度分布（表格形式）：

严重程度	级别定义	数量	占比
CRITICAL(5)	服务完全不可用	[X]	[X.X]%
ERROR(4)	核心功能受损	[X]	[X.X]%
WARNING(3)	性能下降/功能受限	[X]	[X.X]%
NOTICE(2)	需要关注	[X]	[X.X]%
INFO(1)	信息性告警	[X]	[X.X]%
UNKNOWN(0)	未分类	[X]	[X.X]%

3. 关键告警列表（按严重程度降序，取前 20 条）

序号	严重程度	告警名称	所属主机	发生时间	状态
1	CRITICAL	[告警名称]	[主机名]	[MM-DD HH:mm]	活跃/已恢复
2	ERROR	[告警名称]	[主机名]	[MM-DD HH:mm]	活跃/已恢复

4. 资源影响分析

受影响主机 TOP5：按告警数量降序排列，格式：[主机名] - [X] 条告警
告警类型分布：按告警名称关键词分类（如磁盘空间类、CPU 负载类、服务可用性类），统计各类型数量

5. 时间特征分析

告警高峰期：[时间段]（如 14:00-14:30）- [X] 条告警（占总告警数 [X.X]%）
最早告警时间：[MM-DD HH:mm]；最新告警时间：[MM-DD HH:mm]
持续状态：活跃告警 [X] 条；已恢复告警 [X] 条

6. 根因分析与行动建议

主要问题识别（按影响程度排序）：
1. [问题描述，如 “ELK 主机 CPU 负载持续过高”] - 影响范围：[X] 台主机，持续 [X] 分钟
2. [问题描述，如 “数据磁盘剩余空间不足”] - 影响范围：[X] 台主机
行动建议（分优先级）：
- 立即处理：[具体操作，如 “登录 ELK_172.30.6.249 主机，检查 CPU 占用进程并优化”]
- 监控观察：[关注指标，如 “持续监控磁盘空间变化，每 5 分钟检查一次”]
- 长期优化：[系统性方案，如 “调整 CPU 阈值告警配置，优化 ELK 服务资源分配”]

四、时间处理模块（优化版代码）

功能说明

解析意图识别输出的 JSON 参数，补全默认值、校验参数合法性、统一时间格式，为 API 调用提供标准化参数。

# Copyright (c) 2025 Zabbix-MCP
import time
from datetime import datetime
import json
import re
from typing import Any, Dict, List

def _parse_llm_json(text: str) -> Dict[str, Any]:
    """
    解析 LLM 输出的 JSON 字符串（兼容代码块格式与纯字符串）
    :param text: LLM 输出文本（可能含 ```json 代码块）
    :return: 解析后的字典，解析失败返回空字典
    """
    if isinstance(text, dict):
        return text
    if not isinstance(text, str):
        return {}
    # 匹配 ```json 代码块
    m = re.search(r"```json\s*(.*?)\s*```", text, re.DOTALL | re.IGNORECASE)
    s = m.group(1) if m else text.strip()
    try:
        return json.loads(s)
    except Exception as e:
        print(f"JSON 解析失败：{e}")
        return {}

def _to_int_seconds(v: Any) -> int | None:
    """
    将输入值转换为整数时间戳（秒），兼容毫秒格式
    :param v: 待转换值（可能为字符串、整数）
    :return: 转换后的秒级时间戳，失败返回 None
    """
    try:
        x = int(v)
        # 若数值大于 10^12，视为毫秒级时间戳，转换为秒
        if x > 10**12:
            x = x // 1000
        return x
    except Exception:
        return None

def _default_range(current_time_str: str) -> tuple[int, int]:
    """
    生成默认时间范围（最近 1 小时）
    :param current_time_str: 当前时间字符串（格式：%Y-%m-%d %H:%M:%S）
    :return: (from_ts, to_ts) 秒级时间戳元组
    """
    try:
        now = int(datetime.strptime(current_time_str, "%Y-%m-%d %H:%M:%S").timestamp())
    except Exception:
        now = int(time.time())  # 解析失败时使用当前系统时间
    return now - 3600, now

def main(llm_output: str, current_time_str: str) -> dict:
    """
    主函数：处理 LLM 输出，生成标准化 API 调用参数
    :param llm_output: LLM 意图识别输出文本
    :param current_time_str: 当前时间字符串（格式：%Y-%m-%d %H:%M:%S）
    :return: 包含 api_endpoint、method、params、current_time 的字典
    """
    # 解析 LLM 输出的意图配置
    plan = _parse_llm_json(llm_output)
    endpoint = plan.get("endpoint") or plan.get("api_endpoint") or "/alerts/query"
    method = plan.get("method") or ("GET" if endpoint == "/alerts/top" else "POST")
    params = plan.get("params") or {}

    # 处理 /alerts/query 接口参数
    if endpoint == "/alerts/query":
        # 时间戳解析与兜底
        from_ts = _to_int_seconds(params.get("from_ts"))
        to_ts = _to_int_seconds(params.get("to_ts"))
        if from_ts is None or to_ts is None:
            d_from, d_to = _default_range(current_time_str)
            from_ts = d_from if from_ts is None else from_ts
            to_ts = d_to if to_ts is None else to_ts
        # 纠正时间范围顺序（确保 from_ts ≤ to_ts）
        if from_ts > to_ts:
            from_ts, to_ts = to_ts, from_ts

        # 数量限制解析（1-100 范围内）
        limit = params.get("limit")
        try:
            limit = int(limit) if limit is not None else 100
        except Exception:
            limit = 100
        limit = max(1, min(100, limit))  # 限制最大返回 100 条

        # 严重程度解析（默认 [3,4,5]）
        severities = params.get("severities")
        if not isinstance(severities, list) or not severities:
            severities = [3, 4, 5]
        else:
            # 过滤非整数项
            severities = [int(s) for s in severities if isinstance(s, (int, str)) and str(s).isdigit()]
            severities = severities or [3, 4, 5]

        # 主机/主机组解析（去空格）
        host = params.get("host").strip() if isinstance(params.get("host"), str) else ""
        host_group = params.get("host_group").strip() if isinstance(params.get("host_group"), str) else ""

        # 构造输出参数
        output_params = {
            "from_ts": from_ts,
            "to_ts": to_ts,
            "severities": severities,
            "limit": limit,
        }
        if host:
            output_params["host"] = host
        if host_group:
            output_params["host_group"] = host_group

        return {
            "api_endpoint": "/alerts/query",
            "method": "POST",
            "params": output_params,
            "current_time": current_time_str,
        }

    # 处理 /alerts/top 接口参数
    elif endpoint == "/alerts/top":
        # 排序维度（仅支持 severity/frequency）
        by = params.get("by") if params.get("by") in {"severity", "frequency"} else "severity"
        # 数量限制（1-100 范围内）
        limit = params.get("limit")
        try:
            limit = int(limit) if limit is not None else 50
        except Exception:
            limit = 50
        limit = max(1, min(100, limit))

        return {
            "api_endpoint": "/alerts/top",
            "method": "GET",
            "params": {"by": by, "limit": limit},
            "current_time": current_time_str,
        }

    # 处理 /alerts/nl 接口参数
    else:
        # 分析文本（默认兜底文本）
        text = params.get("text").strip() if isinstance(params.get("text"), str) else ""
        text = text or plan.get("text", "").strip() or "分析当前告警状态"

        return {
            "api_endpoint": "/alerts/nl",
            "method": "POST",
            "params": {"text": text},
            "current_time": current_time_str,
        }

五、HTTP 请求配置

5.1 TOP 查询（/alerts/top）

配置项	内容
请求方式	GET
请求 URL	http://10.34.xxx.xxxx:9998/alerts/top
请求头	Authorization: Bearer read Content-Type: application/json
URL 参数	by：排序维度（severity/frequency） limit：返回数量（1-100）
超时设置	默认（可根据实际调整）
重试机制	失败时自动重试 3 次

5.2 时间范围查询（/alerts/query）

配置项	内容
请求方式	POST
请求 URL	http://10.34.xxx.xxxx:9998/alerts/query
请求头	Authorization: Bearer read Content-Type: application/json
请求体（JSON）	由时间处理模块输出的 params 直接填充
超时设置	默认（可根据实际调整）
重试机制	失败时自动重试 3 次

5.3 开放式分析查询（/alerts/nl）

配置项	内容
请求方式	POST
请求 URL	http://10.34.xxx.xxxx:9998/alerts/nl
请求头	Authorization: Bearer read Content-Type: application/json
请求体（JSON）	{“text”: “用户查询文本或兜底文本”}
超时设置	默认（可根据实际调整）
重试机制	失败时自动重试 3 次

六、典型输出示例

示例 1：查询最近 10 分钟的告警数据

在这里插入图片描述

Zabbix 告警分析报告

生成时间：2025-12-01 15:19:23（UTC+8）

1. 基础信息

查询时间范围：2025-12-01 14:09:23 至 2025-12-01 14:19:23（UTC+8）
数据更新时间：2025-12-01 15:19:23
查询条件：严重程度（WARNING/ERROR/CRITICAL）；结果限制：100 条

2. 告警统计概览

总体指标：总告警数 4 条；时间跨度 10 分钟（0.17 小时）；告警密度 24.0 条/小时
严重程度分布：

严重程度	级别定义	数量	占比
ERROR(4)	核心功能受损	3	75.0%
WARNING(3)	性能下降/功能受限	1	25.0%
其他级别	-	0	0.0%

3. 关键告警列表

序号	严重程度	告警名称	所属主机	发生时间	状态
1	ERROR	CPU 负载 > 阈值	ELK_172.30.6.249	14:14:24	活跃
2	ERROR	CPU 负载 > 阈值	ELK_172.30.6.249	14:15:09	活跃
3	ERROR	CPU 负载 > 阈值	ELK_172.30.6.249	14:10:09	已恢复
4	WARNING	/data 磁盘剩余空间 < 1%	dickhouse-replica-1	14:18:24	活跃

4. 资源影响分析

受影响主机 TOP5：
1. ELK_172.30.6.249 - 3 条告警
2. dickhouse-replica-1 - 1 条告警
告警类型分布：
- CPU 负载类：3 条（75%）
- 磁盘空间类：1 条（25%）

5. 时间特征分析

告警高峰期：14:10-14:15 - 3 条告警（占 75%）
最早告警时间：14:10:09；最新告警时间：14:18:24
持续状态：活跃告警 3 条；已恢复告警 1 条

6. 根因分析与行动建议

主要问题识别：
1. ELK_172.30.6.249 主机 CPU 负载持续过高 - 影响 1 台核心主机，持续 9 分钟
2. dickhouse-replica-1 主机 /data 磁盘空间不足 - 影响 1 台存储主机
行动建议：
- 立即处理：登录 ELK_172.30.6.249 执行 top 命令，排查高 CPU 占用进程（如 ES 服务），必要时重启；登录 dickhouse-replica-1 清理 /data 目录冗余文件
- 监控观察：每 5 分钟检查一次 CPU 负载与磁盘空间，确认问题是否缓解
- 长期优化：调整 ELK 服务资源分配（增加 CPU 核心），设置磁盘空间自动清理脚本（保留 10% 冗余）

示例 2：查询今天 TOP100 的告警数据

在这里插入图片描述

Zabbix 告警分析报告（2025-12-01）

生成时间：2025-12-01 15:20:27（UTC+8）

1. 基础信息

查询时间范围：2025-12-01 00:00:00 至 2025-12-01 15:20:27（UTC+8）
数据更新时间：2025-12-01 15:20:27
查询条件：按严重程度排序；结果限制：100 条

2. 告警统计概览

总体指标：总告警数 100 条；时间跨度 140 分钟（2.3 小时）；告警密度 42.8 条/小时
严重程度分布：

严重程度	级别定义	数量	占比
ERROR(4)	核心功能受损	34	34.0%
WARNING(3)	性能下降/功能受限	25	25.0%
NOTICE(2)	需要关注	18	18.0%
INFO(1)	信息性告警	23	23.0%
UNKNOWN(0)	未分类	0	0.0%

3. 关键告警列表（前 5 条）

序号	严重程度	告警名称	所属主机	发生时间	状态
1	ERROR	{HOST.NAME}_CPU 负载 > 阈值	ELK_172.30.6.249	14:05:12	活跃
2	ERROR	{HOST.NAME}_CPU 负载 > 阈值	ELK_172.30.6.249	14:06:38	活跃
3	ERROR	{HOST.NAME}_服务停止响应	ELK_172.30.6.249	13:45:21	已恢复
4	ERROR	{HOST.NAME}_CPU 负载 > 阈值	ELK_172.30.6.249	14:08:15	活跃
5	WARNING	{HOST.NAME}_内存使用率 > 80%	web-01	13:30:45	已恢复
（后续 15 条按相同格式填充）

4. 资源影响分析

受影响主机 TOP5：
1. ELK_172.30.6.249 - 42 条告警
2. web-01 - 18 条告警
3. db-02 - 15 条告警
4. cache-03 - 12 条告警
5. monitor-01 - 8 条告警
告警类型分布：
- CPU 负载类：34 条（34%）
- 内存使用率类：22 条（22%）
- 服务可用性类：18 条（18%）
- 其他类：26 条（26%）

5. 时间特征分析

告警高峰期：14:00-14:30 - 28 条告警（占 28%）
最早告警时间：00:15:32；最新告警时间：15:18:47
持续状态：活跃告警 35 条；已恢复告警 65 条

6. 根因分析与行动建议

主要问题识别：
1. ELK_172.30.6.249 主机 CPU 负载高频告警 - 影响核心日志分析服务，占总告警数 42%
2. 多台主机内存使用率偏高 - 涉及 web、db 等业务节点，需关注资源瓶颈
行动建议：
- 立即处理：批量检查 web-01、db-02 等主机内存使用情况，释放缓存或扩容；优化 ELK 索引分片配置，降低 CPU 消耗
- 监控观察：设置内存使用率、CPU 负载趋势监控，阈值调整为 75% 预警
- 长期优化：梳理业务高峰期（14:00-14:30）流量特征，制定资源弹性扩容方案