SRE Agent 运维落地实践
SRE Agent 运维落地实践
一句话：SRE Agent 不是”让 AI 替你运维”，而是让 AI 做运维中重复的、有规则的、信息密集的部分——告警归因、日志分析、排查辅助、知识检索——把 SRE 从琐事中解放出来做架构改进。
概述
JD 第 4 条提到”SRE Agent 等 AI 应用在工作中的提效与落地”。你的 vault 有极强的 AI Agent 基础（30+ 篇 Agent 开发系列、Claude Code 架构、kagent 详解、MCP 协议、工具封装 SOP），但缺少在运维场景的具体落地。本篇把 AI Agent 能力和 SRE 场景对接。
SRE Agent 架构
graph TD
subgraph 输入层
ALERT[告警事件]
LOG[日志流]
METRIC[指标异常]
TICKET[工单请求]
end
subgraph Agent 核心
PERCEIVE[感知模块<br/>告警/日志/指标归一化]
REASON[推理模块<br/>LLM + 知识库 + 规则]
ACT[行动模块<br/>工具调用 + 操作执行]
LEARN[学习模块<br/>复盘总结 + 知识更新]
end
subgraph 工具层
K8S[kubectl 工具]
GRAF[Grafana 查询]
LOKI[Loki/ELK 查询]
ARGO[ArgoCD 操作]
RUN[Runbook 执行]
end
subgraph 知识层
KB[SRE 知识库<br/>故障案例 / Runbook / 架构文档]
RCA_DB[历史 RCA 库]
SLO_DB[SLO/SLI 定义]
end
ALERT --> PERCEIVE
LOG --> PERCEIVE
METRIC --> PERCEIVE
TICKET --> PERCEIVE
PERCEIVE --> REASON
REASON --> ACT
ACT --> LEARN
LEARN --> KB
REASON --> KB
REASON --> RCA_DB
REASON --> SLO_DB
ACT --> K8S
ACT --> GRAF
ACT --> LOKI
ACT --> ARGO
ACT --> RUN落地场景一：告警智能分析
问题描述
告警风暴时 On-Call 被淹没——一条告警触发后，级联告警可能产生几十条通知。On-Call 需要快速判断：
- 哪条是根因告警？
- 影响范围多大？
- 应该先做什么？
Agent 实现方案
from dataclasses import dataclass
from typing import Optional
@dataclass
class AlertEvent:
alert_id: str
alert_name: str
severity: str # critical / warning / info
service: str
namespace: str
message: str
timestamp: str
labels: dict
annotations: dict
related_alerts: list[str] # 关联告警 ID 列表
class AlertAnalysisAgent:
"""告警智能分析 Agent"""
def __init__(self, llm_client, knowledge_base, grafana_api):
self.llm = llm_client
self.kb = knowledge_base
self.grafana = grafana_api
def analyze(self, alert: AlertEvent) -> dict:
"""分析告警，输出归因和建议"""
# Step 1: 收集上下文
context = self._gather_context(alert)
# Step 2: 检索历史相似案例
similar_cases = self.kb.search_similar_incidents(
service=alert.service,
symptoms=alert.message,
top_k=3
)
# Step 3: 检索相关 Runbook
runbooks = self.kb.search_runbooks(
alert_name=alert.alert_name,
service=alert.service
)
# Step 4: LLM 推理
prompt = self._build_analysis_prompt(alert, context, similar_cases, runbooks)
analysis = self.llm.chat(prompt)
# Step 5: 输出结构化结果
return {
"root_cause_hypothesis": analysis.root_cause,
"impact_assessment": analysis.impact,
"confidence": analysis.confidence,
"recommended_actions": analysis.actions,
"related_runbook": runbooks[0] if runbooks else None,
"similar_incidents": similar_cases,
"auto_remediate": analysis.can_auto_remediate,
}
def _gather_context(self, alert: AlertEvent) -> dict:
"""收集告警上下文信息"""
return {
"recent_changes": self._get_recent_changes(alert.service),
"current_metrics": self.grafana.get_metrics(alert.service),
"pod_status": self._get_pod_status(alert.namespace, alert.service),
"recent_events": self._get_k8s_events(alert.namespace),
"dependency_health": self._check_dependencies(alert.service),
}
def _build_analysis_prompt(self, alert, context, cases, runbooks):
return f"""
你是一个 SRE 告警分析助手。请基于以下信息分析告警。
## 告警信息
- 名称: {alert.alert_name}
- 严重程度: {alert.severity}
- 服务: {alert.service}
- 命名空间: {alert.namespace}
- 消息: {alert.message}
- 时间: {alert.timestamp}
- 关联告警数: {len(alert.related_alerts)}
## 上下文
- 近期变更: {context['recent_changes']}
- 当前指标: {context['current_metrics']}
- Pod 状态: {context['pod_status']}
- K8s 事件: {context['recent_events']}
- 依赖健康度: {context['dependency_health']}
## 历史相似案例
{self._format_cases(cases)}
## 相关 Runbook
{self._format_runbooks(runbooks)}
## 请输出
1. 根因假设（含置信度 0-100%）
2. 影响面评估（受影响用户/服务/SLI 违反情况）
3. 建议操作（按优先级排序）
4. 是否可以自动修复（如果是，给出自愈方案）
"""输出示例
{
"root_cause_hypothesis": "新版本 v2.3.1 内存泄漏导致 Pod OOM，级联触发上游 502",
"confidence": 85,
"impact_assessment": {
"affected_service": "api-gateway",
"slo_violation": "成功率降至 97.5%（SLO 99.9%）",
"estimated_affected_users": 12000,
"error_budget_burn": "2.4% in 15min"
},
"recommended_actions": [
{"priority": "P0", "action": "回滚到 v2.3.0", "command": "argocd app rollback api-gateway"},
{"priority": "P1", "action": "增加 Pod 内存 Limit 临时缓解", "command": "kubectl patch deployment api-gateway ..."},
{"priority": "P2", "action": "排查 v2.3.1 内存泄漏代码", "link": "https://github.com/.../commit/abc123"}
],
"related_runbook": "runbooks/oom-crash-loop.md",
"similar_incidents": ["INC-2026-0315", "INC-2026-0701"],
"auto_remediate": true,
"auto_remediate_plan": "执行 ArgoCD 回滚到上一个健康版本"
}落地场景二：故障诊断辅助
Agent 串联排查工具链
class DiagnosisAgent:
"""故障诊断 Agent：串联排查工具链"""
def __init__(self, tools):
self.tools = tools # kubectl, grafana, loki, argocd, etc.
self.steps = []
def diagnose(self, symptom: str, namespace: str) -> dict:
"""
输入症状描述，输出诊断结果
例如: symptom="api-gateway 502 错误率 15%", namespace="production"
"""
self.steps = []
# Step 1: 收集基础信息
self._add_step("检查 Pod 状态", self._check_pods(namespace))
self._add_step("检查 K8s 事件", self._check_events(namespace))
self._add_step("检查近期变更", self._check_recent_changes(namespace))
# Step 2: 根据初步信息决定下一步
pod_status = self.steps[-2]["result"]
if pod_status.get("crashloop_pods"):
self._add_step("获取 CrashLoop Pod 日志",
self._get_crashloop_logs(namespace, pod_status["crashloop_pods"]))
# Step 3: 分析日志
logs = self.steps[-1]["result"]
if "OOM" in logs:
self._add_step("检查内存使用趋势", self._check_memory_trend(namespace))
self._add_step("检查近期变更", self._check_changes(namespace))
# Step 4: LLM 分析
analysis = self._llm_analyze(symptom, self.steps)
self._add_step("LLM 综合分析", analysis)
# Step 5: 输出诊断报告
return self._build_report(symptom, self.steps)
def _check_pods(self, ns):
"""调用 kubectl 工具检查 Pod 状态"""
return self.tools.kubectl(
f"get pods -n {ns} -o wide --field-selector status.phase!=Running"
)
def _check_events(self, ns):
"""检查 K8s 事件"""
return self.tools.kubectl(
f"get events -n {ns} --sort-by='.lastTimestamp' | tail -20"
)
def _get_crashloop_logs(self, ns, pods):
"""获取 CrashLoop Pod 的日志"""
results = []
for pod in pods:
logs = self.tools.kubectl(f"logs {pod} -n {ns} --tail=50 --previous")
results.append({"pod": pod, "logs": logs})
return results
def _llm_analyze(self, symptom, steps):
"""用 LLM 分析收集到的所有信息"""
context = "\n".join(
f"Step: {s['name']}\nResult: {s['result']}" for s in steps
)
return self.tools.llm(f"""
症状: {symptom}
已收集信息:
{context}
请分析：
1. 最可能的根因是什么？
2. 下一步应该检查什么？
3. 有哪些已排除的可能性？
""")落地场景三：日志智能分析
异常日志检测
import re
from collections import Counter
class LogAnalysisAgent:
"""日志智能分析：从海量日志中提取异常模式"""
def __init__(self, loki_client, llm_client):
self.loki = loki_client
self.llm = llm_client
def analyze_error_logs(self, service: str, time_range: str = "1h") -> dict:
"""分析错误日志，提取异常模式"""
# Step 1: 拉取错误日志
error_logs = self.loki.query(
f'{{app="{service}"}} |= "ERROR" or |= "PANIC" or |= "FATAL"'
f' | json | line_format "{{.msg}}"'
)
# Step 2: 日志聚类（相似日志归并）
patterns = self._cluster_logs(error_logs)
# Step 3: 频率异常检测
anomalies = self._detect_frequency_anomaly(patterns)
# Step 4: LLM 分析异常模式
if anomalies:
analysis = self._llm_analyze_patterns(anomalies)
return {
"total_errors": len(error_logs),
"unique_patterns": len(patterns),
"anomalies": anomalies,
"llm_analysis": analysis,
}
return {"total_errors": len(error_logs), "anomalies": []}
def _cluster_logs(self, logs: list[str]) -> list[dict]:
"""日志聚类：相似日志模板归并"""
# 用正则把变量部分替换为占位符
def normalize(log: str) -> str:
log = re.sub(r'\d+', '<NUM>', log)
log = re.sub(r'[0-9a-f]{8,}', '<HASH>', log)
log = re.sub(r'\d{4}-\d{2}-\d{2}.*', '<TIME>', log)
log = re.sub(r'\d+\.\d+\.\d+\.\d+', '<IP>', log)
return log.strip()
clusters = Counter(normalize(log) for log in logs)
return [
{"pattern": pattern, "count": count, "example": next(
log for log in logs if normalize(log) == pattern
)}
for pattern, count in clusters.most_common(20)
]
def _detect_frequency_anomaly(self, patterns: list[dict]) -> list[dict]:
"""检测频率异常：某错误突然暴增"""
anomalies = []
for p in patterns:
# 查历史频率
historical = self.loki.query_rate(
pattern=p["pattern"], time_range="7d"
)
current = p["count"]
if historical > 0 and current > historical * 3: # 3 倍以上
anomalies.append({
"pattern": p["pattern"],
"current_count": current,
"historical_avg": historical,
"multiplier": current / historical,
})
return anomalies落地场景四：运维知识检索
RAG 驱动的知识助手
class SREKnowledgeAgent:
"""SRE 知识检索 Agent：RAG + 工具调用"""
def __init__(self, vector_db, llm_client, tools):
self.vdb = vector_db # 向量数据库（存故障案例、Runbook、架构文档）
self.llm = llm_client
self.tools = tools # kubectl, grafana 等
def answer(self, question: str) -> str:
"""回答 SRE 问题"""
# Step 1: 检索相关知识
docs = self.vdb.similarity_search(question, k=5)
# Step 2: 判断是否需要实时数据
if self._needs_realtime_data(question):
realtime = self._fetch_realtime_data(question)
docs.extend(realtime)
# Step 3: LLM 生成答案
context = "\n---\n".join(d["content"] for d in docs)
return self.llm.chat(f"""
基于以下知识回答问题。如果知识不足以回答，说明需要什么额外信息。
## 知识库内容
{context}
## 问题
{question}
## 回答要求
1. 先给出直接答案
2. 如果有对应的 Runbook/操作步骤，附上
3. 如果有历史案例，引用
4. 标注信息来源（知识库 or 实时数据）
""")
def _needs_realtime_data(self, question: str) -> bool:
"""判断问题是否需要实时数据"""
keywords = ["现在", "当前", "状态", "怎么样", "健康", "是否", "多少"]
return any(kw in question for kw in keywords)
def _fetch_realtime_data(self, question: str) -> list[dict]:
"""获取实时运维数据"""
docs = []
if "pod" in question.lower() or "k8s" in question.lower():
docs.append({"content": self.tools.kubectl("get pods -A | head -50"), "source": "kubectl"})
if "指标" in question or "metric" in question.lower():
docs.append({"content": self.tools.grafana_query("up"), "source": "grafana"})
return docs典型对话示例
用户: "api-gateway 现在健康吗？"
Agent:
[检索知识库] → 找到 api-gateway 的 SLO 定义、健康检查标准
[获取实时数据] → kubectl get pods -n production | grep api-gateway
grafana query: success_rate{service="api-gateway"}
回答:
"api-gateway 当前状态：⚠️ 部分降级
- Pod: 4/4 Running（正常）
- 成功率: 98.2%（SLO 99.9%，⚠️ 违反）
- P99 延迟: 850ms（SLO 300ms，⚠️ 违反）
- 错误预算: 7 天已消耗 15%
可能原因: 2 小前发布了 v2.3.1
建议: 参考 Runbook [[api-gateway-性能下降排查.md]]
历史案例: INC-2026-0512 有类似症状"与 kagent 的关联
你的 kagent 详解 笔记已经了解了 K8s Agent 框架。SRE Agent 可以基于 kagent 构建：
kagent 架构：
Agent → Task → Step → Tool
SRE Agent 映射：
Agent = SRE 诊断 Agent
Task = "诊断 api-gateway 502 故障"
Step 1 = 收集 Pod 状态 (Tool: kubectl)
Step 2 = 获取日志 (Tool: kubectl logs)
Step 3 = 查询指标 (Tool: grafana)
Step 4 = LLM 分析 (Tool: llm)
Step 5 = 输出诊断报告落地路线
| 阶段 | 场景 | 技术栈 | 难度 | 价值 |
|---|---|---|---|---|
| V1 | 告警智能分析（只读） | Alertmanager → LLM → 飞书推送 | 低 | 高（立竿见影） |
| V2 | 日志异常检测 | Loki → 聚类 → LLM 分析 | 中 | 中 |
| V3 | 故障诊断辅助 | Agent + 工具链 + RAG | 高 | 高 |
| V4 | 自动修复（受限） | Agent + ArgoCD 回滚 + kubectl | 很高 | 高 |
| V5 | 主动巡检 | 定时 Agent + 异常预测 | 很高 | 中 |
安全与权限
SRE Agent 权限模型：
只读 Agent（V1-V2）:
✅ kubectl get/describe/logs
✅ grafana 查询
✅ loki/elk 查询
✅ argocd app get/history
受限操作 Agent（V3-V4）:
⚠️ argocd app rollback（需审批）
⚠️ kubectl scale（需审批）
⚠️ kubectl drain（需审批）
禁止操作（所有阶段）:
❌ kubectl delete
❌ kubectl edit
❌ 直接修改 DB
❌ 删除备份常见问题 / 坑点
| 问题 | 原因 | 解决方案 |
|---|---|---|
| LLM 幻觉给出错误诊断 | LLM 不了解实时状态 | RAG + 实时数据注入，减少推理依赖 |
| Agent 工具调用超时 | kubectl/grafana 慢 | 加超时 + 降级到缓存数据 |
| 自动修复误操作 | 置信度不够就执行 | 低于 80% 置信度不自动执行，转人工 |
| 知识库过时 | 没有更新机制 | 复盘后自动更新知识库 |
| 安全风险 | Agent 权限过大 | 最小权限 + 操作审计 + 高危操作需审批 |
关联知识
- SRE 稳定性工程总览 — 本篇是 AI 赋能稳定性
- AIOps 实践与智能运维 — 互补：AIOps 侧重算法，本篇侧重 Agent
- LLM 辅助 RCA 与日志分析 — 更聚焦的 AI 运维场景
- kagent 详解 — K8s Agent 框架，可作为 SRE Agent 的基座
- 企业级多智能体设计实战 — 多 Agent 协作架构
- 14-MCP协议-标准化定义工具接口 — 工具协议
- 13-自定义工具封装-构建Tools的五步标准SOP — 工具封装
- 18-从Prompt到Harness-记忆与上下文的设计范式 — Agent 记忆
- 可观测性知识总览 — Agent 消费可观测性数据
- 根因定位方法论 — Agent 辅助 RCA 的方法论基础
- 故障生命周期管理 — Agent 在故障生命周期中的定位
参考资源
- kagent: https://github.com/kagent-dev/kagent
- LangChain Agent 文档
- Google AIOps 白皮书
状态
- SRE Agent 架构
- 告警智能分析场景
- 故障诊断辅助场景
- 日志智能分析场景
- 知识检索场景
- 权限与安全
- 落地路线
- 实际落地验证