AI DevOps-инженер — цифровой сотрудник
AI DevOps-инженер автоматизирует операционные задачи: диагностику инцидентов, анализ логов, оптимизацию инфраструктуры, генерацию IaC-кода (Terraform, Ansible), настройку CI/CD пайплайнов. Действует как first-responder при инцидентах и снижает нагрузку on-call инженеров на L1-задачи.
Incident Response агент
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from typing import TypedDict, Annotated, Optional
import operator
llm = ChatOpenAI(model="gpt-4o", temperature=0)
class IncidentState(TypedDict):
alert_data: dict
investigation_steps: Annotated[list, operator.add]
root_cause: Optional[str]
severity: Optional[str]
actions_taken: Annotated[list, operator.add]
resolved: bool
escalation_required: bool
@tool
def get_recent_logs(service: str, minutes: int = 30, level: str = "ERROR") -> str:
"""Получить последние логи сервиса из Loki/Elasticsearch.
Args:
service: Имя сервиса
minutes: Период в минутах
level: Уровень логов (ERROR, WARN, INFO)
"""
logs = loki_client.query(
query=f'{{app="{service}"}} |= "{level}"',
start=f"-{minutes}m",
limit=100,
)
return "\n".join(logs[:50])
@tool
def get_metrics(service: str, metric_names: list[str], minutes: int = 60) -> str:
"""Получить метрики сервиса из Prometheus."""
metrics = {}
for metric in metric_names:
result = prometheus.query_range(
query=f'{metric}{{service="{service}"}}',
start=f"-{minutes}m",
step="1m",
)
metrics[metric] = result
return json.dumps(metrics)
@tool
def check_kubernetes_pods(namespace: str, label_selector: str = "") -> str:
"""Проверить состояние Pod'ов в Kubernetes."""
pods = k8s_client.list_pods(namespace=namespace, label_selector=label_selector)
pod_status = [{
"name": p.metadata.name,
"phase": p.status.phase,
"ready": all(c.ready for c in (p.status.container_statuses or [])),
"restarts": sum(c.restart_count for c in (p.status.container_statuses or [])),
"age_minutes": (datetime.now() - p.metadata.creation_timestamp).seconds // 60,
} for p in pods.items]
return json.dumps(pod_status)
@tool
def restart_deployment(namespace: str, deployment_name: str) -> str:
"""Перезапустить деплоймент в Kubernetes (rollout restart)."""
k8s_apps.patch_namespaced_deployment(
name=deployment_name,
namespace=namespace,
body={"spec": {"template": {"metadata": {"annotations": {
"kubectl.kubernetes.io/restartedAt": datetime.now().isoformat()
}}}}},
)
return f"Деплоймент {deployment_name} перезапускается"
@tool
def scale_deployment(namespace: str, deployment_name: str, replicas: int) -> str:
"""Масштабировать деплоймент."""
if replicas > 20:
return "Ошибка: превышен лимит масштабирования (20 реплик)"
k8s_apps.patch_namespaced_deployment_scale(
name=deployment_name,
namespace=namespace,
body={"spec": {"replicas": replicas}},
)
return f"Деплоймент {deployment_name} масштабирован до {replicas} реплик"
# Агент реагирования на инцидент
incident_tools = [get_recent_logs, get_metrics, check_kubernetes_pods, restart_deployment, scale_deployment]
INCIDENT_RESPONSE_PROMPT = """Ты — Senior SRE/DevOps Engineer. Расследуй инцидент автономно.
При расследовании:
1. Сначала собери данные (логи, метрики, состояние pod'ов)
2. Определи root cause
3. Попробуй устранить автоматически, если это безопасно (restart, scale up)
4. Если требуется ручное вмешательство — эскалируй с подробным контекстом
Никогда не делай автоматически:
- Изменения в production базах данных
- Откат деплоймента без явного указания
- Масштабирование до > 10 реплик
- Удаление ресурсов"""
from langgraph.prebuilt import create_react_agent
incident_agent = create_react_agent(
llm.bind_tools(incident_tools),
tools=incident_tools,
state_modifier=INCIDENT_RESPONSE_PROMPT,
)
Log Analysis агент
class LogAnalyzer:
async def analyze_error_pattern(
self,
service: str,
time_range: str = "1h",
) -> dict:
"""Анализирует паттерны ошибок в логах"""
# Получаем и кластеризуем ошибки
error_logs = await loki_client.query_errors(service, time_range)
clustered = self.cluster_errors(error_logs)
# LLM анализирует паттерны
analysis = await llm.ainvoke(f"""Проанализируй паттерны ошибок:
Топ ошибок (кластеры):
{json.dumps(clustered[:10], ensure_ascii=False, indent=2)}
Временной паттерн: {self.get_time_pattern(error_logs)}
Определи:
1. Root cause наиболее частых ошибок
2. Аномальные паттерны (внезапный рост, цикличность)
3. Рекомендации по устранению""")
return {
"clusters": clustered,
"analysis": analysis.content,
"anomalies": self.detect_anomalies(error_logs),
}
def cluster_errors(self, logs: list[dict]) -> list[dict]:
"""Простая кластеризация по fingerprint ошибки"""
from collections import Counter
fingerprints = Counter()
examples = {}
for log in logs:
# Нормализуем ошибку (убираем динамические части)
fingerprint = re.sub(r'\b\d+\b', 'N', log.get("message", ""))
fingerprint = re.sub(r'[0-9a-f]{8}-[0-9a-f-]{23}', 'UUID', fingerprint)
fingerprints[fingerprint] += 1
if fingerprint not in examples:
examples[fingerprint] = log["message"]
return [
{"fingerprint": fp[:100], "count": count, "example": examples[fp]}
for fp, count in fingerprints.most_common(20)
]
IaC Generator
class InfrastructureCodeGenerator:
async def generate_terraform(
self,
infrastructure_description: str,
cloud_provider: str = "aws",
existing_modules: list[str] = None,
) -> str:
"""Генерирует Terraform конфигурацию"""
modules_context = f"\nДоступные модули: {existing_modules}" if existing_modules else ""
response = await llm.ainvoke(f"""Сгенерируй Terraform конфигурацию для:
{infrastructure_description}
Провайдер: {cloud_provider}
Требования:
- Используй последние stable версии провайдеров
- Следуй best practices: не хардкодь credentials, используй variables и outputs
- Добавь теги для cost allocation
- Включи базовые security groups / IAM policies
{modules_context}
Верни полный HCL код с комментариями.""")
return response.content
async def generate_ansible_playbook(
self,
task_description: str,
target_os: str = "ubuntu",
idempotency_required: bool = True,
) -> str:
"""Генерирует Ansible playbook"""
response = await llm.ainvoke(f"""Сгенерируй Ansible playbook для:
{task_description}
Целевая ОС: {target_os}
Идемпотентность: {'обязательна — все tasks должны быть идемпотентны' if idempotency_required else 'желательна'}
Требования:
- Используй ansible-lint best practices
- Handler'ы для сервисов
- Проверь before/after если применимо
- Verifiable — добавь verify tasks
Верни YAML playbook.""")
return response.content
CI/CD Pipeline Generator
async def generate_github_actions_pipeline(
project_type: str, # "python-fastapi", "node-react", "go"
deployment_target: str, # "kubernetes", "lambda", "ecs"
requirements: list[str], # ["tests", "security-scan", "docker", "terraform"]
) -> str:
response = await llm.ainvoke(f"""Сгенерируй GitHub Actions workflow для:
Тип проекта: {project_type}
Деплой: {deployment_target}
Требования: {requirements}
Включи:
- Параллельные задачи где возможно
- Кэширование зависимостей
- Правильные условия (push main → deploy prod, PR → tests only)
- Environment protection rules для production
- Notify on failure
Верни полный YAML workflow.""")
return response.content
Практический кейс: стартап, 2 DevOps на 15 разработчиков
Ситуация: 2 DevOps инженера, 40+ микросервисов, ночные дежурства изматывали команду. L1 инциденты (OOMKilled, высокая нагрузка, медленные запросы) занимали 60% on-call времени.
AI DevOps First-Responder:
- Обрабатывает PagerDuty алерты автономно
- Собирает диагностические данные (логи, метрики, k8s состояние)
- Выполняет безопасные автоматические действия (restart, scale up)
- Для сложных случаев: будит инженера с полным контекстом вместо raw алерта
Результаты:
- L1 инциденты закрытые автономно: 61%
- Среднее время разбудить инженера ночью: снизилось на 58%
- Mean Time to Recovery (MTTR): 45 мин → 18 мин
- DevOps фокус: архитектура, оптимизация, не рутинные рестарты
- Ночные побудки: -63%
IaC-генерация: 180 PR с Terraform/Ansible кодом за 3 месяца, 91% приняты без существенных правок.
Сроки
- Incident Response агент с инструментами K8s: 2–3 недели
- Log Analysis система: 1–2 недели
- IaC Generator для основных ресурсов: 1–2 недели
- CI/CD Generator + интеграции: 1–2 недели
- Интеграция с PagerDuty/OpsGenie: 1 неделя
- Итого: 6–10 недель