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 тижнів