title: AI-Powered Operations Automation — Production Guide description: AI-powered ops automation: alert triage, incident response, log analysis, capacity forecasting. Original tutorial with 5 production automation patterns. tags: [ai, ai-ops, automation, llm, ops] lang: en date: 2026-10-23 original: true faq: - q: "How to automate alert triage with LLMs?" a: "Feed alert metadata (metrics, logs, service name) to an LLM with a diagnosis prompt template. Use function calling to trigger remediation actions. Always require human confirmation for P0/P1 incidents." - q: "What are the risks of AI-driven auto-remediation?" a: "Main risks: hallucinated commands, incorrect root cause, unintended side effects. Mitigate with: command sandboxing, dry-run mode, rollback procedures, and human-in-the-loop for destructive operations." - q: "How to handle LLM failures in the automation pipeline?" a: "Implement circuit breakers: if LLM response quality drops below threshold, fall back to rule-based automation. Cache successful remediation patterns for reuse." - q: "How to measure AI ops automation ROI?" a: "Track: MTTR reduction, false positive rate decrease, manual intervention frequency, cost savings from prevented outages. Target: 30% MTTR reduction within 3 months." - q: "Can AI replace SRE on-call?" a: "No. AI augments SREs by handling routine alerts and providing diagnostics. Human oversight is essential for complex incidents, decision-making, and accountability. Think 'AI copilot', not 'AI replacement'."
AI-Powered Operations Automation — Production Guide
Use Case: Automate repetitive ops tasks using LLMs — alert triage, incident response, log analysis, capacity planning. Reduce MTTR and free up SRE time for high-value work.
1. Why AI Ops Automation Matters
Traditional automation (scripts, runbooks) handles known patterns. AI automation handles the unknown:
| Capability | Script-Based | AI-Powered |
|---|---|---|
| Pattern matching | Exact match only | Semantic understanding |
| New alert types | Requires manual update | Learns from context |
| Root cause analysis | Rule-based heuristics | Contextual reasoning |
| Response generation | Fixed playbook | Dynamic adaptation |
Key Insight: AI doesn't replace scripts — it handles the cases scripts can't.
2. Alert Triage Automation
2.1 Architecture
┌──────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────┐
│ Prometheus│ ──▶ │ Alert Manager│ ──▶ │ AI Triage │ ──▶ │ Action │
│ Alerts │ │ │ │ Agent │ │ Router │
└──────────┘ └──────────────┘ └──────┬───────┘ └──────────┘
│
┌──────────────┼──────────────┐
▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│ Auto │ │ Escalate │ │ Human │
│ Resolve │ │ to On-Call│ │ Review │
└─────────┘ └─────────┘ └─────────┘
2.2 Alert Triage Implementation
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
TRIAGE_PROMPT = PromptTemplate(
input_variables=["alert_name", "metrics", "recent_changes"],
template="""
You are an SRE triage engineer. Analyze this alert and determine the severity and action.
Alert: {alert_name}
Metrics: {metrics}
Recent Changes: {recent_changes}
Return JSON:
{{
"severity": "P0/P1/P2/P3",
"category": "infrastructure/network/database/application",
"action": "auto_resolve/escalate/review",
"reasoning": "Brief explanation",
"suggested_steps": ["Step 1", "Step 2"]
}}
"""
)
2.3 Webhook Handler
from fastapi import FastAPI, Request
import httpx
app = FastAPI()
@app.post("/alerts/webhook")
async def handle_alert(request: Request):
alert = await request.json()
# Enrich with context
enriched = await enrich_alert(alert)
# Triage with LLM
triage_result = await triage_alert(enriched)
# Route to appropriate handler
if triage_result["action"] == "auto_resolve":
await auto_resolve(enriched, triage_result)
elif triage_result["action"] == "escalate":
await escalate(enriched, triage_result)
else:
await create_ticket(enriched, triage_result)
return {"status": "processed"}
3. Incident Response Automation
3.1 Incident Classification
INCIDENT_TEMPLATES = {
"high_cpu": {
"patterns": ["cpu_usage > 90%", "load_average high"],
"playbook": ["identify_process", "check_threads", "kill_or_restart"]
},
"disk_full": {
"patterns": ["disk_usage > 85%", "no_space_left"],
"playbook": ["find_large_files", "clean_logs", "expand_volume"]
},
"service_down": {
"patterns": ["http_5xx > 50%", "health_check_failed"],
"playbook": ["check_process", "restart_service", "verify"]
}
}
def classify_incident(alert_data: dict) -> str:
"""Classify incident type using pattern matching + LLM fallback"""
for incident_type, config in INCIDENT_TEMPLATES.items():
for pattern in config["patterns"]:
if pattern in alert_data.get("description", ""):
return incident_type
# Fallback to LLM classification
return llm_classify(alert_data)
3.2 Automated Remediation
import subprocess
from typing import List
REMEDIATION_ACTIONS = {
"high_cpu": [
lambda: subprocess.run(["top", "-bn1"], capture_output=True, text=True),
lambda: subprocess.run(["ps", "aux", "--sort=-%cpu"], capture_output=True, text=True),
lambda: subprocess.run(["systemctl", "restart", "affected-service"], capture_output=True, text=True)
],
"disk_full": [
lambda: subprocess.run(["du", "-sh", "/var/log"], capture_output=True, text=True),
lambda: subprocess.run(["find", "/var/log", "-name", "*.log", "-mtime", "+7", "-delete"], capture_output=True, text=True),
lambda: subprocess.run(["df", "-h"], capture_output=True, text=True)
]
}
async def execute_remediation(incident_type: str, dry_run: bool = True) -> dict:
"""Execute remediation with dry-run support"""
actions = REMEDIATION_ACTIONS.get(incident_type, [])
results = []
for action in actions:
if dry_run:
results.append({"action": str(action), "dry_run": True, "status": "simulated"})
else:
try:
output = action()
results.append({"action": str(action), "status": "executed", "output": output.stdout})
except Exception as e:
results.append({"action": str(action), "status": "failed", "error": str(e)})
return {"incident_type": incident_type, "results": results}
4. Log Analysis Automation
4.1 Anomaly Detection
from sklearn.ensemble import IsolationForest
import numpy as np
class LogAnomalyDetector:
def __init__(self):
self.model = IsolationForest(contamination=0.05, random_state=42)
self.baseline = None
def fit_baseline(self, historical_logs: List[str]):
"""Learn normal log patterns"""
features = self._extract_features(historical_logs)
self.model.fit(features)
self.baseline = features.mean(axis=0)
def detect_anomaly(self, new_log: str) -> dict:
"""Detect anomalies in new log entries"""
features = self._extract_features([new_log])
prediction = self.model.predict(features)
score = self.model.score_samples(features)[0]
return {
"is_anomaly": prediction == -1,
"anomaly_score": float(score),
"deviation": np.linalg.norm(features[0] - self.baseline)
}
def _extract_features(self, logs: List[str]) -> np.ndarray:
"""Extract numerical features from logs"""
# Simple feature extraction: error rate, request rate, response time
features = []
for log in logs:
features.append([
log.count("ERROR"),
log.count("WARN"),
len(log.split()),
self._extract_latency(log)
])
return np.array(features)
4.2 Log Summarization
SUMMARIZATION_PROMPT = """
Summarize these log entries, focusing on errors and anomalies:
{logs}
Provide:
1. Summary of key events
2. Error patterns identified
3. Recommended actions
"""
5. Capacity Forecasting
5.1 Trend Analysis
from statsmodels.tsa.holtwinters import ExponentialSmoothing
def forecast_capacity(time_series: list, periods: int = 30) -> dict:
"""Forecast resource usage using exponential smoothing"""
model = ExponentialSmoothing(
time_series,
trend='add',
seasonal='add',
seasonal_periods=24
)
fitted = model.fit()
forecast = fitted.forecast(periods)
# Calculate days until threshold
threshold = 0.85 # 85% capacity
days_to_threshold = None
for i, value in enumerate(forecast):
if value > threshold:
days_to_threshold = i
break
return {
"forecast": forecast.tolist(),
"days_to_threshold": days_to_threshold,
"current_trend": "increasing" if forecast[-1] > time_series[-1] else "stable"
}
5.2 Automated Scaling Recommendations
def generate_scaling_recommendation(metrics: dict) -> dict:
"""Generate scaling recommendation based on forecasts"""
cpu_forecast = forecast_capacity(metrics["cpu_history"])
mem_forecast = forecast_capacity(metrics["memory_history"])
disk_forecast = forecast_capacity(metrics["disk_history"])
recommendations = []
if cpu_forecast["days_to_threshold"]:
recommendations.append({
"resource": "cpu",
"action": "scale_up",
"urgency": "high" if cpu_forecast["days_to_threshold"] < 7 else "medium",
"reason": f"CPU expected to hit 85% in {cpu_forecast['days_to_threshold']} days"
})
if mem_forecast["days_to_threshold"]:
recommendations.append({
"resource": "memory",
"action": "optimize",
"urgency": "medium",
"reason": f"Memory expected to hit 85% in {mem_forecast['days_to_threshold']} days"
})
return {"recommendations": recommendations}
6. Production Patterns
6.1 Pattern 1: Alert-to-Ticket Automation
async def alert_to_ticket(alert: dict) -> str:
"""Convert alert to ITIL ticket"""
ticket_data = {
"title": f"[{alert['severity']}] {alert['alertname']}",
"description": await llm_summarize(alert),
"severity": alert['severity'],
"category": "infrastructure",
"sla_hours": {"P0": 1, "P1": 4, "P2": 24}.get(alert['severity'], 24)
}
return await create_jira_ticket(ticket_data)
6.2 Pattern 2: Post-Incident Report Generation
POST_INCIDENT_PROMPT = """
Generate a post-incident report for the following incident:
Incident Summary: {summary}
Timeline: {timeline}
Root Cause: {root_cause}
Actions Taken: {actions}
Include:
1. Executive summary
2. Impact assessment
3. Timeline of events
4. Root cause analysis
5. Lessons learned
6. Action items with owners and due dates
"""
6.3 Pattern 3: Runbook Generation
RUNBOOK_PROMPT = """
Generate an ops runbook for handling: {incident_type}
Include:
1. Detection criteria
2. Immediate response steps
3. Escalation path
4. Resolution verification
5. Post-incident actions
"""
7. Safety Guards
7.1 Circuit Breaker Pattern
from circuitbreaker import circuit
@circuit(
failure_threshold=5,
recovery_timeout=60,
expected_exception=Exception
)
async def ai_remediate(incident: dict):
"""AI remediation with circuit breaker"""
return await execute_remediation(incident)
# Usage
try:
result = await ai_remediate(incident)
except Exception:
# Fall back to manual process
await notify_oncall(f"AI remediation failed for {incident['type']}, manual intervention required")
7.2 Human-in-the-Loop Requirements
HIGH_RISK_OPERATIONS = [
"systemctl restart",
"docker rm -f",
"kubectl delete",
"rm -rf",
"iptables -F",
"drop database"
]
def requires_human_approval(operation: str) -> bool:
"""Check if operation requires human approval"""
return any(risky in operation.lower() for risky in HIGH_RISK_OPERATIONS)