Архитектурный паттерн «LangGraph, гибридный RAG + Сигнатурный движок»: универсальный граф для потоковых данных

В чем проблема?

При обработке больших потоков неструктурированного текста разработчики обычно выбирают один из двух путей:

• Первый — использовать детерминированные методы. Регулярки, сигнатуры, и подобные инструменты работают быстро. Вот только никак не поймают новые, ранее не описанные паттерны.

• Путь второй — скармливать сырые данные в ЛЛМ напрямую. В реальности этот подход быстро умирает. Отправлять огромные объемы нефильтрованного текста в языковые модели слишком дорого и медленно.

Но зачем выбирать что‑то одно? И да, я знаю поговорку про зайцев, но ведь и python — двустволка.

Мы попытались автоматизировать первую линию SOC. Захотелось объединить гибкость ЛЛМ и надежность сигнатурных движков. Поместилось все это в один асинхронный граф.

Статья об архитектуре, а не готовом решении.
Мы тестировали пайплайн на логах кибербезопасности, но концепция получилась модульной и универсальной. Логи лишь выступают как пример для демонстрации. YARA и Sigma правила реализованы на самописных движках и работают исключительно с текстом. Представленный ниже граф можно адаптировать под разбор отзывов, фильтрацию спама или модерацию внутренних документов компании.

Содержание:

2. Идея архитектуры

3. Устройство графа

4. AI-ядро

5. Детерминированная ветка

6. Метрики

7. Синтетический датасет

8. Выводы

1. Идея архитектуры

Архитектурную схему и логику связей я проектировал сам. Непосредственным развертыванием и внедрением Vector и Kafka занимался тимлид нашей команды, а вся AI‑часть, разработка логики графа и интеграция движков правил лежала на мне. Задача формулировалась в рамках проектного практикума УрФУ как автоматизация 1-й линии SOC с классификацией инцидентов по базе MITRE.

Поток логов

[Log Files]    
    ↓    
[Vector] (Lua-чанкер: 250 строк, overlap 20)    
    ↓
[Kafka] (topic: processed-logs-batches)    
    ↓
[Consumer] (aiokafka, retry 3x, exp backoff, DLQ)    
    ↓
[LangGraph DAG] (8 узлов)    
    ↓
[PostgreSQL] (Log + Report + Pending YARA Rules)

Лог‑файлы читаются Vector, нарезаются Lua‑скриптом в чанки по 250 строк с перекрытием 20 строк (чтобы не разрывать события на границе). Далее уходят в Kafka. Там их ловит consumer с retry‑логикой и Dead Letter Queue для необработанных сообщений. И далее файлы проходят через наш граф.

Теперь еще ближе к архитектуре.

2. Устройство Графа

Большинство решений в этой области — линейные цепочки. Мы пошли немного дальше:

workflow = StateGraph(AnalysisState)

workflow.add_node("prefilter", nodes.prefilter_node)
workflow.add_node("agent1", nodes.agent1_node)
workflow.add_node("description_agent", nodes.description_agent_node)
workflow.add_node("parse_logs", nodes.parse_logs_node)
workflow.add_node("agent2", nodes.agent2_node)
workflow.add_node("yara_scan", nodes.yara_scan_node)
workflow.add_node("sigma_scan", nodes.sigma_scan_node)
workflow.add_node("agent3", nodes.agent3_node)

# AI-ветка
workflow.add_edge("prefilter", "agent1")
workflow.add_edge("agent1", "agent3")
workflow.add_edge("agent1", "description_agent")
workflow.add_edge("description_agent", "agent2")
workflow.add_edge("agent2", "agent3")

# Детерминированная ветка (параллельно AI)
workflow.add_edge("parse_logs", "yara_scan")
workflow.add_edge("parse_logs", "sigma_scan")
workflow.add_edge("yara_scan", "agent3")
workflow.add_edge("sigma_scan", "agent3")
workflow.add_edge(START, "prefilter")
workflow.add_edge(START, "parse_logs")

Как это выглядит визуально:

Есть две параллельные ветки (да, те самые пути).

В первой все логи проходят фильтрацию в prefilter. Далее Agent 1 анализирует отфильтрованные логи, выявляет аномалии и составляет группы строк, для которых Description Agent генерирует описания событий. Agent 2 (RAG) ищет MITRE ATT&CK техники для каждой группы по её описанию.

Во второй логи структурируются в parse_logs. Далее YARA и Sigma сканирование.

Все четыре потока сливаются в Agent 3. Он делает финальную сводку: объединяет YARA, Sigma, RAG и анализ от Agent 1 в отчёт.

Мне хотелось уменьшить затраты на токены, и ускорить работу LLM, а еще показалось не лучшим решением оставлять LLM совсем голодной. Agent 1 изначально задумывался только как описано выше, но пока я смотрел на схему, в голову пришла интересная фича.

Интересная фича: Agent 1 дополнительно отправляет свой анализ напрямую в Agent 3 (ребро agent1 → agent3). И это не дублирование. Так получается zero‑cost бонус — Agent 1 уже прогнал все логи через LLM для группировки. Мы просто просим его на основе того же ввода предположить самостоятельно какие инциденты могут быть. Своеобразный тест на «магию LLM». На выходе мы получаем гипотезы об инцидентах условно бесплатно (без повторной отправки контекста и траты входных токенов). Agent 3 позже решит, что из этого подтверждено, а что нет.

3. AI‑ядро: Agent 1 → Agent 2 (RAG) → Agent 3

В этом разделе не будет промптов, потому что они слишком огромные. Их можно найти в репозитори в /prompts.

Agent 1 — группировка логов

LLM получает окно логов и выделяет из него события. cmd.exe execution на фоне [warning] => событие. Пачка Authentication failed с разных IP => брутфорс.

Кроме этого он пробует самостоятельно проанализировать лог‑файл целиком (фича выше).

RAG — гибридный поиск по MITRE ATT&CK

Недостаточно сказать «что‑то подозрительное было» (от бинарной классификации в таком деле мало толку). Важно знать, какая именно техника из MITRE ATT&CK замечена: T1110 (Brute Force), T1190 (Exploit Public‑Facing Application) и так далее.

В этом пайплайне RAG используется как динамический классификатор, подтягивающий контекст нужной техники MITRE для валидации аномалий (что‑то в стиле «Intent Classification через RAG» или «Few‑shot Classifier»).

Я сделал его асинхронным и с такими этапами:

1. Улучшение запроса LLM — модель переформулирует описание события в поисковый запрос на русском. Русский язык, потому что база техник MITRE в нашей ChromaDB была полностью переведена и переписана мной с сокращением описаний и добавлением ключевых слов.

2. Гибридный поиск — векторный + BM25 с весами 0.6/0.4.

3. LLM re‑rank — модель получает топ-3 техники и выбирает из них лучшую.

{  
  "technique_id": "T1687",  
  "technique_name": "Exploitation for Defense Impairment",  
  "description": "Зафиксирована эксплуатация уязвимостей в средствах защиты (антивирусы, EDR, файрволы) для их отключения или ослабления. Цель - нарушить работу систем обнаружения и реагирования на инциденты.",  
  "tactic": "defense-impairment",  
  "keywords_ru": [  
    "эксплуатация защиты",  
    "отключение EDR",  
    "уязвимость антивируса",  
    "обход защиты",  
    "ослабление защиты"  
  ]  
},

def hybrid_search(
    self,
    query: str,
    k: int = 10,
    vector_weight: float = 0.6,
    bm25_weight: float = 0.4,
    score_threshold: float = 0.3,
) -> list[dict]:
    if not self._vectorstore:
        raise RuntimeError("ChromaDB not initialized")

    if not self._bm25 or not self._technique_docs:
        return self.search(query, k=k, score_threshold=score_threshold)

    try:
        # 1. Получаем 2k результатов из каждого ранжировщика
        vec_results = self._vectorstore.similarity_search_with_score(query, k=k * 2)
        bm25_results = self._bm25.search(query, k=k * 2)

        if not vec_results and not bm25_results:
            return []

        # 2. Нормализация: делим на максимальный score
        vec_max = max((1.0 - s) for _, s in vec_results) if vec_results else 1.0
        bm25_max = max(s for _, s in bm25_results) if bm25_results else 1.0

        combined: dict[str, float] = {}

        for doc, score in vec_results:
            tid = doc.metadata.get("technique_id", "")
            if not tid:
                continue
            norm_sim = (1.0 - score) / vec_max if vec_max > 0 else 0
            combined[tid] = combined.get(tid, 0) + norm_sim * vector_weight  # * 0.6

        for idx, bm25_score in bm25_results:
            if idx < len(self._technique_docs):
                tid = self._technique_docs[idx]["technique_id"]
                norm_bm25 = bm25_score / bm25_max if bm25_max > 0 else 0
                combined[tid] = combined.get(tid, 0) + norm_bm25 * bm25_weight  # * 0.4

        # 3. Сортировка и фильтр по threshold
        results = []
        for tid, score in sorted(combined.items(), key=lambda x: x[1], reverse=True)[:k]:
            if score < score_threshold:
                continue
            for doc in self._technique_docs:
                if doc["technique_id"] == tid:
                    results.append({
                        "content": f"passage: {doc['text']}",
                        "metadata": { ... },
                        "score": score,
                    })
                    break
        return results

    except Exception as e:
        return self.search(query, k=k, score_threshold=score_threshold)

Гибридный поиск помогает повысить качество совпадений: векторный хорошо находит семантические совпадения, но иногда промахивается с точными техническими идентификаторами, специфическими аббревиатурами или названиями утилит. BM25 же работает только с ними.

LLM re‑rank нужен, потому что техники очень близки по описаниям. Часто первая в выдаче не является верной, хотя и похожа. LLM имеет контекст и отклоняет совпадение, если оно ложно.

Agent 3 — сведение веток, валидация и генерация отчёта

Три основных канала (RAG, YARA, Sigma) работают независимо и всегда попадают в финальный отчёт. События Agent 1, совпавшие с любой из трёх веток, идут в финальный инцидент, при этом не дублируются. Те, что не совпали — в блок «Требуют ручной проверки» с мета‑полем unconfirmed_events_count, без влияния на severity.

LLM может найти атаку, которой нет ни в MITRE, ни в YARA/Sigma). Игнорировать такие находки показалось странным. Пусть лучше аналитик видит полный список и решает сам. Это как перекрестная валидация + «человек в контуре».

Для подтверждённых событий Agent 3 формирует отчет со всеми найденными аномалиями и сохраняет его в базу данных. Из нее отчеты попадают в чат с пользователем на сайте.

4. YARA и Sigma — детерминированная ветка

Изначально добавили 8 YARA и 9 Sigma правил как минимальную базу. В целом, правила можно создавать, редактировать и удалять через веб‑интерфейс. Система не привязана к фиксированному набору.

YARA правила

В нашем проекте анализируются только текстовые логи, потому YARA работает с текстом, а не по прямому назначению. Она полноценная, но натянутая на текст. В общем виде — обёртка над yara‑python.
Используется нативный компилятор yara.compile() и rules.match(data=...) — быстрый C‑шный движок, требующий .yar файлы со своим синтаксисом.

import yara

class YaraEngine:
    def __init__(self, rules_path):
        filepaths = {f.stem: str(f) for f in rules_path.glob("*.yar")}
        self._rules = yara.compile(filepaths=filepaths)

    def scan(self, parsed_logs: list[dict]) -> list[dict]:
        results = []
        for log in parsed_logs:
            scan_data = " ".join(filter(None, [
                log.get("message", ""), log.get("uri", ""),
                log.get("user_agent", ""), log.get("referer", "")
            ]))
            matches = self._rules.match(data=scan_data.encode("utf-8"))
            for m in matches:
                results.append({
                    "rule": m.rule,
                    "severity": m.meta.get("severity", "unknown"),
                    "mitre_ref": m.meta.get("mitre_ref", ""),
                })
        return results

Sigma правила

С ними чуть занятнее. Это просто rule‑base движок, со стилизованным под Sigma синтаксисом. YAML парсится в список условий, затем сравнивается через contains, startswith, endswith, re. Чуть медленнее, чем YARA но гибче под структурированные логи.

import yaml

class SigmaEngine:
    def __init__(self, rules_path):
        self._rules = []
        for yml_file in rules_path.glob("*.yml"):
            with open(yml_file) as f:
                data = yaml.safe_load(f)
            selections = {}
            for key, val in data.get("detection", {}).items():
                if isinstance(val, dict):
                    selections[key] = self._parse_selection(val)
            self._rules.append({
                "title": data.get("title"),
                "level": data.get("level", "medium"),
                "selections": selections,
                "condition": data["detection"].get("condition", ""),
            })

    def scan(self, parsed_logs):
        results = []
        for rule in self._rules:
            for log in parsed_logs:
                text = " ".join(filter(None, [log.get(k, "") for k in
                    ("message", "uri", "user_agent", "referer")]))
                matched = []
                for sel_name, conditions in rule["selections"].items():
                    if all(self._check_cond(f, m, v, text)
                           for f, m, v in conditions):
                        matched.append(sel_name)
                if matched:
                    results.append({
                        "title": rule["title"],
                        "severity": rule["level"],
                        "matched_logs": matched,
                    })
        return results

    def _parse_selection(self, sel: dict) -> list[tuple]:
        parsed = []
        for field, value in sel.items():
            modifier = "equals"
            if "|" in field:
                field, modifier = field.split("|", 1)
            for v in (value if isinstance(value, list) else [value]):
                parsed.append((field.lower(), modifier.lower(), str(v)))
        return parsed

Автогенерация YARA‑правил

Когда RAG находит технику без покрытия в YARA, Agent 3 запускает автогенератор:
LLM пишет правило по описанию техники и строкам логов, после чего проходит валидацию, в том числе LLM‑ревью. При ошибке генератор исправляет правило по обратной связи, до 3 попыток. Готовое правило сохраняется в PendingYaraRules, появляется в веб‑интерфейсе, где аналитик утверждает или отклоняет его.

Получается доработка статичных правил в процессе работы, причем снова с «human in the loop».

Движки и автогенерация правил проверены: через веб‑интерфейс загружались относительно реалистичные логи из logs_examples/.

5. Что с метриками?

Метрики (RAG классификации)

На тестовом датасете test2/ — поток 43 минуты синтетических логов, около 850 строк, 38 техник MITRE в ground truth:

• Precision: 85.7% (30 TP / 35 всего предсказаний)

• Recall: 78.9% (30 TP / 38 всего ожидаемых)

• F1: 0.82

Перекашивает в T1190

5 из 8 false negatives — систематический перекос в T1190 «Exploit Public‑Facing Application». Образовалась RAG‑категория «свалка»: всё, что похоже на веб‑атаку, но не подходит под другие техники, вернее кажется, что не подходит по логам) уходит туда. Происходит так, потому что границы между некоторыми техниками размыты.

Более контрастные описания в базе знаний (некоторые техники уже объединены, так как очень похожи семантически сами по себе) могут убрать этот перекос или снизить его.

Дублирование инцидентов

Была задумка, что одно и то же окно логов может быть обработано RAG в рамках разных групп событий (это отдельно прописано в промпте). Но такая инструкция иногда дает на выходе два очень похожих инцидент‑блока. Это проблема отсутствия дедупликации на уровне Agent 3. Часть из 5 false positives приходится на такие дубли — то есть реальное качество RAG немного выше, чем показывают метрики.

Исправить можно пост‑процессингом: сгруппировать одинаковые техники перед финальным отчётом

Производительность

3.5 строки/сек.
Последовательные LLM‑вызовы внутри графа очень сильно замедляют анализ. Agent 1 обрабатывает контекст, примерно 70k токенов, потом RAG re‑rank для каждой группы, пусть и асинхронно — ещё запросы, причем новые YARA правила при тестировании генерировались, и это тоже время. Кроме того нужно еще пробиться к облачному Gemini 2.5 Flash, который заботливо подключил наш тестировщик, так как наши ноутбуки не вывозили нужные нам модели.

У нас нет локального GPU. Решение спроектировано для on‑premise, но тестировать приходится на облачных LLM. «Уроборос» какой‑то… не иначе. Так что на данный момент не удалось замерить реальную скорость на локальном железе.

Возможно батчинг и локальная модель смогут поднять пропускную способность до 10–20 строк/сек.

Для многих задач вне анализа логов часть из этих проблем неактуальна. Данные для анализа могут быть более семантически отдаленными, что повысит качество при правильных описаниях и облегчит их написание. Скорость возрастет при развертывании на локальном железе и уменьшении объема событий для анализа.

6. Синтетический датасет

Тестовые данные генерировались потоком в изолированном контейнере mitre_log_simulator/ (костыль «на любителя») на основе фреймворка Atomic Red Team. Реализована чистая текстовая симуляция, то есть парсинг техник и генерация похожих строк. Все это без выполнения команд в системе. Покрыто 88+ техник матрицы. Атаки маркированы [warning] / [error] и содержат прямые подсказки: cmd.exe execution, multiple failed login attempts.

[Fri Jun 05 16:42:28 2026] [warning] [client 10.0.84.159] cmd.exe execution: cmd.exe /c "..."

Для LLM нет проблемы понять, что это инцидент. Вот только атакующий не станет так выдавать себя: cmd.exe execution, а будет маскироваться под обычный трафик. Вполне вероятно, что в более реалистичном датасете Recall будет ниже.

RAG корректно находит MITRE‑техники даже на нашем упрощенном датасете, следовательно, фундамент работает. А точность будет расти с улучшением данных.

Основной целью была проверка пайплайна именно на потоковых данных, и такие фокусы с синтетикой помогли убедиться в его работоспособности.

7. Выводы

Идея работает, но дорого в текущих условиях, в частности для кибербезопасности.

Уточнение для SOC‑специалистов. В классическом SOC первая линия работает с сигнатурами: SIEM‑правила формируют карточки инцидентов. Мы решали другую задачу. Это не замена SOC‑процессам, а скорее другой слой детекта — для сценариев, где сигнатур ещё нет или их написание нецелесообразно.

Переиспользование

Автоматизация первой линии SOC — наглядный пример. Замена YARA на движок модерации контента, RAG — на базу знаний компании, позволит тому же графу анализировать комментарии или документы.

Паттерн LangGraph + hybrid RAG + deterministic engine на мой взгляд универсален.

Весь код открыт в моем форке на GitHub. В репозитории есть тесты в pipeline_tests/, docker‑compose на 5 сервисов, CI/CD (Docker Hub + зеркало для GitVerse), метрики для RAG. Для быстрого старта рекомендую брать готовый протестированный релиз v1.0.1 (инструкция в README).