AI 告警处理与智能巡检 · 运维手册

Q: AI 如何处理误报告警？

多级过滤：① 规则引擎预过滤（已知误报模式）② LLM 语义分析（上下文理解）③ 人工确认（高优先级告警）。用 few-shot prompt 提高准确率。

Q: AI 巡检报告如何生成？

定时触发：① 收集系统指标 ② 查询日志异常 ③ LLM 综合分析 ④ 生成 Markdown 报告。报告包含：健康评分、风险项、建议措施。

Q: 如何评估 AI 告警处理效果？

核心指标：MTTR 缩短比例、误报率、人工干预频率。每周复盘误判案例，持续优化 prompt 和知识库。

Q: AI 能自动执行修复吗？

低风险操作可以（如清理日志、重启服务），高风险操作（删除数据、修改配置）必须人工确认。设置审批流。

适用场景：用 LLM Agent 自动化处理运维告警、生成巡检报告，降低人工响应成本，缩短 MTTR。

1. AI 告警处理架构

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  告警源      │ ──▶ │  预处理      │ ──▶ │  Agent 核心  │
│ (Prometheus)│     │  (去重/聚合) │     │  (LLM+RAG)  │
└─────────────┘     └─────────────┘     └──────┬──────┘
                                               │
                    ┌──────────────────────────┘
                    │
        ┌───────────┼───────────┬───────────┐
        ▼           ▼           ▼           ▼
   ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐
   │ 工具调用 │ │ 知识库   │ │ 工单系统 │ │ 通知渠道 │
   │ (只读)   │ │ (RAG)   │ │ (Jira)  │ │ (钉钉)  │
   └─────────┘ └─────────┘ └─────────┘ └─────────┘
        │
        ▼
   ┌─────────┐
   │ 人工确认 │ ◀── 高风险操作
   └─────────┘
        │
        ▼
   ┌─────────┐
   │ 执行修复 │
   └─────────┘

2. 告警接收与预处理

2.1 Alertmanager 路由配置

# alertmanager.yml
route:
  group_by: ['alertname', 'instance']
  group_wait: 30s
  group_interval: 5m
  repeat_interval: 4h
  receiver: 'ai_agent'

receivers:
  - name: 'ai_agent'
    webhook_configs:
      - url: 'http://ai-agent:8080/alerts'
        send_resolved: true

2.2 告警去重与聚合

# alert_processor.py
import hashlib
from datetime import datetime, timedelta

class AlertProcessor:
    def __init__(self):
        self.seen_alerts = {}  # hash → last_seen

    def deduplicate(self, alerts):
        """告警去重"""
        unique = []
        for alert in alerts:
            alert_hash = self._hash_alert(alert)
            last_seen = self.seen_alerts.get(alert_hash)

            # 5 分钟内相同告警去重
            if last_seen and (datetime.now() - last_seen) < timedelta(minutes=5):
                continue

            self.seen_alerts[alert_hash] = datetime.now()
            unique.append(alert)

        return unique

    def _hash_alert(self, alert):
        return hashlib.md5(
            f"{alert['alertname']}:{alert['instance']}:{alert['summary']}".encode()
        ).hexdigest()

3. Agent 核心逻辑

3.1 根因分析 Prompt

ROOT_CAUSE_PROMPT = """你是资深运维专家。根据以下告警信息，分析可能的根因：

告警信息：
{alert}

上下文：
{context}

请按以下格式输出：
1. 根因分析：[可能原因 1-3 条]
2. 排查步骤：[具体命令/操作]
3. 修复建议：[短期 + 长期措施]
4. 风险等级：[P0/P1/P2/P3]
"""

3.2 Agent 执行流程

# agent_core.py
class AlertAgent:
    def __init__(self, llm, rag_db, tool_executor):
        self.llm = llm
        self.rag_db = rag_db
        self.tools = tool_executor

    def process(self, alert):
        # 1. 检索相关知识
        context = self.rag_db.search(alert['description'], top_k=3)

        # 2. LLM 根因分析
        analysis = self.llm.invoke(
            ROOT_CAUSE_PROMPT.format(alert=alert, context=context)
        )

        # 3. 判断是否需要工具调用
        if self._needs_tool(analysis):
            result = self.tools.execute(analysis['suggested_actions'])
            analysis['verification'] = result

        # 4. 生成工单
        ticket = self._generate_ticket(alert, analysis)

        # 5. 风险评估
        if ticket['severity'] in ['P0', 'P1']:
            self._notify_human(ticket)  # 通知人工确认
        else:
            self._auto_resolve(ticket)  # 自动处理

        return ticket

4. 工具沙箱设计

4.1 只读工具列表

READONLY_TOOLS = {
    "check_service": "systemctl status {}",
    "query_logs": "journalctl -u {} -n 100",
    "check_metrics": "curl -s http://prometheus:9090/api/v1/query?query={}",
    "disk_usage": "df -h",
    "memory_info": "free -h",
    "process_list": "ps aux | head -20"
}

# 危险操作必须人工确认
DANGEROUS_OPERATIONS = [
    "rm -rf",
    "systemctl restart",
    "docker restart",
    "iptables",
    "chmod 777"
]

4.2 安全执行器

import subprocess
import timeout_decorator

@timeout_decorator.timeout(30)
def safe_execute(cmd: str) -> dict:
    """安全执行命令"""
    # 检查命令白名单
    allowed = any(cmd.startswith(prefix) for prefix in READONLY_TOOLS.keys())
    if not allowed:
        return {"error": "Command not in whitelist"}

    # 检查危险操作
    for danger in DANGEROUS_OPERATIONS:
        if danger in cmd:
            return {"error": f"Dangerous operation detected: {danger}"}

    # 执行并限制输出
    try:
        result = subprocess.run(
            cmd, shell=True, capture_output=True, text=True, timeout=30
        )
        return {
            "stdout": result.stdout[:5000],
            "stderr": result.stderr[:5000],
            "returncode": result.returncode
        }
    except Exception as e:
        return {"error": str(e)}

5. 智能巡检系统

5.1 巡检清单

INSPECTION_CHECKLIST = [
    {"name": "CPU 负载", "command": "uptime", "threshold": 4.0},
    {"name": "内存使用", "command": "free | awk '/Mem/ {printf \"%.1f\", $3/$2 * 100}'", "threshold": 85},
    {"name": "磁盘使用", "command": "df -h / | awk 'NR==2 {print $5}'", "threshold": 90},
    {"name": "关键服务", "command": "systemctl is-active nginx prometheus grafana", "expected": "active"},
    {"name": "证书有效期", "command": "echo | openssl s_client -connect example.com:443 2>/dev/null | openssl x509 -noout -dates", "threshold_days": 30}
]

5.2 巡检报告生成

def generate_inspection_report(check_results):
    """生成巡检报告"""
    report = f"""# 系统巡检报告
日期：{datetime.now().strftime('%Y-%m-%d %H:%M')}

## 健康评分
"""
    passed = sum(1 for r in check_results if r['status'] == 'pass')
    total = len(check_results)
    score = passed / total * 100
    report += f"**{score:.0f}分** ({passed}/{total} 项通过)\n\n"

    report += "## 检查详情\n\n"
    for r in check_results:
        icon = "✅" if r['status'] == 'pass' else "⚠️"
        report += f"- {icon} {r['name']}: {r['value']}\n"

    report += "\n## AI 分析\n"
    analysis = llm.invoke(f"分析以下巡检结果，给出优化建议：\n{check_results}")
    report += analysis

    return report

6. 工单系统集成

6.1 Jira 工单创建

import jira

def create_ticket(alert, analysis):
    jira_client = jira.JIRA('http://jira.example.com', basic_auth=('user', 'pass'))

    issue_dict = {
        'project': {'key': 'OPS'},
        'summary': f"[AI] {alert['alertname']}",
        'description': f"""
## 告警信息
{json.dumps(alert, indent=2)}

## AI 分析
{analysis}
""",
        'issuetype': {'name': 'Incident'},
        'priority': {'name': _severity_to_priority(analysis['severity'])}
    }

    return jira_client.create_issue(fields=issue_dict)

6.2 工单状态同步

def sync_ticket_status(ticket_id, status):
    """同步工单状态"""
    status_map = {
        'auto_resolved': 'Closed',
        'pending_review': 'In Review',
        'needs_attention': 'Open'
    }

    jira_client = jira.JIRA('http://jira.example.com', basic_auth=('user', 'pass'))
    issue = jira_client.issue(ticket_id)
    issue.update(fields={'status': {'name': status_map.get(status, 'Open')}})

7. 效果评估

7.1 核心指标

指标	计算方法	目标值
MTTR 缩短比例	(原 MTTR - 现 MTTR) / 原 MTTR	> 30%
告警处理准确率	正确分类数 / 总数	> 85%
人工干预频率	干预次数 / 总告警数	< 20%
误报率	误报数 / 总告警数	< 10%

7.2 评估脚本

def evaluate_weekly(alerts, tickets):
    """周报评估"""
    metrics = {
        'total_alerts': len(alerts),
        'auto_resolved': sum(1 for t in tickets if t['resolution'] == 'auto'),
        'avg_response_time': np.mean([a['response_time'] for a in alerts]),
        'false_positive_rate': sum(1 for a in alerts if a['is_false_positive']) / len(alerts)
    }
    return metrics

8. 常见问题排查

问题 1：告警风暴

现象：短时间内大量告警涌入

解决： - 启用 Alertmanager 静默和抑制规则 - 实现告警聚合（相同 root cause 合并） - 设置突发率限制

问题 2：误报率高

现象：AI 给出的建议不准确

解决： - 优化 RAG 知识库质量 - 添加 few-shot 示例到 prompt - 收集误报案例持续训练

问题 3：工具执行失败

现象：Agent 调用工具返回错误

解决： - 增加工具返回值的错误处理 - 设置工具调用超时和重试 - 记录工具执行日志便于排查

9. 延伸阅读

FAQ

AI 如何处理误报告警？多级过滤：① 规则引擎预过滤（已知误报模式）② LLM 语义分析（上下文理解）③ 人工确认（高优先级告警）。用 few-shot prompt 提高准确率。

AI 巡检报告如何生成？定时触发：① 收集系统指标 ② 查询日志异常 ③ LLM 综合分析 ④ 生成 Markdown 报告。报告包含：健康评分、风险项、建议措施。

如何评估 AI 告警处理效果？核心指标：MTTR 缩短比例、误报率、人工干预频率。每周复盘误判案例，持续优化 prompt 和知识库。

AI 能自动执行修复吗？低风险操作可以（如清理日志、重启服务），高风险操作（删除数据、修改配置）必须人工确认。设置审批流。