AIOps｜日志告警去重与降噪：基于相似度的告警合并策略

告警洪流问题

在大型运维系统中，一次网络抖动可能触发数百条告警——交换机端口Down告警、服务器不可达告警、应用健康检查失败告警、数据库连接超时告警，海量的告警邮件和短信瞬间涌入值班人员的通知渠道。这种现象被称为"告警洪流"或"告警风暴"，它带来三个严重后果：

1. 告警疲劳：值班人员在短时间内收到大量告警，无法分辨哪些是紧急的、哪些是派生的，导致关键告警被淹没在噪声中
2. 响应延迟：大量低优先级告警占据了注意力和处理时间，真正需要紧急处置的问题反而被推迟
3. 信任丧失：频繁的误报和重复告警让运维团队逐渐忽视告警系统，形成"狼来了"效应

告警去重与降噪的目标是：在告警到达通知渠道之前，自动识别并合并相似告警，将数百条原始告警压缩为几条有意义的摘要通知。核心技术手段是文本相似度计算和告警聚类合并。

文本相似度算法

余弦相似度

余弦相似度通过计算两个向量的夹角余弦值来衡量相似程度，适用于TF-IDF向量化后的文本比较。它的优势是对文本长度不敏感，能够捕捉关键词重叠程度。

import math
from collections import Counter

def tokenize(text):
    """简单分词：适用于英文和中英混合告警文本"""
    import re
    # 提取英文单词和数字
    tokens = re.findall(r'[a-zA-Z_]+|\d+', text.lower())
    # 过滤过短的token
    return [t for t in tokens if len(t) > 1]

def tfidf_vectorize(texts):
    """将文本列表转换为TF-IDF向量"""
    # 统计文档频率
    doc_freq = Counter()
    for text in texts:
        unique_tokens = set(tokenize(text))
        for token in unique_tokens:
            doc_freq[token] += 1

    total_docs = len(texts)
    idf = {token: math.log(total_docs / (1 + freq))
           for token, freq in doc_freq.items()}

    # 构建TF-IDF向量
    vectors = []
    for text in texts:
        tokens = tokenize(text)
        tf = Counter(tokens)
        vec = {token: (count / len(tokens)) * idf.get(token, 0)
               for token, count in tf.items()}
        vectors.append(vec)

    return vectors

def cosine_similarity(vec1, vec2):
    """计算两个稀疏向量的余弦相似度"""
    dot_product = sum(vec1.get(k, 0) * vec2.get(k, 0)
                      for k in set(vec1) & set(vec2))
    norm1 = math.sqrt(sum(v * v for v in vec1.values()))
    norm2 = math.sqrt(sum(v * v for v in vec2.values()))

    if norm1 == 0 or norm2 == 0:
        return 0.0
    return dot_product / (norm1 * norm2)

编辑距离

编辑距离（Levenshtein Distance）衡量两个字符串之间通过插入、删除、替换操作转换所需的最小步数。它适合短文本告警的精确比对，特别是当告警文本只有变量值不同时。

def levenshtein_distance(s1, s2):
    """计算两个字符串的编辑距离"""
    m, n = len(s1), len(s2)
    dp = [[0] * (n + 1) for _ in range(m + 1)]

    for i in range(m + 1):
        dp[i][0] = i
    for j in range(n + 1):
        dp[0][j] = j

    for i in range(1, m + 1):
        for j in range(1, n + 1):
            if s1[i-1] == s2[j-1]:
                dp[i][j] = dp[i-1][j-1]
            else:
                dp[i][j] = 1 + min(dp[i-1][j],    # 删除
                                   dp[i][j-1],    # 插入
                                   dp[i-1][j-1])  # 替换

    return dp[m][n]

def normalized_similarity(s1, s2):
    """归一化编辑距离为0-1之间的相似度分数"""
    max_len = max(len(s1), len(s2))
    if max_len == 0:
        return 1.0
    return 1.0 - levenshtein_distance(s1, s2) / max_len

Jaccard相似度

Jaccard相似度计算两个集合的交集与并集之比，适合基于token集合的快速比对：

def jaccard_similarity(text1, text2):
    """基于token集合的Jaccard相似度"""
    set1 = set(tokenize(text1))
    set2 = set(tokenize(text2))

    intersection = set1 & set2
    union = set1 | set2

    if not union:
        return 0.0
    return len(intersection) / len(union)

算法选择策略

不同相似度算法各有适用场景：

实践中常采用两级策略：先用Jaccard做粗筛快速排除明显不相似的告警对，再用余弦相似度或编辑距离做精细比较。

告警聚类

基于密度的聚类（DBSCAN）

告警聚类不需要预先知道类别数量，DBSCAN是自然选择——它能自动发现告警簇，并将噪声点（孤立告警）单独保留。对于告警去重场景，相似度矩阵替代欧氏距离即可。

import numpy as np
from collections import defaultdict

class AlertClusterer:
    """基于相似度矩阵的告警聚类"""

    def __init__(self, sim_threshold=0.65, min_samples=2):
        self.sim_threshold = sim_threshold
        self.min_samples = min_samples

    def cluster(self, alerts):
        """对告警列表进行聚类
        alerts: [{'id': str, 'text': str, 'source': str, 'severity': str, ...}]
        返回: list of clusters, 每个cluster是alert列表
        """
        n = len(alerts)
        if n == 0:
            return []

        # 预计算TF-IDF向量
        texts = [a.get('text', a.get('title', '')) for a in alerts]
        vectors = tfidf_vectorize(texts)

        # 构建相似度邻接关系
        neighbors = defaultdict(list)
        for i in range(n):
            for j in range(i + 1, n):
                sim = cosine_similarity(vectors[i], vectors[j])
                if sim >= self.sim_threshold:
                    neighbors[i].append(j)
                    neighbors[j].append(i)

        # DBSCAN聚类
        visited = set()
        clusters = []
        noise = []

        for i in range(n):
            if i in visited:
                continue

            # 检查是否为核心点
            if len(neighbors[i]) + 1 < self.min_samples:
                noise.append(i)
                visited.add(i)
                continue

            # 扩展簇
            cluster = [i]
            visited.add(i)

            queue = list(neighbors[i])
            while queue:
                j = queue.pop(0)
                if j in visited:
                    continue
                visited.add(j)
                cluster.append(j)

                if len(neighbors[j]) + 1 >= self.min_samples:
                    for k in neighbors[j]:
                        if k not in visited:
                            queue.append(k)

            clusters.append([alerts[idx] for idx in cluster])

        # 噪声点各自成簇
        for idx in noise:
            clusters.append([alerts[idx]])

        return clusters

    def get_cluster_summary(self, cluster):
        """生成告警簇的摘要信息"""
        if not cluster:
            return {}

        # 统计来源
        sources = Counter(a.get('source', 'unknown') for a in cluster)

        # 选取代表性告警（最长的文本通常信息最全）
        representative = max(cluster, key=lambda a: len(a.get('text', '')))

        # 严重程度取最高
        severity_rank = {'critical': 4, 'high': 3, 'warning': 2, 'info': 1}
        max_severity = max(cluster,
            key=lambda a: severity_rank.get(
                a.get('severity', 'info').lower(), 0))

        return {
            'count': len(cluster),
            'representative_alert': representative,
            'sources': dict(sources),
            'max_severity': max_severity.get('severity', 'info'),
            'first_seen': min(a.get('timestamp', '') for a in cluster),
            'last_seen': max(a.get('timestamp', '') for a in cluster),
        }

聚类效果评估

# 使用示例
alerts = [
    {'id': '1', 'text': 'Host web-01 CPU usage 95% exceeded threshold',
     'source': 'prometheus', 'severity': 'warning'},
    {'id': '2', 'text': 'Host web-02 CPU usage 92% exceeded threshold',
     'source': 'prometheus', 'severity': 'warning'},
    {'id': '3', 'text': 'Host db-01 CPU usage 98% exceeded threshold',
     'source': 'prometheus', 'severity': 'critical'},
    {'id': '4', 'text': 'Memory usage high on web-01 node',
     'source': 'zabbix', 'severity': 'warning'},
    {'id': '5', 'text': 'Disk space low on /var/lib/docker',
     'source': 'node_exporter', 'severity': 'info'},
    {'id': '6', 'text': 'Host web-03 CPU usage 91% exceeded threshold',
     'source': 'prometheus', 'severity': 'warning'},
    {'id': '7', 'text': 'Connection refused to redis-prod-01 port 6379',
     'source': 'healthcheck', 'severity': 'critical'},
    {'id': '8', 'text': 'Connection refused to redis-prod-02 port 6379',
     'source': 'healthcheck', 'severity': 'critical'},
]

clusterer = AlertClusterer(sim_threshold=0.55, min_samples=2)
clusters = clusterer.cluster(alerts)

for i, cluster in enumerate(clusters):
    summary = clusterer.get_cluster_summary(cluster)
    print(f"簇 {i}: {summary['count']}条告警")
    print(f"  代表: {summary['representative_alert']['text'][:60]}")
    print(f"  最高级别: {summary['max_severity']}")
    print(f"  来源: {summary['sources']}")

预期输出中，告警1/2/3/6会被聚为CPU告警簇，7/8被聚为Redis连接拒绝簇，4和5各自独立。

时间窗口合并

滑动窗口去重

告警聚类后，还需要在时间维度上做窗口合并——同一告警簇在短时间窗口内的重复实例只通知一次，同时记录重复次数。这既减少了通知数量，又保留了频率信息供异常判断。

from datetime import datetime, timedelta
from collections import defaultdict

class TimeWindowAlertMerger:
    """基于时间窗口的告警合并器"""

    def __init__(self, window_minutes=5):
        self.window = timedelta(minutes=window_minutes)
        self.pending = defaultdict(list)  # cluster_key -> [alerts]

    def _cluster_key(self, alert):
        """基于来源和严重程度生成粗粒度簇键"""
        return f"{alert.get('source', '')}:{alert.get('severity', '')}"

    def add_alert(self, alert):
        """添加告警到合并器，返回是否应该立即通知"""
        key = self._cluster_key(alert)
        now = datetime.fromisoformat(
            alert.get('timestamp', datetime.now().isoformat()))

        # 清理过期窗口
        self.pending[key] = [
            a for a in self.pending[key]
            if now - datetime.fromisoformat(a.get('timestamp',
                datetime.now().isoformat())) < self.window
        ]

        # 检查窗口内是否已有类似告警
        self.pending[key].append(alert)

        # 如果是窗口内第一条，返回True（需要通知）
        if len(self.pending[key]) == 1:
            return True
        return False

    def flush_window(self, key=None):
        """强制刷新窗口，返回合并后的摘要"""
        if key:
            keys = [key]
        else:
            keys = list(self.pending.keys())

        results = []
        for k in keys:
            alerts = self.pending.get(k, [])
            if not alerts:
                continue

            results.append({
                'cluster_key': k,
                'count': len(alerts),
                'first_time': alerts[0].get('timestamp', ''),
                'last_time': alerts[-1].get('timestamp', ''),
                'alerts': alerts
            })
            self.pending[k] = []

        return results

滚动窗口聚合

对于高频率告警，滚动窗口比滑动窗口更适合——每个固定时间区间输出一次聚合结果：

class RollingAlertWindow:
    """滚动窗口告警聚合"""

    def __init__(self, window_seconds=60):
        self.window_seconds = window_seconds
        self.buffer = []
        self.window_start = datetime.now()

    def add(self, alert):
        """添加告警，返回None（仍在窗口内）或合并结果（窗口结束）"""
        now = datetime.now()
        alert_time = datetime.fromisoformat(
            alert.get('timestamp', now.isoformat()))
        self.buffer.append(alert)

        if (now - self.window_start).total_seconds() >= self.window_seconds:
            result = self._merge()
            self.buffer = []
            self.window_start = now
            return result
        return None

    def _merge(self):
        """合并窗口内所有告警"""
        if not self.buffer:
            return None

        clusterer = AlertClusterer(sim_threshold=0.55)
        clusters = clusterer.cluster(self.buffer)

        merged = []
        for cluster in clusters:
            summary = clusterer.get_cluster_summary(cluster)
            merged.append({
                'count': summary['count'],
                'representative': summary['representative_alert'],
                'sources': summary['sources'],
                'severity': summary['max_severity'],
                'time_range': f"{summary['first_seen']} ~ {summary['last_seen']}"
            })

        return {
            'window_start': self.window_start.isoformat(),
            'total_alerts': len(self.buffer),
            'cluster_count': len(merged),
            'clusters': merged
        }

相似度阈值调优

阈值对效果的影响

相似度阈值是告警合并系统最关键的超参数。阈值过高会导致相似告警无法合并（去重不足），阈值过低会导致不相关的告警被错误合并（过度聚合）。

class ThresholdOptimizer:
    """通过历史告警数据调优相似度阈值"""

    def __init__(self, labeled_alerts):
        """labeled_alerts: [{'text': str, 'label': int}, ...]
        label相同的告警属于同一簇"""
        self.labeled = labeled_alerts

    def evaluate_threshold(self, threshold):
        """评估给定阈值下的合并效果"""
        texts = [a['text'] for a in self.labeled]
        vectors = tfidf_vectorize(texts)
        n = len(texts)

        # 构建预测的簇
        union_find = list(range(n))

        def find(x):
            while union_find[x] != x:
                union_find[x] = union_find[union_find[x]]
                x = union_find[x]
            return x

        def union(x, y):
            px, py = find(x), find(y)
            if px != py:
                union_find[px] = py

        for i in range(n):
            for j in range(i + 1, n):
                sim = cosine_similarity(vectors[i], vectors[j])
                if sim >= threshold:
                    union(i, j)

        # 统计预测簇和真实簇
        pred_labels = [find(i) for i in range(n)]
        true_labels = [a['label'] for a in self.labeled]

        # 计算合并率和误合率
        merged_pairs = 0
        should_merge_pairs = 0
        false_merge_pairs = 0
        total_pairs = 0

        for i in range(n):
            for j in range(i + 1, n):
                total_pairs += 1
                if pred_labels[i] == pred_labels[j]:
                    merged_pairs += 1
                    if true_labels[i] != true_labels[j]:
                        false_merge_pairs += 1
                if true_labels[i] == true_labels[j]:
                    should_merge_pairs += 1

        merge_rate = merged_pairs / should_merge_pairs if should_merge_pairs else 0
        false_rate = false_merge_pairs / merged_pairs if merged_pairs else 0

        return {
            'threshold': threshold,
            'merge_rate': merge_rate,      # 应合并对中实际合并的比例
            'false_merge_rate': false_rate  # 合并对中错误合并的比例
        }

    def find_optimal_threshold(self, thresholds=None):
        """在候选阈值中寻找最佳值"""
        if thresholds is None:
            thresholds = [0.3, 0.4, 0.5, 0.55, 0.6, 0.65, 0.7, 0.75, 0.8]

        results = [self.evaluate_threshold(t) for t in thresholds]

        # 目标：合并率 > 0.9 且误合率 < 0.05
        best = None
        for r in results:
            if r['merge_rate'] >= 0.9 and r['false_merge_rate'] <= 0.05:
                if best is None or r['merge_rate'] > best['merge_rate']:
                    best = r
            elif best is None:
                best = r

        return best, results

动态阈值调整

不同告警来源的文本特征差异较大，固定阈值难以全局最优。可以根据告警来源或类型动态调整阈值：

class DynamicThresholdMerger:
    """按告警来源使用不同阈值的合并器"""

    DEFAULT_THRESHOLD = 0.65

    # 按来源调优的阈值表
    SOURCE_THRESHOLDS = {
        'prometheus': 0.60,   # 监控指标告警模板化程度高
        'zabbix': 0.55,       # Zabbix告警文本更统一
        'log_alert': 0.70,    # 日志告警文本变化大，需更高阈值
        'healthcheck': 0.50,  # 健康检查告警高度模板化
        'security': 0.75,    # 安全告警更谨慎，减少误合
    }

    def get_threshold(self, alert):
        source = alert.get('source', '').lower()
        return self.SOURCE_THRESHOLDS.get(source, self.DEFAULT_THRESHOLD)

    def should_merge(self, alert1, alert2):
        """根据来源动态选择阈值判断是否合并"""
        threshold = min(
            self.get_threshold(alert1),
            self.get_threshold(alert2)
        )
        vec1 = tfidf_vectorize([alert1.get('text', '')])[0]
        vec2 = tfidf_vectorize([alert2.get('text', '')])[0]
        sim = cosine_similarity(vec1, vec2)
        return sim >= threshold

完整告警去重系统

将上述组件组合为完整的告警去重管道：

from collections import Counter

class AlertDeduplicationSystem:
    """端到端告警去重与降噪系统"""

    def __init__(self, window_minutes=5, sim_threshold=0.65,
                 min_cluster_size=2):
        self.window_merger = TimeWindowAlertMerger(window_minutes)
        self.clusterer = AlertClusterer(sim_threshold, min_cluster_size)
        self.merged_count = 0
        self.total_count = 0

    def process_alert(self, alert):
        """处理单条告警，返回通知内容或None（被去重）"""
        self.total_count += 1

        # 第一层：时间窗口去重
        should_notify = self.window_merger.add_alert(alert)

        if should_notify:
            return self._format_notification(alert)
        else:
            self.merged_count += 1
            return None

    def process_batch(self, alerts):
        """批量处理告警，返回合并后的通知列表"""
        self.total_count += len(alerts)
        clusters = self.clusterer.cluster(alerts)

        notifications = []
        for cluster in clusters:
            if len(cluster) >= 2:
                self.merged_count += len(cluster) - 1

            summary = self.clusterer.get_cluster_summary(cluster)
            notification = self._format_cluster_notification(summary)
            notifications.append(notification)

        return notifications

    def _format_notification(self, alert):
        """格式化单条告警通知"""
        return (
            f"[{alert.get('severity', 'INFO').upper()}] "
            f"{alert.get('source', 'unknown')}: "
            f"{alert.get('text', alert.get('title', ''))}"
        )

    def _format_cluster_notification(self, summary):
        """格式化告警簇摘要通知"""
        alert = summary['representative_alert']
        count = summary['count']
        sources = ', '.join(f"{k}({v})" for k, v
                           in summary['sources'].items())

        header = f"[{summary['max_severity'].upper()}] 告警合并 ({count}条)"
        body = (
            f"来源: {sources}\n"
            f"时间: {summary['first_seen']} ~ {summary['last_seen']}\n"
            f"代表告警: {alert.get('text', alert.get('title', ''))}\n"
        )
        return f"{header}\n{body}"

    def get_stats(self):
        """获取去重统计"""
        return {
            'total_alerts': self.total_count,
            'deduplicated': self.merged_count,
            'dedup_rate': (self.merged_count / self.total_count
                          if self.total_count > 0 else 0)
        }

运行示例

system = AlertDeduplicationSystem(
    window_minutes=5, sim_threshold=0.55)

# 模拟一次故障引发的告警风暴
alert_storm = [
    {'text': 'Host web-01 CPU 95% exceeded 80%', 'source': 'prom',
     'severity': 'warning', 'timestamp': '2026-09-07T14:32:01'},
    {'text': 'Host web-02 CPU 93% exceeded 80%', 'source': 'prom',
     'severity': 'warning', 'timestamp': '2026-09-07T14:32:02'},
    {'text': 'Host web-03 CPU 91% exceeded 80%', 'source': 'prom',
     'severity': 'warning', 'timestamp': '2026-09-07T14:32:03'},
    {'text': 'HTTP 500 rate on api-gateway 12%', 'source': 'log',
     'severity': 'critical', 'timestamp': '2026-09-07T14:32:10'},
    {'text': 'HTTP 500 rate on api-gateway 15%', 'source': 'log',
     'severity': 'critical', 'timestamp': '2026-09-07T14:32:15'},
    {'text': 'Database connections exhausted', 'source': 'db_mon',
     'severity': 'critical', 'timestamp': '2026-09-07T14:32:20'},
]

notifications = system.process_batch(alert_storm)

print("=== 合并后的通知 ===")
for n in notifications:
    print(n)
    print("---")

stats = system.get_stats()
print(f"\n去重率: {stats['dedup_rate']:.1%} "
      f"({stats['deduplicated']}/{stats['total_alerts']} 条被合并)")

小结

告警去重与降噪是AIOps体系中最直接见效的能力之一。通过文本相似度算法（余弦相似度、编辑距离、Jaccard）度量告警文本之间的语义距离，配合密度聚类算法自动发现告警簇，再叠加时间窗口合并策略，可以将一次故障引发的上百条告警压缩为几条精准的摘要通知。相似度阈值的调优是系统效果的关键，建议基于标注历史数据寻找合并率和误合率的平衡点，并按告警来源使用差异化阈值。整套系统可以部署为Alertmanager和通知系统之间的独立中间层，在不改动现有告警规则的前提下显著降低告警噪声。