AI 自动化运维与 Agent 编排
适用场景：利用 LLM Agent 自动化处理运维告警、生成故障报告、执行常规巡检，降低人工运维成本。
1. Agent 架构设计
1.1 单 Agent 模式
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 告警源 │ ──▶ │ Agent │ ──▶ │ 执行器 │
│ (Prometheus)│ │ (LLM + RAG) │ │ (API/CLI) │
└─────────────┘ └─────────────┘ └──────┬──────┘
│
▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 通知渠道 │ ◀── │ 日志记录 │ ◀── │ 结果反馈 │
│ (钉钉/邮件) │ │ (MySQL) │ │ │
└─────────────┘ └─────────────┘ └─────────────┘
1.2 多 Agent 协作模式
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 接收 Agent │ ──▶ │ 诊断 Agent │ ──▶ │ 执行 Agent │
│ (分流) │ │ (根因分析) │ │ (修复操作) │
└─────────────┘ └─────────────┘ └──────┬──────┘
│
▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 通知 Agent │ ◀── │ 验证 Agent │ ◀── │ 结果确认 │
│ (通知) │ │ (效果检查) │ │ │
└─────────────┘ └─────────────┘ └─────────────┘
2. Prometheus 告警集成
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 Agent 告警处理脚本
# agent_handler.py
import requests
from datetime import datetime
class AlertAgent:
def __init__(self, llm_url="http://localhost:11434", rag_db=None):
self.llm_url = llm_url
self.rag_db = rag_db
def process_alert(self, alert):
"""处理告警并生成回复"""
# 1. 检索相关知识
context = self.rag_db.search(alert.get("description", ""), top_k=3)
# 2. 构建 prompt
prompt = f"""运维告警处理请求：
告警名称：{alert.get('alertname')}
实例：{alert.get('instance')}
指标值：{alert.get('value')}
时间：{alert.get('starts_at')}
相关知识：
{context}
请：
1. 分析可能的根因
2. 提供排查步骤
3. 给出修复建议（如适用）
"""
# 3. 调用 LLM
response = requests.post(
f"{self.llm_url}/api/generate",
json={"model": "qwen2.5:7b", "prompt": prompt, "stream": False}
)
# 4. 记录日志
self.log_alert(alert, response.text)
return response.text
def log_alert(self, alert, response):
"""记录到数据库"""
# 实现数据库写入逻辑
pass
3. 故障诊断 Agent
3.1 诊断流程
# diagnostic_agent.py
from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain.tools import Tool
from langchain.chat_models import ChatOllama
class DiagnosticAgent:
def __init__(self):
self.llm = ChatOllama(model="qwen2.5:7b", temperature=0)
# 注册工具
self.tools = [
Tool(
name="check_service_status",
func=self.check_service,
description="检查服务状态"
),
Tool(
name="query_metrics",
func=self.query_metrics,
description="查询 Prometheus 指标"
),
Tool(
name="read_logs",
func=self.read_logs,
description="读取服务日志"
),
Tool(
name="execute_command",
func=self.execute_cmd,
description="执行诊断命令（只读）"
)
]
self.agent = create_openai_functions_agent(self.llm, self.tools)
self.executor = AgentExecutor(agent=self.agent, tools=self.tools, verbose=True)
def diagnose(self, problem_desc: str) -> str:
"""诊断问题"""
return self.executor.run(problem_desc)
def check_service(self, service_name: str) -> str:
"""检查服务状态"""
# 实现 systemctl/docker ps 等逻辑
pass
def query_metrics(self, metric_name: str, instance: str = None) -> str:
"""查询指标"""
# 实现 PromQL 查询
pass
def read_logs(self, service: str, lines: int = 100) -> str:
"""读取日志"""
# 实现 journalctl/docker logs
pass
def execute_cmd(self, cmd: str) -> str:
"""执行命令（安全沙箱）"""
# 白名单校验 + 超时控制
pass
3.2 典型故障诊断流程
用户输入："Nginx 返回 502 错误"
│
▼
Agent 调用 check_service("nginx")
│
├─ 服务状态：running ✓
│
▼
Agent 调用 query_metrics("nginx_upstream_response_time")
│
├─ 上游响应时间 > 30s ⚠️
│
▼
Agent 调用 read_logs("nginx", lines=50)
│
├─ 发现：connect() failed (111: Connection refused)
│
▼
Agent 结论：上游服务不可达，检查后端应用状态
│
▼
建议：检查 upstream 服务是否宕机或过载
4. 工单生成 Agent
4.1 工单结构化
# ticket_agent.py
import json
from datetime import datetime
class TicketAgent:
def __init__(self, llm_url="http://localhost:11434"):
self.llm_url = llm_url
def generate_ticket(self, alert_data: dict) -> dict:
"""生成结构化工单"""
prompt = f"""根据以下告警信息生成运维工单：
告警内容：{json.dumps(alert_data, ensure_ascii=False)}
请输出 JSON 格式：
{{
"title": "工单标题",
"severity": "P0/P1/P2/P3",
"category": "网络/主机/应用/数据库",
"description": "详细描述",
"suggested_actions": ["步骤1", "步骤2"],
"estimated_fix_time": "预计修复时间",
"related_knowledge": "相关文档链接"
}}
"""
response = requests.post(
f"{self.llm_url}/api/generate",
json={"model": "qwen2.5:7b", "prompt": prompt, "stream": False}
)
# 解析 JSON
try:
ticket = json.loads(response.json().get("response", "{}"))
ticket["created_at"] = datetime.now().isoformat()
ticket["status"] = "open"
return ticket
except:
return self._fallback_ticket(alert_data)
def _fallback_ticket(self, alert_data):
"""降级工单模板"""
return {
"title": f"告警：{alert_data.get('alertname')}",
"severity": "P2",
"category": "unknown",
"description": str(alert_data),
"suggested_actions": ["人工排查"],
"estimated_fix_time": "待定",
"created_at": datetime.now().isoformat(),
"status": "open"
}
5. 安全边界与审核
5.1 命令白名单
# safety.py
ALLOWED_COMMANDS = {
"check": [
"systemctl status {}",
"docker ps",
"docker logs --tail 100 {}",
"journalctl -u {} -n 100",
"curl -s http://localhost:{}/healthz",
"kubectl get pods",
"kubectl describe pod {}",
"kubectl logs --tail 100 {}",
],
"query": [
"curl -s 'http://prometheus:9090/api/v1/query?query={}'",
"mysql -e '{}'",
"redis-cli info",
]
}
DENIED_PATTERNS = [
"rm -rf",
"mkfs",
"dd if=",
"> /etc/",
"chmod 777",
"wget http",
"curl http",
"sudo rm",
]
def validate_command(cmd: str) -> bool:
"""验证命令安全性"""
# 检查白名单
for category, patterns in ALLOWED_COMMANDS.items():
for pattern in patterns:
if pattern.format("*") in cmd or cmd.startswith(pattern.split("{")[0]):
return True
# 检查危险模式
for pattern in DENIED_PATTERNS:
if pattern in cmd:
return False
return False
5.2 执行沙箱
import subprocess
import timeout_decorator
@timeout_decorator.timeout(30)
def safe_execute(cmd: str, timeout: int = 30) -> dict:
"""安全执行命令"""
if not validate_command(cmd):
return {"error": "Command not allowed"}
try:
result = subprocess.run(
cmd,
shell=True,
capture_output=True,
text=True,
timeout=timeout,
env={**os.environ, "HOME": "/tmp/sandbox"}
)
return {
"stdout": result.stdout[:10000], # 限制输出长度
"stderr": result.stderr[:10000],
"returncode": result.returncode
}
except subprocess.TimeoutExpired:
return {"error": "Command timed out"}
except Exception as e:
return {"error": str(e)}
6. 效果评估
6.1 评估指标
| 指标 | 计算方法 | 目标值 |
|---|---|---|
| 告警处理准确率 | 正确分类数 / 总数 | > 85% |
| MTTR 缩短比例 | (原MTTR - 现MTTR) / 原MTTR | > 30% |
| 人工干预频率 | 干预次数 / 总告警数 | < 20% |
| 用户满意度 | 调研评分 | > 4.0/5.0 |
6.2 评估脚本
# evaluation.py
import json
from datetime import datetime, timedelta
class AgentEvaluator:
def __init__(self, db_conn):
self.db = db_conn
def evaluate_weekly(self):
"""周报评估"""
start = datetime.now() - timedelta(days=7)
# 获取本周告警处理记录
alerts = self.db.query(
"SELECT * FROM alert_logs WHERE created_at > %s",
(start,)
)
metrics = {
"total_alerts": len(alerts),
"auto_resolved": sum(1 for a in alerts if a['resolution'] == 'auto'),
"manual_intervention": sum(1 for a in alerts if a['required_manual']),
"avg_response_time": self.calc_avg_response_time(alerts),
"false_positive_rate": self.calc_false_positive_rate(alerts)
}
return metrics
def calc_avg_response_time(self, alerts):
"""计算平均响应时间"""
times = [a['response_time'] for a in alerts if a.get('response_time')]
return sum(times) / len(times) if times else 0
def calc_false_positive_rate(self, alerts):
"""计算误报率"""
false_positives = sum(1 for a in alerts if a.get('is_false_positive'))
return false_positives / len(alerts) if alerts else 0
7. 常见问题排查
问题 1：Agent 循环调用工具
现象：Agent 陷入死循环，反复调用同一工具。
解决：
# 设置最大工具调用次数
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
max_iterations=5,
handle_parsing_errors=True
)
问题 2：LLM 输出格式不正确
现象：解析 JSON 失败。
解决：
# 使用 function calling 而非自由文本
from langchain.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant. Output JSON only."),
("user", "{input}")
])
问题 3：幻觉导致错误诊断
现象：Agent 给出看似合理但错误的建议。
解决： - 增加 RAG 上下文检索 - 使用 fewer-shot prompt 示例 - 对关键决策要求人工确认