Агент для разбора логов и поиска причин сбоев

Разбор логов — задача, которую человек делает медленно и без удовольствия: открыть journalctl, прокрутить сотни строк, найти момент, когда всё пошло не так, сопоставить с другими сервисами, вспомнить, что менялось в конфигурации накануне. Именно здесь LLM особенно полезна — не потому что она умнее человека, а потому что ей не лень прочитать тысячу строк лога и удержать их все в контексте одновременно, находя закономерности, которые человек пропустил бы из-за банальной усталости.

В этой статье соберём агента, который специализируется именно на разборе логов: получает описание проблемы или временной промежуток, собирает логи с нужных серверов, фильтрует и группирует их, и выдаёт структурированный анализ вероятной причины сбоя. Возьмём практический пример: на srv-web01 в 03:14 ночи участились ошибки 502 Bad Gateway, и нужно понять, что произошло — на самом сервере или на связанном с ним srv-db01, в сети 192.168.10.0/24.

Прежде чем переходить к реализации, стоит понять, чем такой агент принципиально отличается от агента автоматического реагирования из более ранней статьи серии. Там агент действует по сигналу мониторинга и быстро выбирает из ограниченного набора известных исправлений. Здесь задача другая — не быстрое действие, а вдумчивый анализ произвольного, часто неструктурированного объёма логов, когда причина сбоя заранее не известна и не укладывается в шаблон. Это разные роли агента, и совмещать их в одном промпте обычно плохая идея: диагностический анализ требует больше контекста и времени на размышление, чем экспресс-реакция на типовой сбой.

Шаг 1. Сбор логов с нескольких источников в единый набор

Проблема редко проявляется в одном единственном логе. 502 на веб-сервере может быть следствием как проблем с самим nginx, так и с приложением за ним, так и с базой данных, к которой это приложение обращается. Собираем логи со всех потенциально релевантных источников за нужный промежуток времени.

def collect_logs_for_incident(hosts_services, start_time, end_time):
    collected = {}
    for host, service in hosts_services:
        since = start_time.strftime("%Y-%m-%d %H:%M:%S")
        until = end_time.strftime("%Y-%m-%d %H:%M:%S")
        cmd = f'journalctl -u {service} --since "{since}" --until "{until}" --no-pager'
        result = ssh_run(host, cmd)
        collected[f"{host}/{service}"] = result.get("stdout", "")
    return collected

incident_sources = [
    ("srv-web01", "nginx"),
    ("srv-web01", "app"),
    ("srv-db01", "postgresql"),
]

Список источников для конкретного инцидента можно как задавать вручную, так и выводить автоматически из топологии сервисов — если у тебя уже есть карта зависимостей (какое приложение на каком сервере обращается к какой базе), агент может сам определить, какие логи стоит подтянуть, отталкиваясь от того, какой сервис сообщил о проблеме.

Шаг 2. Фильтрация шума перед тем, как отдавать логи модели

Сырые логи забиты информационными строками, которые не несут диагностической ценности — ротация, health-check запросы, стандартные сообщения о старте воркеров. Прежде чем считать токены, отфильтруй явный шум, оставив строки с уровнями warning и выше, а также строки непосредственно вокруг них для контекста.

import re

NOISE_PATTERNS = [
    r"GET /healthz",
    r"worker process \d+ started",
    r"reloading configuration",
]

def filter_noise(log_text, context_lines=2):
    lines = log_text.splitlines()
    keep = set()
    for i, line in enumerate(lines):
        is_noise = any(re.search(p, line) for p in NOISE_PATTERNS)
        is_relevant = re.search(r"\b(error|warn|critical|fatal|timeout|refused)\b", line, re.I)
        if is_relevant and not is_noise:
            for j in range(max(0, i - context_lines), min(len(lines), i + context_lines + 1)):
                keep.add(j)
    return "\n".join(lines[i] for i in sorted(keep))

Такая фильтрация обычно сокращает объём текста в 5-10 раз без потери значимой информации, и именно она делает возможным анализ логов за несколько часов активности, а не только за последние пару минут, без риска упереться в лимит контекста модели или получить неоправданно высокий счёт за токены.

Шаг 3. Чанкинг для действительно больших объёмов

Если инцидент разворачивался не за минуты, а за часы, и логов набирается на десятки мегабайт даже после фильтрации, единого запроса к модели может быть недостаточно. Разбей логи на чанки по времени и сначала попроси модель кратко резюмировать каждый чанк, а затем — свести резюме в общий анализ.

def chunk_by_time(filtered_log, chunk_minutes=15):
    chunks = defaultdict(list)
    for line in filtered_log.splitlines():
        ts = extract_timestamp(line)
        if ts:
            bucket = ts.replace(minute=(ts.minute // chunk_minutes) * chunk_minutes, second=0)
            chunks[bucket].append(line)
    return chunks

def summarize_chunks(chunks):
    summaries = []
    for bucket, lines in sorted(chunks.items()):
        summary = ask_llm_summarize("\n".join(lines), context=f"период {bucket}")
        summaries.append({"time": str(bucket), "summary": summary})
    return summaries

Двухэтапный подход — сначала локальные резюме, потом общий анализ поверх них — типичный паттерн map-reduce для работы с большими текстами через LLM, и он одинаково хорошо работает что для разбора логов, что для анализа больших документов в других задачах.

Шаг 4. Промпт для root cause analysis с явной структурой ответа

Финальный запрос к модели должен объединить резюме по чанкам (или отфильтрованный лог целиком, если он небольшой) и попросить не просто «что случилось», а структурированный разбор с явным указанием на неопределённость, если причина не очевидна.

RCA_SYSTEM_PROMPT = (
    "Ты анализируешь логи инцидента на инфраструктуре itpages.lab.local. "
    "На основе предоставленных логов с нескольких серверов определи наиболее вероятную "
    "цепочку событий, приведшую к сбою. Ответь в формате JSON: "
    "{\"timeline\": [список ключевых событий с временем], "
    "\"root_cause\": \"наиболее вероятная причина\", "
    "\"confidence\": число от 0 до 1, "
    "\"affected_services\": [список затронутых сервисов], "
    "\"recommended_fix\": \"конкретное действие или null\"}. "
    "Если данных недостаточно для уверенного вывода, честно укажи это в confidence."
)

def analyze_incident(logs_by_source):
    combined = "\n\n".join(f"=== {src} ===\n{text}" for src, text in logs_by_source.items())
    response = llm_client.messages.create(
        model="claude-sonnet-4-5",
        system=RCA_SYSTEM_PROMPT,
        messages=[{"role": "user", "content": combined}],
        max_tokens=800,
    )
    return json.loads(response.content[0].text)

Явное требование указывать низкую уверенность при недостатке данных — важная деталь промпта: без неё модель склонна выдавать уверенно звучащий, но не всегда обоснованный вывод просто потому, что от неё ждут ответа. Честный «confidence: 0.3, возможно связано с исчерпанием соединений к базе, но логов недостаточно для точного вывода» куда полезнее уверенного, но ошибочного диагноза.

Шаг 5. Сопоставление с историей похожих инцидентов

Если ты уже какое-то время ведёт журнал инцидентов (тот самый, из статьи про аудит), полезно давать модели не только текущие логи, но и краткие сводки похожих прошлых случаев — это резко повышает точность диагноза для повторяющихся проблем.

def find_similar_past_incidents(current_summary, incident_history, top_k=3):
    scored = []
    for past in incident_history:
        similarity = simple_keyword_overlap(current_summary, past["root_cause"])
        scored.append((similarity, past))
    scored.sort(key=lambda x: x[0], reverse=True)
    return [p for _, p in scored[:top_k]]

Для начала достаточно простого сравнения по ключевым словам, как в примере выше — полноценный семантический поиск по векторной базе данных, если он уже используется где-то в инфраструктуре, даст более точные совпадения, но и упрощённый вариант заметно помогает, особенно для проблем, которые повторяются раз в несколько недель и о которых уже все забыли к моменту следующего срабатывания.

Шаг 6. Отчёт для человека, а не только структурированный JSON

Структурированный JSON удобен для автоматической обработки, но человеку, который открывает результат утром после ночного инцидента, нужен читаемый текст. Финальным шагом стоит превратить JSON-ответ модели в короткий отчёт, который можно сразу отправить в чат команды или приложить к тикету.

def format_incident_report(analysis, incident_id):
    lines = [f"Разбор инцидента {incident_id}", ""]
    lines.append(f"Вероятная причина: {analysis['root_cause']} "
                  f"(уверенность {analysis['confidence']:.0%})")
    lines.append(f"Затронутые сервисы: {', '.join(analysis['affected_services'])}")
    lines.append("")
    lines.append("Хронология:")
    for event in analysis["timeline"]:
        lines.append(f"  {event['time']} — {event['description']}")
    if analysis.get("recommended_fix"):
        lines.append("")
        lines.append(f"Рекомендуемое действие: {analysis['recommended_fix']}")
    return "\n".join(lines)

Такой отчёт занимает пять секунд на чтение вместо получаса ручного разбора логов, и даже если диагноз модели окажется неточным, хронология событий, собранная из логов нескольких серверов в единую последовательность, сама по себе экономит человеку значительную часть работы — обычно самое трудоёмкое в разборе инцидента это не поставить диагноз, а просто сопоставить события по времени между разными источниками.

Шаг 7. Ограничение доступа агента разбора логов к записи

Агент из этой статьи только читает — он не выполняет исправляющих команд, только собирает и анализирует логи. Тем не менее стоит явно ограничить его SSH-пользователя правами на чтение логов и базовую диагностику, не давая даже теоретической возможности что-либо изменить на сервере, даже если баг в коде агента попытается вызвать не тот SSH-клиент. Разделение ролей «агент, который смотрит» и «агент, который чинит» — с разными пользователями и разными ключами на целевых серверах — снижает поверхность атаки даже в рамках одной инфраструктуры.

# на srv-web01 и srv-db01: отдельная группа для read-only агентов
groupadd ai-readonly
useradd -m -s /usr/sbin/nologin -g ai-readonly ai-log-reader
usermod -aG systemd-journal ai-log-reader  # доступ к journalctl без sudo

Группа systemd-journal в большинстве дистрибутивов даёт доступ на чтение журналов systemd без необходимости выдавать пользователю права sudo вообще — этого достаточно для всего, что делает агент разбора логов, и избыточных прав не остаётся.

Частые ошибки и как их избежать

• Передача модели сырых логов без фильтрации. Помимо лишних расходов на токены, шум в логах затрудняет модели поиск действительно значимых событий — отфильтрованный лог почти всегда даёт более точный анализ, чем полный.
• Игнорирование логов смежных сервисов. Причина сбоя на веб-сервере часто лежит в базе данных или в сети, а не в самом веб-сервисе — анализ только одного источника логов регулярно приводит к неверному диагнозу.
• Требование категоричного ответа при недостатке данных. Если промпт не даёт модели возможности честно сказать «недостаточно данных для уверенного вывода», она будет выдавать правдоподобно звучащие, но не всегда верные диагнозы, что хуже явного признания неопределённости.

Автор — Команда IT-Pages