Зависший AI-агент / loop

Самая дорогая поломка в AI-коде — не падение, а тишина. Агент думает, ходит в tools, тратит токены, но никогда не закончит сам:

• зациклился на тех же tool-вызовах с разными аргументами;
• ждёт ответа от внешнего сервиса, который тихо умер (а у LLM нет таймаута);
• застрял в «давай я ещё разочек посмотрю», стек planner-ов вырос до 30 уровней.

Защита от этого — heartbeat-уведомления Notifly.

Шаблон 1: heartbeat «агент жив, шаг N»

Создайте heartbeat в админке с интервалом в 1.5–2× больше, чем максимально ожидаемое время одного шага агента. Полученный URL включите в цикл:

import os, requests

PING_URL = os.environ["AGENT_PING_URL"]   # https://.../heartbeat/ping/H...

def step(state):
    # один шаг agent-loop: tool / model / tool / ...
    ...
    requests.get(PING_URL, timeout=3)     # маркер «я жив»
    return new_state

Если шаг не уложится в intervalSec + graceSec — Notifly пришлёт алёрт. Recovery-сообщение тоже стоит включить — когда агент «оживёт» (пройдёт следующий ping), вы получите подтверждение.

Шаблон 2: heartbeat «общая продолжительность»

Если шаги короткие, а проблема в том, что агент идёт по кругу — heartbeat не сработает. Здесь нужен таймер сверху:

import time, threading, requests

DEADLINE = time.time() + 30 * 60   # 30 минут — потолок

def watchdog():
    while time.time() < DEADLINE:
        time.sleep(30)
    requests.post(f"{os.environ['NOTIFLY_URL']}/message",
                  params={"token": os.environ["NOTIFLY_TOKEN"]},
                  json={"title":   "🤖⏳ Агент превысил 30 минут",
                        "message": "Похоже, loop. Проверьте логи / убейте процесс.",
                        "priority": 9},
                  timeout=5)

threading.Thread(target=watchdog, daemon=True).start()
agent.run(...)   # если закончит вовремя — поток умрёт вместе с процессом

Шаблон 3: ловим тривиальный loop в tool-calls

Сравниваем хеш последних N tool-вызовов: если совпадает — почти наверняка loop. Раннее обнаружение, push, можно тут же остановить.

import collections, hashlib, json

class LoopGuard:
    def __init__(self, window=6, repeats=3):
        self.window  = window
        self.repeats = repeats
        self.hist    = collections.deque(maxlen=window)

    def step(self, tool_name, args):
        h = hashlib.sha1(json.dumps([tool_name, args], sort_keys=True).encode()).hexdigest()
        self.hist.append(h)
        if len(self.hist) == self.window and self.hist.count(h) >= self.repeats:
            notify("🔁 Агент в loop",
                   f"Повторяется {tool_name} с теми же аргументами {self.repeats}+ раз",
                   prio=9)
            raise RuntimeError("agent loop detected")

guard = LoopGuard()
for tool_call in agent.iter():
    guard.step(tool_call.name, tool_call.arguments)
    tool_call.execute()

Шаблон 4: внешний watchdog через Notifly heartbeat

Если у агента нет надёжного места «обернуть цикл», запустите его как systemd-сервис, который пингует heartbeat:

[Service]
ExecStart=/usr/local/bin/run-agent.sh
ExecStartPost=/bin/sh -c 'while kill -0 $MAINPID; do curl -fsS $AGENT_PING_URL; sleep 60; done'

Тут heartbeat пингуется снаружи, и алёрт прилетит при любом «зависании» основного процесса (включая kill -STOP).

Что слать в push

Чтобы из telegram-окна решить «убить или пусть подумает ещё»:

• сколько прошло секунд / итераций;
• последние 3 tool-вызова (имя, длина аргументов);
• текущий «план» (если агент его публикует);
• сколько уже потрачено токенов / денег за этот run.

Связанные рецепты

• Завершение долгой задачи — обратное: «успел вовремя».
• Расходы на LLM API — иногда первый сигнал о loop — счёт.
• Heartbeat (dead-man-switch) — общая концепция.