从告警到处置：我们如何设计一个 AI 运维助手 OpsPilot

做 OpsPilot 的起点不是“给运维系统接一个大模型”，而是一个更具体的问题：线上告警来了以后，能不能把人工排查里最重复、最依赖上下文的部分收敛成一条可控的工作流？

在真实运维场景里，告警本身往往只是一个入口。Prometheus 告诉你某个服务延迟升高，Grafana 告诉你某条规则触发了，日志系统里可能还有一堆相关报错，Kubernetes 里还要看 Deployment、Pod、Events。处理人需要判断这是不是重复告警、是否影响生产、应该先看指标还是日志、能不能自动重启、是否需要人工审批。

所以 OpsPilot 的目标不是让 LLM 自由发挥，而是把它放进一条工程化链路里：接收告警、标准化、调度、分析、调用工具、安全检查、审批处置、记录结果。LLM 负责推理和归纳，系统负责边界、状态和流程。

总体架构

OpsPilot 当前采用分层架构。上游可以来自 Prometheus、Grafana 或自定义 webhook，也可以通过 Puller 主动拉取 Prometheus、Elasticsearch 这类数据源。进入系统后，所有告警先被归一化成统一的 AlertEvent，再交给调度层做去重、优先级排序和并发处理，最后由 Agent Session 进入 LangGraph 工作流。

这个结构里最重要的一条设计原则是：core 提供机制，builtin 提供策略。

core 里放抽象、注册表、配置模型、调度器、工作流这些框架能力；builtin 里放具体的 Prometheus/Grafana normalizer、Shell/K8s/Notify/Database 工具等实现。这样系统不会把“某一种告警源”或者“某一个工具”硬编码进核心流程，后续接新的数据源或工具时，扩展点会比较清楚。

接入层：先把异构告警统一起来

不同告警源的 payload 差异很大。Prometheus AlertManager 的主体是 alerts[]，Grafana webhook 里有 state、evalMatches，自定义 webhook 又可能是另一个 schema。如果下游每一层都直接处理这些原始 payload，复杂度会很快扩散。

所以接入层的第一件事是把所有来源转换成统一的 AlertEvent：

class AlertEvent(BaseModel):
    alert_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    fingerprint: str = ""

    source: str = ""
    project: str = ""

    status: AlertStatus = AlertStatus.FIRING
    severity: Severity = Severity.WARNING

    alert_name: str = ""
    title: str = ""
    description: str = ""

    metrics: list[MetricSample] = Field(default_factory=list)
    labels: dict[str, str] = Field(default_factory=dict)
    annotations: dict[str, str] = Field(default_factory=dict)

    logs: list[str] = Field(default_factory=list)
    generator_url: str = ""
    related_alerts: list[str] = Field(default_factory=list)

    triggered_at: datetime | None = None
    received_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
    resolved_at: datetime | None = None

    raw_payload: dict | None = None

这个模型是接入层和下游之间的契约。下游调度器不关心它来自 Prometheus 还是 Grafana，只关心 severity、labels、fingerprint、metrics 这些标准字段。与此同时，raw_payload 被保留下来，方便调试和回溯，不会因为标准化丢掉原始信息。

内置 normalizer 目前覆盖三类：

• Prometheus：解析 AlertManager webhook，提取 labels.alertname、labels.severity、annotations.description、fingerprint 等字段。
• Grafana：把 alerting、ok、resolved 映射成统一状态，并把 evalMatches 转成指标样本。
• Generic：使用严格 Pydantic schema，未知字段直接拒绝，适合自定义系统稳定接入。

这里的关键取舍是，接入层不要试图“理解故障”，它只做格式归一化和基础校验。真正的上下文组装、分析和处置放到 Agent 层。

调度层：让告警有序进入处理流程

告警进入系统后不能立即全部扔给 LLM。生产环境里经常会遇到同一个故障连续触发几十条告警，或者低优先级信息把队列占满。如果调度层不处理这些问题，后面的 AI 分析再强也会被噪声拖垮。

OpsPilot 的调度层做了三件事：去重、优先级队列、Worker 池。

去重逻辑优先使用上游系统给出的 fingerprint。如果没有 fingerprint，就基于 source + alert_name + 指定 labels 计算一个 hash：

def _compute_key(self, event: AlertEvent) -> str:
    if event.fingerprint:
        return f"fp:{event.fingerprint}"

    parts = [event.source, event.alert_name]
    for label in self.key_labels:
        parts.append(event.labels.get(label, ""))
    raw = "|".join(parts)
    digest = hashlib.sha256(raw.encode()).hexdigest()[:16]
    return f"hash:{digest}"

默认参与去重的标签是 service 和 namespace，窗口期是 300 秒。也就是说，同一个服务、同一个命名空间、同一个告警名在 5 分钟内重复出现，会被认为是同一类告警。这个策略比较保守：不会因为实例号不同就把同一故障拆成大量独立任务，也不会永久压住后续真正的新事件。

队列使用 asyncio.PriorityQueue，优先级是 critical > warning > info。同级别事件保持 FIFO：

_SEVERITY_PRIORITY: dict[Severity, int] = {
    Severity.CRITICAL: 0,
    Severity.WARNING: 1,
    Severity.INFO: 2,
}

def __lt__(self, other: _QueueItem) -> bool:
    if self.priority != other.priority:
        return self.priority < other.priority
    return self.sequence < other.sequence

队列满时，系统会优先驱逐最旧的 info 告警。如果队列里没有可驱逐的低优先级事件，才拒绝新事件。这比简单地“满了就拒绝”更适合告警系统，因为 critical 告警不应该被低优先级信息阻塞。

WorkerPool 则负责并发消费队列。每个 worker 是一个独立的 asyncio task，单个告警处理失败不会拖垮整个 worker 池。这样调度层既能控制吞吐，又能给下游 Agent 层提供稳定输入。

Agent 层：LLM 不是直接执行命令

OpsPilot 的 Agent 层基于 LangGraph。它不是简单地把告警文本拼给 LLM，然后执行模型返回的命令，而是把分析过程拆成明确节点：组装上下文、LLM 分析、安全检查、审批、工具执行、报告生成。

从代码上看，图的边界非常明确：

builder.add_edge(START, "assemble_context")
builder.add_edge("assemble_context", "llm_analyze")

builder.add_conditional_edges("llm_analyze", _route_after_llm, {
    "check_safety": "check_safety",
    "generate_report": "generate_report",
})

builder.add_conditional_edges("check_safety", _route_after_safety, {
    "approval_node": "approval_node",
    "execute_tools": "execute_tools",
})

builder.add_conditional_edges("approval_node", _route_after_approval, {
    "execute_tools": "execute_tools",
    "llm_analyze": "llm_analyze",
})

这里有一个很重要的边界：LLM 可以提出工具调用，但不能绕过安全检查。安全检查会把工具分为 safe、dangerous、blocked。安全工具可以直接执行，危险工具进入人工审批，被拒绝后会回到 LLM 分析节点，让模型尝试其他方案；超过最大迭代次数则升级人工。

这个设计避免了一个常见误区：以为 AI 运维就是让模型直接跑 shell 或 kubectl。实际上，越靠近生产环境，越需要把“能不能执行”从模型回答里拆出来，变成系统规则。

处置层和存储层：把动作和结果沉淀下来

处置层目前内置了四类工具：Shell、Kubernetes、通知、数据库。Shell 工具有命令黑名单和超时，K8s 工具直接调用 Kubernetes REST API，通知工具支持 Slack、飞书、钉钉、Webhook，数据库工具区分查询类和写入类操作。

工具执行只是闭环的一部分。每次分析、工具调用、审批结果和最终报告都需要被记录，否则系统无法复盘，也无法回答“这次故障到底发生了什么”。因此 OpsPilot 还实现了 SQLite + SQLAlchemy 的存储层，用于保存告警记录、Session 状态、工具调用和审批历史。

Session 是这里的核心编排单元。它保存初始告警、关联告警、实时注入消息、事件时间线和运行结果。后续如果同项目的相关告警进来，可以被合并进正在运行的 session，而不是重新开启一条孤立分析链路。

当前状态和下一步

截至目前，OpsPilot 已经完成了这些核心模块：

• 配置系统：YAML 加载、环境变量替换、全局 actions 和项目级覆盖。
• 接入层：Prometheus、Grafana、Generic normalizer，Webhook 和 Puller。
• 调度层：去重、优先级队列、WorkerPool。
• 决策层：LangGraph 工作流、安全检查、审批中断。
• 处置层：ToolProvider 机制，内置 Shell、K8s、Notify、Database 工具，支持 MCP 扩展。
• 存储层：SQLite async ORM 和 repository。
• Agent Session：实时追踪、关联告警合并、session 持久化。
• CLI：已有 status 命令，后续还需要补全 sessions、approve、alerts、inspect 等命令。

还没有完成的部分也很明确：完整 REST API、Web UI、端到端部署和更完整的 CLI 操作面。这些是产品化之前必须补齐的能力。

回头看这个项目，我认为最重要的不是“接上了 LLM”，而是给 LLM 找到了一个合适的位置。它负责分析、总结和提出操作建议；系统负责把告警标准化、控制调度顺序、检查工具风险、要求人工审批、记录所有状态。

AI 运维系统真正有价值的地方，不是让模型替人乱跑命令，而是把原本分散在指标、日志、集群和人工经验里的判断过程，收敛成一条可解释、可中断、可审计的工程闭环。