Как создать AI-агент для обнаружения и обработки аномалий во временных рядах: пошаговый гайд

Зачем вашему мониторингу нужен не просто скрипт, а думающий агент

Вы ставите Z-score или изолированный лес на поток данных. Получаете тысячу алертов в день. 90% из них — ложные срабатывания. Инженеры перестают реагировать, пропускают реальный пожар. Знакомая история? Проблема не в методах, а в подходе. Статические правила и одна модель не справляются с динамичным миром.

AI-агент — это не очередной скрипт с if-else. Это система, которая сама решает: какой метод применить сейчас, как интерпретировать выброс, нужно ли его корректировать или бить тревогу. Она учится на ваших правках и адаптируется. В этой статье мы соберем такого агента с нуля, используя архитектурные паттерны из наших предыдущих материалов и актуальные на 2026 год инструменты.

1 Собираем фундамент: данные и простые методы

Прежде чем запускать тяжелую артиллерию, нужно понять базовые сигналы. Начнем с классики, которая работает в 80% случаев.

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# Генерация синтетических данных временного ряда с аномалиями (актуальный синтаксис 2026)
def generate_ts_data(start_date='2025-01-01', periods=1000, freq='H'):
    np.random.seed(42)
    dates = pd.date_range(start=start_date, periods=periods, freq=freq)
    
    # Базовый тренд + сезонность
    trend = np.linspace(0, 10, periods)
    seasonal = 5 * np.sin(2 * np.pi * np.arange(periods) / 24)
    noise = np.random.normal(0, 0.5, periods)
    values = trend + seasonal + noise
    
    # Внедряем аномалии
    anomaly_indices = np.random.choice(periods, size=15, replace=False)
    values[anomaly_indices] += np.random.uniform(8, 15, 15)  # Резкие всплески
    
    return pd.DataFrame({'timestamp': dates, 'value': values})

df = generate_ts_data()
print(df.head())

Теперь реализуем двух классических "детективов": Z-score для нормальных распределений и IQR для всего остального. В 2026 году их все еще используют, но с умом.

def detect_zscore_anomalies(series, window=30, threshold=3):
    """Скользящий Z-score. Не используйте глобальное среднее!"""
    mean = series.rolling(window=window, center=True).mean()
    std = series.rolling(window=window, center=True).std()
    z_scores = np.abs((series - mean) / std)
    return z_scores > threshold

def detect_iqr_anomalies(series, window=50):
    """Скользящий IQR. Устойчив к выбросам в самом окне."""
    q1 = series.rolling(window, center=True).quantile(0.25)
    q3 = series.rolling(window, center=True).quantile(0.75)
    iqr = q3 - q1
    lower_bound = q1 - 1.5 * iqr
    upper_bound = q3 + 1.5 * iqr
    return (series < lower_bound) | (series > upper_bound)

# Применяем
df['zscore_anomaly'] = detect_zscore_anomalies(df['value'])
df['iqr_anomaly'] = detect_iqr_anomalies(df['value'])

2 Архитектура агента: планировщик, исполнитель и специалисты

Вот где начинается магия. Мы не будем пихать всю логику в один промпт к LLM. Вспомним архитектуру из статьи «Как спроектировать современного AI-агента». Наш агент будет состоять из:

• Планировщик (Planner): Решает, какой метод или комбинацию методов запустить. Анализирует свойства ряда (стационарность, дисперсию, наличие тренда).
• Исполнитель (Executor): Запускает выбранные детекторы — статистические или ML-модели.
• Суб-агенты (Sub-agents): Узкие специалисты. Один отвечает за статистические методы, другой — за нейросетевые, третий — за пост-обработку. Принципы их использования мы разбирали в материале про суб-агентов в AI-разработке.

# Каркас основных классов агента (упрощенно)
from abc import ABC, abstractmethod
from typing import List, Dict, Any
import pandas as pd

class DetectorSubAgent(ABC):
    """Базовый класс для детектора-суб-агента."""
    def __init__(self, config: Dict):
        self.config = config
    
    @abstractmethod
    def fit(self, series: pd.Series):
        pass
    
    @abstractmethod
    def detect(self, series: pd.Series) -> pd.Series:
        """Возвращает булевый ряд с аномалиями."""
        pass

class StatisticalDetector(DetectorSubAgent):
    """Суб-агент, владеющий статистическими методами (Z-score, IQR, ESD)."""
    def __init__(self, config):
        super().__init__(config)
        self.method = config.get('method', 'iqr')
        
    def detect(self, series):
        if self.method == 'iqr':
            return detect_iqr_anomalies(series, **self.config.get('params', {}))
        elif self.method == 'zscore':
            return detect_zscore_anomalies(series, **self.config.get('params', {}))
        # ... другие методы

class MLDetector(DetectorSubAgent):
    """Суб-агент для ML-моделей (Isolation Forest, Autoencoder)."""
    def __init__(self, config):
        super().__init__(config)
        # Загрузка актуальных на 2026 версий библиотек
        # from sklearn.ensemble import IsolationForest (версия 2.0+)
        # или tf.keras для автоэнкодеров
        pass
    # ... реализация

3 Мозг операции: планировщик на основе LLM

Планировщик — это LLM (например, GPT-5, Claude 4 или открытая модель типа Llama 4 400B), которой мы даем анализ ряда и список доступных инструментов. Его задача — не считать, а думать.

# Пример промпта для планировщика (актуальные модели на 2026)
PLANNER_SYSTEM_PROMPT = """
Ты — планировщик системы обнаружения аномалий. Тебе на вход приходит временной ряд (его статистики) и история.
Твоя задача: выбрать оптимальный метод обнаружения аномалий из доступных инструментов.

Доступные инструменты (детекторы):
1. statistical_iqr - Хорош для данных без нормального распределения, устойчив к локальным выбросам.
2. statistical_zscore - Для стационарных данных, похожих на нормальное распределение.
3. ml_isolation_forest - Для многомерных аномалий и сложных паттернов.
4. ml_autoencoder - Для обнаружения novelty, когда мало примеров аномалий.

Анализируй ряд. Обрати внимание на:
- Стационарность (тест Дики-Фулера).
- Распределение (скошенность, эксцесс).
- Наличие тренда/сезонности.
- Исторические метки аномалий (если есть).

Верни JSON: {"detector": "имя_детектора", "reasoning": "обоснование", "params": {}}.
"""

# На практике это вызов к API LLM с контекстом
# async def plan_detection(series_stats: Dict, history: List) -> Dict:

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

4 Исполнитель и пост-обработка: что делать с найденными выбросами

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

• Игнорировать: Если это известный артефакт (например, деплой в 3 ночи).
• Скорректировать: Заменить на интерполированное значение для чистоты дальнейшего анализа.
• Эскалировать: Отправить алерт в Slack/Telegram/PagerDuty.

class AnomalyPostProcessor:
    """Принимает решение по каждой обнаруженной аномалии."""
    def __init__(self, knowledge_base):
        self.kb = knowledge_base  # База знаний об известных событиях
        
    def decide(self, anomaly_timestamp, anomaly_value, context):
        # Проверяем, не попадает ли timestamp на плановое техобслуживание
        if self.kb.is_known_event(anomaly_timestamp):
            return {'action': 'ignore', 'reason': 'known_event'}
        
        # Если аномалия слишком резкая и значения за пределами физических лимитов
        if anomaly_value > context['physical_limit']:
            return {'action': 'escalate', 'severity': 'critical', 'channel': 'pager'}
        
        # В остальных случаях — корректируем и логируем
        return {'action': 'correct', 'method': 'linear_interpolation'}

# Интеграция в исполнителя
executor_results = ml_detector.detect(series)
anomaly_indices = series[executor_results].index

for idx in anomaly_indices:
    decision = post_processor.decide(idx, series.loc[idx], context_dict)
    execute_decision(decision)  # Отправляет алерт или корректирует ряд

5 Сборка пайплайна и оценка в бою

Теперь нужно связать все компоненты в надежный пайплайн, который не упадет от пропущенного значения и не забьет вашу очередь сообщений мусором.

from typing import Optional
import asyncio

class AnomalyDetectionAgent:
    """Главный агент, orchestrating весь процесс."""
    def __init__(self, planner_llm, detectors: Dict, post_processor):
        self.planner = planner_llm
        self.detectors = detectors
        self.post_processor = post_processor
        self.memory = []  # Stateful memory для хранения истории решений
        
    async def run(self, series: pd.Series, context: Optional[Dict] = None) -> Dict:
        """Основной цикл: Plan -> Detect -> Post-process -> Learn."""
        # 1. Анализ ряда и планирование
        series_stats = self._compute_stats(series)
        plan = await self.planner.generate_plan(series_stats, self.memory)
        
        # 2. Исполнение: запуск выбранного детектора
        detector = self.detectors.get(plan['detector'])
        if not detector:
            # Fallback на самый надежный метод
            detector = self.detectors['statistical_iqr']
        raw_anomalies = detector.detect(series)
        
        # 3. Пост-обработка и действия
        decisions = []
        for idx in series[raw_anomalies].index:
            decision = self.post_processor.decide(idx, series.loc[idx], context)
            decisions.append(decision)
            self._execute_action(decision)  # Неблокирующий вызов
        
        # 4. Обновление памяти для будущего обучения
        self.memory.append({
            'timestamp': pd.Timestamp.now(),
            'plan': plan,
            'anomalies_count': raw_anomalies.sum(),
            'decisions': decisions
        })
        
        return {'plan': plan, 'anomalies_found': raw_anomalies.sum(), 'decisions': decisions}
    
    def _execute_action(self, decision):
        # Интеграция с внешними системами: Slack, PagerDuty, база данных
        if decision['action'] == 'escalate':
            send_alert(decision)  # Ваша функция отправки
        elif decision['action'] == 'correct':
            correct_data(decision)  # Запись корректировки в хранилище

Оценка такого агента — отдельная задача. Нельзя мерить только точностью и recall. Нужно считать стоимость ошибок: ложное срабатывание — это время инженера, пропущенная аномалия — это простой или убыток. Метрики должны быть бизнес-ориентированными. Подробнее об этом мы писали в статье про систему оценки AI-агентов.

Чего не делать: ошибки, которые сломают вашего агента

• Использовать глобальные пороги для всех метрик. Лимит в 1000 RPS для одного сервиса — норма, для другого — аномалия. Агент должен знать контекст каждой метрики.
• Забывать про сезонность. Детектировать аномалии в трафике без учета времени суток и дня недели — гарантированные ложные срабатывания ночью и пропуск атак днем.
• Доверять LLM без валидации. Планировщик может выбрать экзотический метод, для которого нет данных. Исполнитель должен иметь fallback-стратегию.
• Хранить состояние в памяти процесса. При перезапуске агент потеряет всю историю. Используйте внешнее хранилище (Redis, база данных), как в архитектуре production-ready агента.
• Игнорировать операционные расходы. Запуск тяжелой модели автоэнкодера на каждом тике для 10 000 метрик разорит вас. Планировщик должен учитывать стоимость вычислений.

Вопросы, которые вы хотели задать, но стеснялись

Главный секрет не в сложности моделей, а в архитектуре, которая позволяет им работать вместе. Начните с простого агента, который выбирает между Z-score и IQR на основе дисперсии ряда. Добавьте память, чтобы он запоминал свои ошибки. Затем подключите первый ML-детектор. Итеративно. Так вы построите систему, которая действительно снижает операционную нагрузку, а не становится еще одной источником головной боли.