Сочетание статистического обнаружения с агентным принятием решений.
Делиться
Введение
Как специалист по анализу данных, работающий над прогнозированием временных рядов, я сталкивался с аномалиями и выбросами бесчисленное количество раз. В данных по прогнозированию спроса, финансам, трафику и продажам я постоянно натыкаюсь на всплески и спады, которые трудно интерпретировать.
Обработка аномалий обычно представляет собой серую зону, редко бывает однозначной, но указывает на более глубокие проблемы. Некоторые аномалии являются реальными сигналами, такими как праздники, погодные явления, рекламные акции или вирусные тренды; другие — просто сбои в данных, но на первый взгляд они выглядят одинаково. Чем быстрее мы обнаружим аномалии в данных, тем быстрее сможем принять меры для предотвращения снижения производительности и ущерба.
Мы имеем дело с критически важными данными временных рядов, и обнаружение аномалий имеет решающее значение. Если удалить истинное событие, будет удалена ценная точка данных сигнала, а если сохранить ложный сигнал тревоги, обучающие данные будут содержать шум.
Большинство детекторов на основе машинного обучения выявляют всплески активности, используя Z-баллы, пороговые значения межквартильного размаха или другие статические методы без учета контекста. Благодаря недавним достижениям в области искусственного интеллекта у нас появился лучший вариант для разработки агента обработки аномалий, который анализирует каждый случай. Агент, который обнаруживает необычное поведение, проверяет контекст и решает, следует ли исправить данные, оставить их как реальный сигнал или пометить для проверки.
В этой статье мы шаг за шагом создаём такого агента, который сочетает в себе простое статистическое обнаружение с агентом искусственного интеллекта, выступающим в качестве первой линии защиты данных временных рядов, сокращая ручное вмешательство и сохраняя при этом наиболее важные сигналы. Мы будем обнаруживать и обрабатывать аномалии в данных по COVID-19 с помощью автономного принятия решений, основанного на степени серьёзности аномалии, используя:
- Оперативные эпидемиологические данные из API disease.sh.
- Статистическое обнаружение аномалий.
- Классификация степени тяжести.
- Агент искусственного интеллекта на базе GroqCloud, который принимает автономные решения о том, следует ли:
- Устранить аномалию
- Сохраните аномалию
- Отметить аномалию для проверки человеком
Это агентная аналитика принятия решений, а не просто обнаружение аномалий.
Почему одного лишь традиционного метода обнаружения аномалий недостаточно?
Существуют традиционные методы машинного обучения, такие как изолированные леса, предназначенные для обнаружения аномалий, но им не хватает сквозной координации принятия решений. Они не способны достаточно быстро реагировать на них в производственных средах. Мы внедряем агента ИИ, чтобы восполнить этот пробел, преобразуя исходные оценки аномалий в автономные, сквозные решения, динамически принимаемые на основе данных в реальном времени.
Традиционное обнаружение аномалий
Традиционный метод обнаружения аномалий использует конвейерный подход, представленный ниже:
Ограничения традиционного метода обнаружения аномалий
- Работает со статическими правилами и позволяет вручную устанавливать пороговые значения.
- Она одномерна и обрабатывает простые данные.
- Отсутствует контекстное обоснование.
- Принятие решений, определяемое человеком.
- Действие, управляемое вручную.
Обнаружение и устранение аномалий с помощью агента искусственного интеллекта.
Обнаружение аномалий агентом ИИ осуществляется в соответствии с конвейерным подходом, представленным ниже:
Почему на практике это работает лучше?
- Работает с данными в реальном времени.
- Она многомерна и способна обрабатывать сложные данные.
- Работает на основе контекстного мышления.
- Адаптивное и самообучающееся принятие решений.
- Предпринимайте самостоятельные действия.
Выбор реалистичного набора данных для нашего примера.
Мы используем реальные данные о COVID-19 для выявления аномалий, поскольку они содержат много шума, показывают всплески, а полученные результаты помогают улучшить общественное здравоохранение.
Что мы хотим, чтобы решил ИИ-агент?
Цель состоит в том, чтобы непрерывно отслеживать данные по COVID-19, выявлять аномалии, определять степень их серьезности, а также принимать автономные решения и определять необходимые действия:
- Отметить аномалию для проверки человеком
- Устранить аномалию
- Сохраните аномалию
Источник данных
Для получения данных мы используем бесплатные данные disease.sh в режиме реального времени через API. Этот API предоставляет данные о ежедневном количестве подтвержденных случаев, смертей и выздоровлений. При реализации ИИ-агента мы сосредоточимся на ежедневном количестве случаев, которое идеально подходит для обнаружения аномалий.
Лицензия на данные: В этом руководстве используются исторические данные о количестве случаев COVID-19, полученные через API disease.sh. Базовый набор данных (репозиторий данных по COVID-19 факультета компьютерных наук Университета Джонса Хопкинса) распространяется по лицензии CC BY 4.0, которая разрешает коммерческое использование с указанием авторства. (Дата обращения: 22 января 2026 г.)
Как все эти части складываются воедино?
Архитектура системы обнаружения аномалий в данных по COVID-19 с использованием агента искусственного интеллекта выглядит следующим образом:
Пошаговое создание ИИ-агента
Давайте шаг за шагом разберемся, как загрузить данные с помощью disease.sh, обнаружить аномалии, классифицировать их и реализовать агента ИИ, который будет рассуждать и предпринимать соответствующие действия в зависимости от серьезности аномалий.
Шаг 1: Установите необходимые библиотеки.
Первым шагом является установка необходимых библиотек, таких как phidata, groq, python-dotenv, tabulate и streamlit.
pip install phidata pip install groq pip install python-dotenv #библиотека для загрузки файла .env pip install tabulate pip install streamlit
Шаг 2: Настройка файла среды
Откройте свою IDE и создайте папку проекта, а внутри этой папки создайте файл среды «.env» для хранения GROQ_API_KEY.
GROQ_API_KEY=»ваш_ключ_groq_api_здесь»
Шаг 3: Загрузка данных
Прежде чем создавать какого-либо агента, нам нужен источник данных, достаточно зашумленный, чтобы выявлять реальные аномалии, но достаточно структурированный для анализа. Ежедневные данные о количестве случаев COVID-19 хорошо подходят, поскольку они содержат задержки в отчетности, внезапные всплески и изменения режима. Для простоты мы намеренно ограничиваемся одним одномерным временным рядом.
Загрузите данные из скрипта disease.sh, используя URL-адрес запроса, и извлеките дату и количество случаев заболевания за день на основе выбранной страны и количества дней, за которые вы хотите получить данные. Данные преобразуются в структурированный датафрейм путем анализа JSON, форматирования даты и хронологической сортировки.
# ————————————— # ЗАГРУЗКА ДАННЫХ (disease.sh) # ————————————— def load_live_covid_data(country: str , days:int): url = f»https://disease.sh/v3/covid-19/historical/{country}?lastdays={days}» response = requests.get(url) data = response.json()[«timeline»][«cases»] df = ( pd.DataFrame(list(data.items()), columns=[«Date», «Cases»]) .assign(Date=lambda d: pd.to_datetime(d[«Date»], format=»%m/%d/%y»)) .sort_values(«Date») .reset_index(drop=True) ) return df
Шаг 4: Выявление аномалий
Теперь мы будем выявлять аномальное поведение в данных временных рядов COVID-19, обнаруживая внезапные всплески и тенденции быстрого роста. Количество случаев в целом стабильно, а большие отклонения или резкие увеличения указывают на значимые аномалии. Теперь мы будем выявлять аномалии с помощью статистических методов и бинарной маркировки для детерминированного и воспроизводимого обнаружения аномалий. Для обнаружения аномалий рассчитываются два параметра.
- Обнаружение всплесков
- Внезапный всплеск данных выявляется с помощью Z-показателя; если какая-либо точка данных выходит за пределы диапазона Z-показателя, это должно быть аномалией.
- Выявление темпов роста
- Рассчитывается темп роста в сравнении с предыдущим днем; если он превышает 40%, это отмечается как подозрительное событие.
# ————————————— # ОБНАРУЖЕНИЕ АНОМАЛИЙ # ————————————— def detect_anomalies(df): values = df[«Cases»].values mean, std = values.mean(), values.std() spike_idx = [ i for i, v in enumerate(values) if abs(v — mean) > 3 * std ] growth = np.diff(values) / np.maximum(values[:-1], 1) growth_idx = [i + 1 for i, g in enumerate(growth) if g > 0.4] anomalies = set(spike_idx + growth_idx) df[«Anomaly»] = [«YES» if i in anomalies else «NO» for i in range(len(df))] return df
Если обнаружена аномалия по одному из параметров — пику или росту, или по обоим параметрам, — значение параметра «Аномалия» устанавливается в «ДА»; в противном случае — в «НЕТ».
Шаг 5: Классификация степени тяжести
Не все аномалии одинаковы; мы будем классифицировать их как « КРИТИЧНЫЕ », « ПРЕДУПРЕЖДЕНИЕ » или « НЕЗНАЧИТЕЛЬНЫЕ », чтобы направлять решения ИИ-агента. Для классификации серьезности используются фиксированные скользящие окна и пороговые значения на основе правил. Серьезность классифицируется только при наличии аномалии; в противном случае параметры «Серьезность», «Решение агента» и «Действие» в датафрейме устанавливаются в значение «пусто».
# ————————————— # КОНФИГУРАЦИЯ # ————————————— ROLLING_WINDOW = 7 MIN_ABS_INCREASE = 500 # ————————————— # КЛАССИФИКАЦИЯ СТЕПЕНИ ТЯЖЕСТИ # ————————————— def compute_severity(df): df = df.sort_values(«Date»).reset_index(drop=True) df[«Severity»] = «» df[«Agent Decision»] = «» df[«Action»] = «» for i in range(len(df)): if df.loc[i, «Anomaly»] == «YES»: if i < ROLLING_WINDOW: df.loc[i, "Severity"] = "" curr = df.loc[i, "Cases"] baseline = df.loc[i - ROLLING_WINDOW:i- 1, "Cases"].mean() abs_inc = curr - baseline growth = abs_inc / max(baseline, 1) if abs_inc < MIN_ABS_INCREASE: df.loc[i, "Severity"] = "" if growth >= 1.0: df.loc[i, «Severity»] = «CRITICAL» elif growth >= 0.4: df.loc[i, «Severity»] = «WARNING» else: df.loc[i, «Severity»] = «MINOR» return df
В приведенном выше коде для классификации степени серьезности аномалии каждая аномалия сравнивается с 7-дневными историческими данными ( ROLLING_WINDOW = 7), и рассчитываются абсолютный и относительный рост.
- Абсолютный рост
Параметр конфигурации MIN_ABS_INCREASE = 500 определяет изменения ниже этого значения как очень незначительные, пренебрежимо малые. Если абсолютный рост меньше MIN_ABS_INCREASE, то его следует игнорировать, а параметр серьезности оставить пустым. Абсолютный рост позволяет обнаружить значимое влияние в реальных условиях, не реагирует на шум или незначительные колебания и предотвращает ложные срабатывания при высоком проценте роста.
- Относительный рост:
Относительный рост помогает выявлять взрывные тенденции. Если рост превышает или равен 100% по сравнению с базовым уровнем, это означает внезапный всплеск и обозначается как « КРИТИЧНЫЙ »; если рост превышает 40%, это означает устойчивое ускорение и требует мониторинга, и обозначается как « ПРЕДУПРЕЖДЕНИЕ »; в противном случае обозначается как « НЕЗНАЧИТЕЛЬНЫЙ ».
После классификации степени тяжести, система готова к автономному принятию решений и действиям со стороны ИИ-агента.
Шаг 6: Создайте запрос для агента ИИ.
Ниже приведено описание того, как ИИ-агент рассуждает и принимает решения на основе структурированного контекста и предопределенной степени серьезности при обнаружении аномалии. Агент ограничен тремя явно выраженными действиями и должен возвращать один детерминированный ответ для безопасной автоматизации.
def build_agent_prompt(obs): return f»»» Вы — агент мониторинга данных о COVID-19 на основе ИИ. Наблюдаемая аномалия: Дата: {obs['date']} Случаи: {obs['cases']} Тяжесть: {obs['severity']} Правила принятия решений: — FIX_ANOMALY: шум, колебания в отчетах — KEEP_ANOMALY: реальный сигнал вспышки — FLAG_FOR_REVIEW: серьезная или неоднозначная аномалия Ответьте ТОЛЬКО одним из следующих способов: FIX_ANOMALY KEEP_ANOMALY FLAG_FOR_REVIEW «»»
В запрос явно указываются три параметра: дата, количество зарегистрированных случаев и степень тяжести, что помогает агенту ИИ принимать решения автономно.
Шаг 7: Создайте своего агента в GroqCloud.
В настоящее время мы создаём автономного агента ИИ с использованием GroqCloud, который принимает интеллектуальные контекстные решения на основе обнаруженных аномалий и степени их серьёзности, а также предпринимает соответствующие действия. Три предопределённых действия для агента ИИ обеспечивают использование только проверенных результатов.
# ————————————— # СОЗДАНИЕ АГЕНТА ИИ # ————————————— agent = Agent( name=»CovidAnomalyAgent», model=Groq(id=»openai/gpt-oss-120b»), instructions=»»» Вы — агент ИИ, отслеживающий данные временных рядов COVID-19 в режиме реального времени. Выявите аномалии и примите решение в соответствии с аномалией: «FIX_ANOMALY», «KEEP_ANOMALY», «FLAG_FOR_REVIEW».»»» ) for i in range(len(df)): if df.loc[i, «Anomaly»] == «YES»: obs = build_observation(df, i) prompt = build_agent_prompt(obs) response = agent.run(prompt) decision = response.messages[-1].content.strip() decision = decision if decision in VALID_ACTIONS else «FLAG_FOR_REVIEW» df = agent_action(df, i, decision)
Создан ИИ-агент под названием « CovidAnomalyAgent », который использует модель LLM, размещенную на платформе GroqCloud, для быстрого и низкозадержечного анализа. ИИ-агент выполняет четко определенную подсказку, анализирует данные, проводит контекстный анализ, принимает автономное решение и предпринимает действия в рамках безопасных ограничений.
Искусственный интеллект не занимается обработкой аномалий, а принимает интеллектуальные решения для каждой обнаруженной аномалии. Решение агента точно отражает серьезность аномалии и необходимые действия.
# ————————————— # Решатель действий агента # ————————————— def agent_action(df, idx, action): df.loc[idx, «Решение агента»] = action if action == «FIX_ANOMALY»: fix_anomaly(df, idx) elif action == «KEEP_ANOMALY»: df.loc[idx, «Действие»] = «Принято как реальный сигнал вспышки» elif action == «FLAG_FOR_REVIEW»: df.loc[idx, «Действие»] = «Отмечено для проверки человеком» return df
ИИ-агент игнорирует нормальные точки данных без аномалий и рассматривает только точки данных со значением « АНОМАЛИЯ = ДА ». ИИ-агент ограничен тремя допустимыми решениями: « ИСПРАВИТЬ_АНОМАЛИЯ », « СОХРАНИТЬ_АНОМАЛИЯ » и « ФЛАГ ДЛЯ ПРОВЕРКИ », и, соответственно, предпринимаются действия, определенные в таблице ниже:
| Решение агента | Действие |
| ИСПРАВЛЕНИЕ АНОМАЛИИ | Автоматическая коррекция, выполненная агентом искусственного интеллекта. |
| KEEP_ANOMALY | Принято за реальный сигнал о вспышке заболевания. |
| FLAG_FOR_REVIEW | Помечено для проверки человеком. |
В случае незначительных аномалий агент ИИ автоматически исправляет данные, сохраняет допустимые аномалии в неизменном виде и помечает критические случаи для проверки человеком.
Шаг 8: Устранение аномалии
Незначительные аномалии вызваны шумом в отчетах и корректируются с помощью локального сглаживания скользящим средним по недавним историческим значениям.
# ————————————— # ИСПРАВЛЕНИЕ АНОМАЛИИ # ————————————— def fix_anomaly(df, idx): window = df.loc[max(0, idx — 3):idx — 1, «Cases»] if len(window) > 0: df.loc[idx, «Cases»] = int(window.mean()) df.loc[idx, «Severity»] = «» df.loc[idx, «Action»] = «Автоматически исправлено агентом ИИ»
Программа берет данные за последние 3 дня, вычисляет их среднее значение и сглаживает аномалию, заменяя полученное значение этим средним. Метод локального скользящего среднего позволяет обрабатывать временные всплески и сбои в данных.
После устранения аномалии точка данных перестает считаться опасной, а уровень серьезности намеренно снижается во избежание путаницы. Показатель «Действие» обновляется до «Автоматически исправлено агентом ИИ».
Полный код
Пожалуйста, ознакомьтесь с полным кодом статистического обнаружения аномалий и реализацией агента ИИ для обработки аномалий.
https://github.com/rautmadhura4/anomaly_detection_agent/tree/main
Результаты
Давайте сравним результаты для страны «Индия» с различными уровнями выявленной серьезности и тем, как ИИ-агент с ними справляется.
Сценарий 1: Нативная реализация
Первая попытка — это нативная реализация, в которой мы обнаруживаем незначительные аномалии, а агент ИИ автоматически их исправляет. Ниже представлен снимок таблицы данных по COVID-19 в Индии с указанием степени тяжести заболевания.
Мы также внедрили панель мониторинга Streamlit для анализа решений и действий ИИ-агента. На приведенном ниже снимке результатов видно, что ИИ-агент исправил различные мелкие аномалии.
Этот метод наиболее эффективен, когда аномалии представляют собой локализованный шум, а не изменения режима.
Сценарий 2: Граничное условие
Здесь обнаруживаются критические аномалии, и агент ИИ помечает их для проверки, как показано на снимке таблицы данных по COVID-19 в Индии с указанием степени тяжести.
На панели управления Streamlit в виде снимка результатов отображаются решения и действия ИИ-агента. Вы можете видеть, что все критические аномалии были отмечены для проверки человеком ИИ-агентом.
Ограничение по степени тяжести предотвращает разрушительные автокоррекции при аномалиях с высоким уровнем воздействия.
Сценарий 3: Ограничение
В рамках сценария с ограничениями были выявлены предупреждающие и критические аномалии, как показано на снимке таблицы данных по COVID-19 в Индии с указанием степени тяжести.
На панели управления Streamlit решения и действия ИИ-агента отображаются ниже в виде снимка результатов. Вы можете видеть, что критическая аномалия помечена для проверки человеком ИИ-агентом, но аномалия уровня WARNING автоматически исправляется. Во многих реальных условиях аномалию уровня WARNING следует сохранять и отслеживать, а не исправлять.
Этот сбой подчеркивает, почему необходимо настраивать пороговые значения для предупреждений и почему проверка человеком остается крайне важной.
Используйте полный код и попробуйте выполнить обнаружение аномалий в наборе данных COVID-19 с различными параметрами.
Перспективы развития и усовершенствования
Мы использовали очень ограниченный набор данных и реализовали обнаружение аномалий на основе правил, но в будущем можно внести некоторые улучшения в реализацию ИИ-агента:
- В нашей реализации обнаруживается аномалия, и решение принимается только на основе количества случаев. В будущем данные могут быть более подробными и содержать такие характеристики, как записи о госпитализации, данные о вакцинации и другие.
- Здесь обнаружение аномалий осуществляется с помощью статистических методов, которые в будущем могут быть дополнены методами машинного обучения для выявления более сложных закономерностей.
- В настоящее время мы внедрили одноагентную архитектуру; в будущем для повышения масштабируемости, наглядности и отказоустойчивости может быть реализована многоагентная архитектура.
- В будущем в рамках обратной связи с участием человека следует также уделять внимание улучшению качества принимаемых решений.
Итоговые выводы
Более интеллектуальные агенты ИИ позволяют создавать оперативные системы искусственного интеллекта, которые принимают решения, используя контекстное мышление, предпринимают действия для устранения аномалий и при необходимости передают задачу человеку. При создании агента ИИ для обнаружения аномалий следует учитывать несколько практических моментов:
- Для выявления аномалий используйте статистические методы и внедряйте агентов искусственного интеллекта для принятия решений в контексте.
- Незначительные аномалии можно безопасно корректировать автоматически, поскольку обычно они воспринимаются как шум. Критические аномалии никогда не следует корректировать автоматически и следует помечать для проверки экспертами в данной области, чтобы не подавлять сигналы реального мира.
- Данный агент ИИ не должен использоваться в ситуациях, когда аномалии напрямую запускают необратимые действия.
При правильном сочетании статистических методов и подхода с использованием агентов искусственного интеллекта, обнаружение аномалий превращается из простой системы оповещения в контролируемую систему, основанную на принятии решений, без ущерба для безопасности.
Источник: towardsdatascience.com
