Как вы ограничиваете бесконечный цикл агента?

Краткий тезис

Агент, взаимодействующий с внешними инструментами, может зациклиться — многократно вызывать один и тот же инструмент с тем же результатом, не приближаясь к ответу. Для предотвращения применяют жёсткий лимит шагов (max_iterations), таймауты на выполнение инструментов, детектор повторяющихся действий и оценку прогресса. Критически важна комбинация методов: ни один из них не даёт 100% защиты, а также требуется логирование и мониторинг для послеаварийного анализа. В продакшене обязательно добавляют human-in-the-loop как ручной предохранитель.

1. Термин: Бесконечный цикл агента (Agent infinite loop)

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

Почему это критично

• Расходует квоты API (деньги) и время пользователя.
• Может привести к thrashing — многократному созданию и удалению ресурсов.
• В крайних случаях — бесконечное выполнение, блокирующее очередь агентов.

Термин «Thrashing» — ситуация, когда система тратит больше ресурсов на управление, чем на полезную работу.

2. Жёсткий лимит шагов (max_iterations)

Самый простой и надёжный способ. Устанавливаем максимальное количество итераций агента (обычно 10–15). Если лимит превышен — завершаем с ошибкой или возвращаем частичный результат.

Реализация в коде (Python, псевдо-агент):

MAX_ITERATIONS = 12

def run_agent(query):
    state = {"query": query, "history": [], "step": 0}
    while state["step"] < MAX_ITERATIONS:
        action = agent_think(state)  # LLM решает, что делать
        if action == "final_answer":
            return format_final_answer(state)
        result = execute_tool(action)
        state["history"].append((action, result))
        state["step"] += 1
    raise RuntimeError(f"Agent reached max iterations ({MAX_ITERATIONS})")

Вариации:

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

3. Таймаут на выполнение инструмента

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

Пример с asyncio

import asyncio

async def call_tool_with_timeout(tool_name, args, timeout=30):
    try:
        return await asyncio.wait_for(tool_function(tool_name, args), timeout=timeout)
    except asyncio.TimeoutError:
        return {"error": f"Tool {tool_name} timed out after {timeout}s"}

Дополнительно

• Можно задавать разный таймаут для разных инструментов (база данных — 10 с, внешний API — 60 с).
• При повторяющихся таймаутах — временно отключать инструмент или использовать fallback.

Термин «Fallback» — резервный инструмент или стратегия, когда основной недоступен.

4. Обнаружение повторяющихся действий (Action deduplication)

Агент может зациклиться на одном и том же действии (одинаковый инструмент и одинаковые аргументы). Обнаружение: ведём хэш-таблицу (или set) из (tool_name + args). Если такое действие уже выполнялось — прерываем.

Пример реализации

action_set = set()

def execute_safe(agent_action):
    key = (agent_action.tool, json.dumps(agent_action.args, sort_keys=True))
    if key in action_set:
        raise LoopDetectedError(f"Repeated action: {key}")
    action_set.add(key)
    return agent_action.execute()

Когда срабатывает

• Агент дважды вызывает один поисковый запрос с одинаковыми параметрами.
• Ошибка "нет доступа" → агент повторяет запрос с теми же кредами.

Проблема

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

5. Оценка прогресса (Progress check)

Агент может не повторять одно и то же действие, но и не продвигаться к цели. Например, он может перебирать бесконечные варианты одного инструмента. Для оценки прогресса смотрят на изменение состояния (observation) за последние N шагов.

Метрики прогресса

• Изменение текста наблюдения если последние 3 наблюдения практически одинаковы (cosine similarity > 0.95) → стоп.
• Достижение подцели разбиваем задачу на подзадачи и проверяем, сколько выполнено.
• Прирост уверенности если LLM-оценка уверенности не растёт → стоп.

Пример с хэшированием наблюдений

from collections import deque

observation_history = deque(maxlen=3)

def check_progress(new_observation):
    observation_history.append(new_observation)
    if len(observation_history) < 3:
        return True
    # проверяем, что наблюдения различаются (например, по содержанию)
    hash_0 = hash(observation_history[0])
    hash_1 = hash(observation_history[1])
    hash_2 = hash(observation_history[2])
    if hash_0 == hash_1 == hash_2:
        return False   # нет прогресса
    return True

Термин «Observation» — результат выполнения инструмента (текст, данные, ошибка), который возвращается агенту.

6. Human-in-the-loop (HITL)

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

Реализация

def agent_loop_with_hitl():
    for step in range(MAX_ITERATIONS):
        # ... обычная логика
        if step % 5 == 0 and step > 0:
            user_input = input("Agent requests approval. Continue? (y/n): ")
            if user_input.lower() != 'y':
                return "Cancelled by user"
    ...

Термин «Human-in-the-loop» — процесс, в котором человек может вмешаться в автоматическое принятие решений агента.

7. Дополнительные техники

7.1. Детектор семантических луп

Анализировать не точные хэши, а семантическое сходство действий. Например, агент вызывает поиск с синонимами ( "погода Москва", "погода в Москве", "температура Москва" ). Это разные строки, но по сути одно и то же. Используйте эмбеддинги и порог косинусной близости.

7.2. Бюджет токенов

Ограничивать суммарное количество потраченных токенов (входных + выходных) на выполнение задачи. Когда бюджет исчерпан — принудительная остановка.

7.3. Флаг "нет изменений в плане"

Агент может явно декларировать свой план (например, в формате JSON). Если план не меняется на протяжении нескольких шагов — вероятен цикл.

7.4. Эвристики по числу вызовов одного инструмента

Например, не более 3 вызовов одного и того же инструмента подряд (если не было значимого изменения контекста).

8. Мониторинг и логирование

Без сбора метрик вы не узнаете, что циклы происходят. Рекомендуется:

• Логировать каждый шаг агента: что он решил, какой инструмент вызвал, с какими аргументами, какой результат.
• Метрики: количество шагов на задачу, число повторных действий, процент задач, завершившихся по лимиту.
• Alerting — если частота превышения лимитов растёт → менять конфигурацию или дообучать промпт.

Пример структуры лога

{
  "agent_id": "alpha-01",
  "task_id": "t-123",
  "step": 7,
  "action": {
    "tool": "search_web",
    "args": {"q": "погода москва 2025"}
  },
  "observation": "...",
  "stopped_by": "max_iterations"
}

9. Обработка ошибок и деградация

Если агент остановлен из-за цикла, что возвращать пользователю?

• Красивое сообщение «Не удалось найти ответ за отведённое время. Попробуйте уточнить запрос».
• Частичный ответ если на предыдущих шагах была полезная информация, собрать её в краткий отчёт.
• Эскалация передать задачу человеку или системе с большим лимитом.

10. Тестирование ограничений

Напишите unit-тесты на каждое ограничение в отдельности, а также интеграционные тесты, где агент искусственно зацикливается.

Пример теста (pytest):

def test_agent_loops_fails_after_max_iterations():
    agent = MockAgent(behave="loops forever")
    with pytest.raises(MaxIterationsReached):
        run_agent(agent, query="test")

Создайте синтетические сценарии: агент получает от инструмента всегда одно и то же; агент не находит ответа и не знает, как остановиться.

Пет-проект для закрепления

Задача Реализовать простого AI-агента (на базе OpenAI API), который умеет выполнять две задачи: поиск по Википедии (через wikipedia-api) и калькулятор. Добавить все описанные ограничения. Измерить, сколько задач завершились нормально, сколько были прерваны, и при каких настройках (max_iterations=5, 10, 15) достигается лучший баланс.

Инструменты Python, библиотека openai, wikipedia, asyncio, pytest.

Шаги:

1. Напишите класс Agent с методами think, execute_tool, check_loops.
2. Реализуйте ограничения: max_iterations, timeout, deduplication, progress check.
3. Создайте 10 тестовых запросов, включая заведомо зацикливающиеся (например, "вычисли 1+1, потом 2+2, потом снова 1+1").
4. Запустите серию экспериментов с разными max_iterations.
5. Выведите статистику: среднее число шагов, число прерываний, время выполнения.

Ожидаемый результат Вы увидите, что при малых лимитах (5) многие полезные задачи не успевают завершиться, при больших (30) — зацикленные задачи расходуют много токенов. Оптимум — 10–15.

Связь с другими вопросами

Навигация

• Предыдущий: 49
• Следующий: 51
• Индекс: 00. Индекс разборов