【AIOps】Prometheus/夜莺接入DeepSeek大模型

前言

前面写过使用夜莺作为 Prometheus 告警引擎的文章，在此基础上接入 DeepSeek 进行故障分析，给出可能引起故障的原因和排查方法，大大降低平均故障处理时间（MTTR）。工作流程：Prometheus周期采集 Exporter的指标并存储在本地，夜莺周期查询 Prometheus中的指标是否符合故障告警规则，在产生故障后将故障信息发送给 DeepSeek ，DeepSeek通过分析给出的故障原因和处理建议，拼接原告警信息和 DeepSeek分析结果一并发给用户。对于没有使用夜莺的小伙伴可以直接使用 webhook来执行脚本。

流程图

AIOps Primetheus接入DeepSeek

1. 新建通知模板

告警通知 --> 通知模板 --> 新增, 新增 aiops模板

#### {{if .IsRecovered}}<font color="#008800">💚 {{.RuleName}}恢复</font>{{else}}<font color="#FF0000">💔 {{.RuleName}}告警</font>{{end}}

---
**级别状态**：{{if .IsRecovered}}<font color="#008800">S{{.Severity}}</font>{{else}}<font color="#FF0000">S{{.Severity}}</font>{{end}}   
{{if eq (index .TagsMap "job") "web_status"}}   
**归属公司**：{{index .TagsMap "company"}}   
**项目名称**：{{index .TagsMap "project_cn"}}   
**系统名称**：{{index .TagsMap "name"}}   
{{if .IsRecovered}}**恢复内容**：{{index .TagsMap "name"}} 当前已恢复正常！   
{{else}}**告警内容**：{{index .TagsMap "name"}} 当前无法访问！   
{{end}}
**系统地址**：[{{index .TagsMap "instance"}}]({{index .TagsMap "instance"}})   
{{end}}
{{if .IsRecovered}}**触发时间**：{{timeformat .FirstTriggerTime}}   
**恢复时间**：{{timeformat .LastEvalTime}}{{else}}**触发时间**：{{timeformat .FirstTriggerTime}}{{end}}

2. 新建通知媒介

告警通知 --> 通知设置 --> 通知媒介 --> 添加

名称：aiops
标识：aiops

3. 配置通知脚本

告警通知 --> 通知设置 --> 通知脚本 --> 使用脚本

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
import sys
import json
import requests

class Sender(object):
    @classmethod
    def send_email(cls, payload):
        # already done in go code
        pass

    @classmethod
    def send_wecom(cls, payload):
        # already done in go code
        pass
	# 钉钉机器人
    DINGTALK_URL = "https://oapi.dingtalk.com/robot/send?access_token=XXXXXXXXXX"
	# DeepSeek key
    DEEPSEEK_URL = "https://api.deepseek.com/v1/chat/completions"
    DEEPSEEK_KEY = "sk-XXXXXXXXXX"

    @classmethod
    def call_deepseek(cls, message):
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {cls.DEEPSEEK_KEY}"
        }
      
        data = {
            "model": "deepseek-chat",
            "messages": [{
                "role": "user",
                "content": f"""
告警信息：{message} 
你是一名运维领域的专家，请分析告警信息给出可能原因、处理建议和紧急程度
排版要求：AI故障分析标题为蓝色h4大小，语言简洁突出重点
"""
            }]
        }

        try:
            response = requests.post(cls.DEEPSEEK_URL, headers=headers, json=data)
            response.raise_for_status()
            return response.json()['choices'][0]['message']['content']
        except Exception as e:
            print(f"Deepseek API error: {str(e)}")
            return "无法获取处理建议"

    @classmethod
    def send_dingtalk(cls, payload):
        original_message = payload.get('tpls').get("dingtalk", "dingtalk not found")
        analysis = cls.call_deepseek(original_message)
      
        final_message = f"""{original_message}

---
{analysis}
"""
      
        headers = {
            "Content-Type": "application/json;charset=utf-8"
        }
      
        body = {
            "msgtype": "markdown",
            "markdown": {
                "title": "告警通知",
                "text": final_message
            }
        }

        response = requests.post(cls.DINGTALK_URL, headers=headers, data=json.dumps(body))
        print(f"notify_dingtalk: status_code={response.status_code} response_text={response.text}")

    @classmethod
    def send_mm(cls, payload):
        # already done in go code
        pass

    @classmethod
    def send_sms(cls, payload):
        pass

    @classmethod
    def send_voice(cls, payload):
        pass

def main():
    payload = json.load(sys.stdin)
    with open(".payload", 'w') as f:
        f.write(json.dumps(payload, indent=4))
    for ch in payload.get('event').get('notify_channels'):
        send_func_name = "send_{}".format(ch.strip())
        if not hasattr(Sender, send_func_name):
            print("function: {} not found", send_func_name)
            continue
        send_func = getattr(Sender, send_func_name)
        send_func(payload)

def hello():
    print("hello nightingale")

if __name__ == "__main__":
    if len(sys.argv) == 1:
        main()
    elif sys.argv[1] == "hello":
        hello()
    else:
        print("I am confused")

DeepSeek 分析回答需要一定时间，超时时间建议配置 120s。

4. 告警规则启用 AIOps 通知方式

告警管理 --> 告警规则 ，原有的规则勾上 aiops通知媒介

5. 告警测试

原告警信息
接入大模型后的告警信息

在这里只是简单的将告警信息发送给 DeepSeek 大模型进行分析然后将分析的结果发送给我们，分析的结果一般没办法完全符合我们的预期。下一步需要进行RAG(检索增强生成)，通过检索外部知识库的方式，将知识库中的相关内容作为提示输入给大模型，从而给出更加符合我们预期的分析结果。在 AIOps建设的过程中，运维知识库的建设将非常重要，RAG的结果符合我们的预期后，再结合工作流、AI Agent的方式进行告警收敛、故障自愈等操作。相关开源工具 --- Dify