Rilevamento e interruzione dei cicli infiniti
Limiti al numero massimo di iterazioni, rilevamento delle azioni ripetute e circuit breaker per i cicli
Rilevamento e interruzione dei cicli infiniti è una lezione AI Agents gratuita su CoddyKit. Questa è la lezione 3 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Agents, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Agents include 4 lezioni in totale.
La minaccia dei cicli infiniti
Un agente intrappolato in un ciclo infinito consuma token, blocca risorse e non produce alcun output utile. In produzione, questo si traduce direttamente in denaro sprecato e frustrazione per gli utenti.
Tre meccanismi collaborano per prevenire i cicli infiniti: limiti massimi di iterazioni, rilevamento delle azioni ripetute e timeout.
Protezione 1: limite massimo di iterazioni
La protezione più semplice e importante è un limite rigido per il numero di passaggi. Ogni ciclo dell'agente deve averne uno. Quando il limite viene raggiunto, l'agente restituisce la migliore risposta disponibile in quel momento oppure un messaggio di errore esplicito.
MAX_ITERATIONS = 20
def run_agent(query: str) -> dict:
history = []
for step in range(1, MAX_ITERATIONS + 1):
action = decide_action(query, history)
if action['type'] == 'final_answer':
return {'status': 'ok', 'answer': action['answer'], 'steps': step}
result = execute_tool(action['tool'], action['args'])
history.append({'step': step, 'tool': action['tool'], 'result': result})
# Hard stop — max iterations reached
return {
'status': 'max_iterations_reached',
'answer': None,
'steps': MAX_ITERATIONS
}Protezione 2: rilevare le azioni ripetute
Le azioni ripetute sono il segno distintivo di un ciclo infinito. Tenga traccia della cronologia delle coppie (nome dello strumento, argomenti). Se la stessa coppia compare più di N volte, l'agente è bloccato: interrompa il ciclo e inserisca un messaggio di errore.
import hashlib
import json
def action_hash(tool_name: str, args: dict) -> str:
payload = json.dumps({'tool': tool_name, 'args': args}, sort_keys=True)
return hashlib.md5(payload.encode()).hexdigest()
def run_agent_with_repeat_detection(query: str) -> dict:
history = []
action_counts = {}
for step in range(1, 21):
action = decide_action(query, history)
if action['type'] == 'final_answer':
return {'status': 'ok', 'answer': action['answer']}
key = action_hash(action['tool'], action['args'])
action_counts[key] = action_counts.get(key, 0) + 1
if action_counts[key] > 2: # seen this exact action more than twice
history.append({
'role': 'system',
'content': f'You have called {action["tool"]} with the same arguments {action_counts[key]} times. '
f'This approach is not working. Try a completely different strategy or state what you know so far.'
})
continue
result = execute_tool(action['tool'], action['args'])
history.append({'tool': action['tool'], 'result': result})
return {'status': 'loop_detected', 'answer': None}Tenere traccia delle azioni ripetute in una finestra
Anziché tenere traccia dei conteggi complessivi, rilevi le ripetizioni in una finestra scorrevole degli ultimi N passaggi. In questo modo è possibile intercettare i cicli che variano leggermente, ma si ripetono in un periodo breve.
from collections import deque
def is_cycling(recent_actions: deque, window: int = 6) -> bool:
if len(recent_actions) < window:
return False
# Check if the last window/2 actions repeat the first window/2
half = window // 2
first_half = list(recent_actions)[:half]
second_half = list(recent_actions)[half:window]
return first_half == second_half
# In the agent loop:
# recent_actions = deque(maxlen=6)
# recent_actions.append(action_hash(tool, args))
# if is_cycling(recent_actions):
# print('Cycling detected — breaking loop')
# break
recent = deque(['a', 'b', 'a', 'b'], maxlen=6)
print(is_cycling(recent, window=4)) # True — cycling detectedProtezione 3: timeout basato sul tempo reale con signal.alarm
Sui sistemi Unix, signal.alarm() genera un SIGALRM dopo il numero di secondi specificato. In questo modo si ottiene un timeout rigido anche se il ciclo dell'agente è bloccato in una chiamata lenta a uno strumento.
import signal
class AgentTimeout(Exception):
pass
def timeout_handler(signum, frame):
raise AgentTimeout('Agent exceeded time limit')
def run_agent_with_signal_timeout(query: str, timeout_seconds: int = 60) -> dict:
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(timeout_seconds) # set the alarm
try:
result = run_core_agent_loop(query)
signal.alarm(0) # cancel the alarm on success
return result
except AgentTimeout:
signal.alarm(0)
return {'status': 'timeout', 'answer': None}
except Exception as e:
signal.alarm(0)
raise
# Note: signal.alarm is Unix-only (Linux/Mac)Timeout con threading.Timer (multipiattaforma)
threading.Timer funziona su tutte le piattaforme, incluso Windows. Imposti un flag dopo il periodo di timeout: il ciclo dell'agente controlla il flag e termina se è impostato.
import threading
def run_agent_with_timer_timeout(query: str, timeout_seconds: int = 60) -> dict:
timed_out = threading.Event()
def set_timeout():
timed_out.set()
timer = threading.Timer(timeout_seconds, set_timeout)
timer.start()
history = []
try:
for step in range(1, 21):
if timed_out.is_set():
return {'status': 'timeout', 'answer': None, 'steps': step}
action = decide_action(query, history)
if action['type'] == 'final_answer':
return {'status': 'ok', 'answer': action['answer']}
result = execute_tool(action['tool'], action['args'])
history.append({'tool': action['tool'], 'result': result})
finally:
timer.cancel() # always cancel if done before timeout
return {'status': 'max_steps', 'answer': None}Timeout asincrono con asyncio.wait_for
Nelle architetture di agenti asincroni, utilizzi asyncio.wait_for(coroutine, timeout=N). Genera asyncio.TimeoutError se la coroutine non viene completata entro il numero di secondi specificato.
import asyncio
async def run_async_agent(query: str) -> dict:
history = []
for step in range(1, 21):
action = await async_decide_action(query, history)
if action['type'] == 'final_answer':
return {'status': 'ok', 'answer': action['answer']}
result = await async_execute_tool(action['tool'], action['args'])
history.append({'tool': action['tool'], 'result': result})
return {'status': 'max_steps', 'answer': None}
async def run_with_timeout(query: str, timeout: float = 60.0) -> dict:
try:
return await asyncio.wait_for(run_async_agent(query), timeout=timeout)
except asyncio.TimeoutError:
return {'status': 'timeout', 'answer': None}
# asyncio.run(run_with_timeout('What is Python?', timeout=30.0))Inserire istruzioni di uscita nella cronologia
Quando rileva un ciclo, non si limiti a interromperlo senza segnalarlo. Inserisca un messaggio di sistema nella cronologia della conversazione, spiegando che cosa è accaduto e chiedendo all'agente di concludere. In questo modo l'LLM ha la possibilità di produrre una risposta finale prima dell'interruzione.
def inject_loop_escape(history: list, step: int, reason: str):
message = (
f'[SYSTEM] You have been running for {step} steps. Reason: {reason}. '
f'You MUST now provide a FINAL_ANSWER based on what you have found so far, '
f'even if the information is incomplete. Do not call any more tools.'
)
history.append({'role': 'system', 'content': message})
# In the agent loop, when approaching the limit:
# if step >= MAX_ITERATIONS - 2:
# inject_loop_escape(history, step, 'approaching max iteration limit')
# Or when a repeat is detected:
# if action_counts[key] > 2:
# inject_loop_escape(history, step, 'repeated action detected')
if __name__ == '__main__':
demo_history = []
inject_loop_escape(demo_history, step=18, reason='approaching max iteration limit')
print(demo_history[-1]['content'])
Registrare l'interruzione di un ciclo
Registri sempre l'attivazione di una protezione del ciclo. In questo modo crea un resoconto della frequenza e dei motivi per cui gli agenti si bloccano: dati estremamente utili per migliorare i prompt e le implementazioni degli strumenti.
import logging
logger = logging.getLogger('agent_guard')
def check_and_break_loop(step: int, action_counts: dict, current_key: str) -> bool:
count = action_counts.get(current_key, 0)
if count > 2:
logger.warning(
f'Infinite loop detected at step {step}. '
f'Action hash {current_key[:8]} seen {count} times. '
f'Breaking loop.'
)
return True # signal to break
if step >= 18: # approaching limit
logger.warning(
f'Approaching max iterations at step {step}. '
f'Injecting escape prompt.'
)
return False
if __name__ == '__main__':
import sys
logging.basicConfig(level=logging.WARNING, format='%(message)s', stream=sys.stdout)
demo_counts = {'search:{"q": "weather"}': 3}
check_and_break_loop(step=10, action_counts=demo_counts, current_key='search:{"q": "weather"}')
Combinare tutte e tre le protezioni
Gli agenti in produzione dovrebbero combinare tutte e tre le protezioni: numero massimo di iterazioni (obbligatorio), rilevamento delle ripetizioni (intercetta i cicli) e timeout (intercetta le chiamate bloccanti agli strumenti). Insieme rendono il ciclo estremamente robusto.
import threading
import hashlib
import json
def run_production_agent(query: str) -> dict:
MAX_STEPS = 20
TIMEOUT_SEC = 120
timed_out = threading.Event()
timer = threading.Timer(TIMEOUT_SEC, timed_out.set)
timer.start()
history = []
action_counts = {}
try:
for step in range(1, MAX_STEPS + 1):
if timed_out.is_set():
return {'status': 'timeout'}
action = decide_action(query, history)
if action['type'] == 'final_answer':
return {'status': 'ok', 'answer': action['answer']}
key = hashlib.md5(json.dumps(action, sort_keys=True).encode()).hexdigest()
action_counts[key] = action_counts.get(key, 0) + 1
if action_counts[key] > 2:
inject_loop_escape(history, step, 'repeat detected')
continue
result = execute_tool(action['tool'], action['args'])
history.append({'tool': action['tool'], 'result': result})
finally:
timer.cancel()
return {'status': 'max_steps'}Testare le protezioni del ciclo con test unitari
Scriva test unitari dedicati per le protezioni del ciclo. Crei un agente simulato che chiami sempre lo stesso strumento e verifichi che la protezione lo rilevi entro il numero di passaggi previsto.
from unittest.mock import MagicMock
def test_repeat_detection_breaks_loop():
# Create a mock that always returns the same action
always_same_action = MagicMock(return_value={
'type': 'tool',
'tool': 'search_web',
'args': {'query': 'same query'}
})
always_success = MagicMock(return_value='some result')
result = run_agent_with_repeat_detection(
query='test',
decide_action=always_same_action,
execute_tool=always_success
)
# Should stop due to loop detection, not run all 20 steps
assert result['status'] in ('loop_detected', 'max_iterations_reached')
# Should not have run all 20 steps (loop should be detected by step 6-7)
print('Loop guard test passed')Verifica delle conoscenze: rilevare i cicli infiniti
Verifichi la Sua comprensione delle tecniche per rilevare e interrompere i cicli infiniti.
Riepilogo: rilevare e interrompere i cicli infiniti
Ora è in grado di proteggere gli agenti dai cicli infiniti con tre protezioni complementari:
- Numero massimo di iterazioni: limite rigido per i passaggi, obbligatorio in ogni ciclo dell'agente
- Rilevamento delle ripetizioni: calcoli l'hash delle coppie (strumento, argomenti) e interrompa il ciclo quando compaiono più di N volte
- Timeout:
signal.alarm()su Unix,threading.Timersu più piattaforme,asyncio.wait_for()per le operazioni asincrone - Inserimento di istruzioni di uscita: dia all'agente la possibilità di concludere prima dell'interruzione forzata
- Registri sempre l'attivazione di una protezione: sono dati preziosi per migliorare i prompt
Domande Frequenti
La lezione «Rilevamento e interruzione dei cicli infiniti» è gratuita?
Sì — il testo completo di «Rilevamento e interruzione dei cicli infiniti» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Agents, passa a CoddyKit PRO. Il corso AI Agents include 4 lezioni in totale.
Cosa imparerò in «Rilevamento e interruzione dei cicli infiniti»?
Limiti al numero massimo di iterazioni, rilevamento delle azioni ripetute e circuit breaker per i cicli Eserciti AI Agents con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.
Ho bisogno di esperienza per iniziare AI Agents?
Non è richiesta alcuna esperienza precedente. AI Agents su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 3 di 4.
Quanto tempo richiede la lezione «Rilevamento e interruzione dei cicli infiniti»?
La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.
Posso scrivere ed eseguire codice in questa lezione AI Agents?
Sì. Ogni lezione AI Agents include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.
Tutte le lezioni di questo corso
- Errori comuni nei cicli degli agenti
- Trace logging per i passaggi degli agenti
- Rilevamento e interruzione dei cicli infiniti
- Tecniche di debugging passo passo