Debugging agent loop: 6 pattern di deadlock e come uscirne

Debugging agent loop: 6 pattern di deadlock e come uscirne

Il tuo agent è partito con un obiettivo chiaro. Dieci minuti dopo è ancora lì che gira, ha consumato 5.000 token e €4 di API, e l'unico progresso visibile è un log pieno di "let me try X again". Non è un caso raro: è uno dei failure mode più frequenti degli agent LLM in produzione, e uno dei più difficili da diagnosticare senza gli strumenti giusti.

Il problema è strutturale. Un LLM agent è un ciclo: osserva, ragiona, agisce, osserva il risultato, ragiona di nuovo. Questo loop è la sua forza - permette all'agent di correggere errori e adattarsi. Ma è anche la sua vulnerabilità: basta una condizione ambigua, un tool che restituisce output malformato, o un contesto che degrada, e il loop smette di convergere. L'agent continua a girare, ma non va da nessuna parte.

In questo episodio catalogo i 6 pattern di deadlock che ho osservato e documentato, mostro come rilevarli automaticamente con Python, e costruisco le strategie di recovery che uso nel ml-service del mio portfolio.

I 6 Pattern di Deadlock: Tassonomia e Fingerprint

Prima di costruire i detector, devi sapere cosa stai cercando. Ho classificato sei failure mode distinti, ognuno con un fingerprint rilevabile nel trace dell'agent. Non sono mutualmente esclusivi - un agent in crisi può esibire più pattern contemporaneamente, il che rende il debugging ancora più complesso.

Ogni pattern ha un fingerprint distinto nel trace. Questo è il punto chiave: non puoi aspettarti che l'agent si accorga da solo di essere in loop - il modello LLM non ha visibilità sulla propria storia di esecuzione a meno che tu non la metta esplicitamente nel contesto. Il rilevamento deve essere infrastrutturale, esterno al loop.

Snippet 1 - Loop Detection con State Hash

Il primo strumento da implementare è il più semplice: un hash dello stato dell'agent ad ogni iterazione. Se lo stesso hash compare due volte, sei in loop deterministico. Questo cattura l'hard loop e parte della tool oscillation.

import hashlib
import json
from dataclasses import dataclass, field
from typing import Any


@dataclass
class AgentStep:
    """Rappresenta un singolo step nell'esecuzione dell'agent."""
    iteration: int
    thought: str
    tool_name: str | None
    tool_args: dict[str, Any]
    tool_result: str | None


def compute_state_hash(step: AgentStep) -> str:
    """
    Calcola un hash deterministico dello stato dell'agent.
    Ignora l'iteration number per catturare loop esatti.
    """
    # Normalizza: ordina le chiavi per hash stabile
    state = {
        "tool_name": step.tool_name,
        "tool_args": step.tool_args,
        # Includi i primi 200 char del thought per catturare reflection loop
        "thought_prefix": step.thought[:200] if step.thought else None,
    }
    payload = json.dumps(state, sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(payload.encode()).hexdigest()


class CycleDetector:
    """
    Rileva cicli esatti nello stato dell'agent tramite state hashing.
    Cattura: hard loop, tool oscillation (con finestra scorrevole).
    """

    def __init__(self, window_size: int = 10):
        # Finestra scorrevole: controlla solo gli ultimi N hash
        self.window_size = window_size
        self._hash_history: list[str] = []

    def check(self, step: AgentStep) -> tuple[bool, str | None]:
        """
        Ritorna (is_cycle, hash_duplicato).
        Se is_cycle è True, l'hash compare già nella finestra.
        """
        current_hash = compute_state_hash(step)

        # Finestra scorrevole: solo gli ultimi window_size hash
        recent = self._hash_history[-self.window_size:]

        if current_hash in recent:
            return True, current_hash

        self._hash_history.append(current_hash)
        return False, None

    def reset(self) -> None:
        """Reset dopo una recovery riuscita."""
        self._hash_history.clear()

La finestra scorrevole è importante: non vuoi confrontare ogni step con tutta la storia (costoso), ma solo con gli ultimi N. Per la tool oscillation A→B→A la finestra di 10 è più che sufficiente. Per cicli più lunghi (es. A→B→C→A), aumenta la finestra a 20-30.

Snippet 2 - I 6 Deadlock Pattern come Enum Classificato

Prima di costruire il detector completo, definisci i pattern come enum tipizzato. Questo ti permette di loggare strutturalmente il tipo di deadlock rilevato, filtrarli nei dashboard di monitoring, e applicare la recovery strategy corretta per ciascuno.

from enum import Enum, auto
from dataclasses import dataclass


class DeadlockPattern(Enum):
    """
    I 6 pattern di deadlock documentati negli agent loop LLM.
    Ognuno corrisponde a una recovery strategy specifica.
    """
    # Tool A invocato, poi disfatto da tool B, poi A di nuovo: ciclo binario
    TOOL_OSCILLATION = auto()

    # L'agent ha perso l'obiettivo originale: output non correlati al goal
    CONTEXT_ROT = auto()

    # Troppo tempo in thought/reflection, zero azioni concrete
    REFLECTION_PARALYSIS = auto()

    # L'agent chiama tool che non esistono o non sono nello schema
    TOOL_HALLUCINATION = auto()

    # Action bloccata da guardrail, rephrase, stessa action bloccata ancora
    GUARDRAIL_LOOP = auto()

    # Tool flaky: retry storm senza circuit breaker
    RETRY_STORM = auto()


@dataclass(frozen=True)
class DeadlockEvent:
    """Evento di deadlock rilevato, loggato nel trace strutturato."""
    pattern: DeadlockPattern
    iteration: int
    evidence: str          # Descrizione leggibile dell'evidenza
    severity: str          # "warning" | "critical"
    suggested_action: str  # Azione di recovery consigliata


# Mappa pattern → recovery strategy predefinita
RECOVERY_MAP: dict[DeadlockPattern, str] = {
    DeadlockPattern.TOOL_OSCILLATION: "apply_state_latch",
    DeadlockPattern.CONTEXT_ROT: "inject_objective_anchor",
    DeadlockPattern.REFLECTION_PARALYSIS: "force_action_prompt",
    DeadlockPattern.TOOL_HALLUCINATION: "inject_tool_schema",
    DeadlockPattern.GUARDRAIL_LOOP: "escalate_to_human",
    DeadlockPattern.RETRY_STORM: "activate_circuit_breaker",
}

Snippet 3 - Max Iteration Safety Guard

Il guard più semplice e più critico: un limite hard sulle iterazioni, con fallback strutturato. Non deve essere un semplice counter - deve portare informazioni sul perché l'agent si è fermato, per permettere debug e retry intelligente.

Nel ml-service del mio portfolio uso un limite di 10 iterazioni per l'agent RAG principale, con un risultato di fallback che include il partial context accumulato. Questo permette al chiamante di decidere se riprovare con un task ridotto o escalare.

from dataclasses import dataclass, field
from typing import Any, Callable
import time


@dataclass
class AgentResult:
    """Risultato strutturato dell'esecuzione agent, successo o fallimento."""
    success: bool
    output: Any | None
    iterations_used: int
    stop_reason: str          # "completed" | "max_iterations" | "deadlock" | "error"
    deadlock_events: list[DeadlockEvent] = field(default_factory=list)
    partial_context: str | None = None   # Contesto parziale se stop anticipato
    execution_time_ms: float = 0.0


class MaxIterationGuard:
    """
    Safety guard che impone un limite hard alle iterazioni dell'agent.
    Integra cycle detection e restituisce AgentResult strutturato.

    Esempio uso nel ml-service (agent RAG con max 10 iterazioni):
        guard = MaxIterationGuard(max_iterations=10, agent_id="rag-main")
        result = guard.run(agent_fn=my_agent_step, initial_state=state)
    """

    def __init__(
        self,
        max_iterations: int,
        agent_id: str,
        cycle_detector: CycleDetector | None = None,
        on_max_reached: Callable[[list[AgentStep]], str] | None = None,
    ):
        if max_iterations < 1:
            raise ValueError("max_iterations deve essere almeno 1")

        self.max_iterations = max_iterations
        self.agent_id = agent_id
        self._cycle_detector = cycle_detector or CycleDetector()
        # Hook opzionale: genera partial context quando si raggiunge il limite
        self._on_max_reached = on_max_reached
        self._steps: list[AgentStep] = []
        self._deadlock_events: list[DeadlockEvent] = []

    def run(
        self,
        agent_fn: Callable[[list[AgentStep]], AgentStep | None],
        initial_state: list[AgentStep] | None = None,
    ) -> AgentResult:
        """
        Esegue il loop agent con guard attivo.
        agent_fn: funzione che riceve la storia degli step e ritorna il prossimo step
                  o None se il task è completato.
        """
        self._steps = list(initial_state or [])
        start_ms = time.monotonic() * 1000

        for i in range(self.max_iterations):
            # Esegui lo step
            step = agent_fn(self._steps)

            # None = agent ha completato il task
            if step is None:
                return AgentResult(
                    success=True,
                    output=self._steps[-1].tool_result if self._steps else None,
                    iterations_used=i + 1,
                    stop_reason="completed",
                    deadlock_events=self._deadlock_events,
                    execution_time_ms=time.monotonic() * 1000 - start_ms,
                )

            self._steps.append(step)

            # Controlla ciclo esatto
            is_cycle, _ = self._cycle_detector.check(step)
            if is_cycle:
                event = DeadlockEvent(
                    pattern=DeadlockPattern.TOOL_OSCILLATION,
                    iteration=i + 1,
                    evidence=f"State hash duplicato a iterazione {i + 1}",
                    severity="critical",
                    suggested_action=RECOVERY_MAP[DeadlockPattern.TOOL_OSCILLATION],
                )
                self._deadlock_events.append(event)
                partial = self._on_max_reached(self._steps) if self._on_max_reached else None
                return AgentResult(
                    success=False,
                    output=None,
                    iterations_used=i + 1,
                    stop_reason="deadlock",
                    deadlock_events=self._deadlock_events,
                    partial_context=partial,
                    execution_time_ms=time.monotonic() * 1000 - start_ms,
                )

        # Limite raggiunto senza completamento
        partial = self._on_max_reached(self._steps) if self._on_max_reached else None
        return AgentResult(
            success=False,
            output=None,
            iterations_used=self.max_iterations,
            stop_reason="max_iterations",
            deadlock_events=self._deadlock_events,
            partial_context=partial,
            execution_time_ms=time.monotonic() * 1000 - start_ms,
        )

Snippet 4 - Repetition Detector con N-gram Semantici

Lo state hash rileva loop esatti. Non rileva la reflection paralysis o il context rot, dove l'output cambia superficialmente ma rimane semanticamente identico. Per questo serve un detector basato su similarità testuale.

L'approccio più leggero senza dipendere da un modello di embedding è usare n-gram di caratteri con Jaccard similarity. Per casi dove hai già un embedding model disponibile (come nel ml-service con sentence-transformers), puoi usare la cosine similarity invece.

from collections import Counter
import re


def extract_char_ngrams(text: str, n: int = 4) -> Counter:
    """
    Estrae n-gram di caratteri da un testo normalizzato.
    N-gram di char sono robusti a variazioni sintattiche minori.
    """
    # Normalizza: lowercase, rimuovi spazi multipli
    normalized = re.sub(r"\s+", " ", text.lower().strip())
    return Counter(normalized[i:i+n] for i in range(len(normalized) - n + 1))


def jaccard_similarity(a: Counter, b: Counter) -> float:
    """Jaccard similarity tra due bag-of-ngrams. Range [0, 1]."""
    if not a and not b:
        return 1.0
    intersection = sum((a & b).values())
    union = sum((a | b).values())
    return intersection / union if union > 0 else 0.0


class RepetitionDetector:
    """
    Rileva reflection paralysis e context rot confrontando output
    consecutivi con Jaccard similarity su character n-gram.

    Soglie suggerite:
    - similarity > 0.85: output quasi identico → reflection paralysis
    - similarity > 0.70 per 3+ step consecutivi → soft loop semantico
    """

    def __init__(
        self,
        similarity_threshold: float = 0.85,
        consecutive_threshold: int = 2,
        ngram_size: int = 4,
    ):
        self.similarity_threshold = similarity_threshold
        self.consecutive_threshold = consecutive_threshold
        self.ngram_size = ngram_size
        self._recent_ngrams: list[Counter] = []
        self._consecutive_similar: int = 0

    def check(self, text: str) -> tuple[bool, float]:
        """
        Controlla se il testo è semanticamente simile ai precedenti.
        Ritorna (is_repetition, max_similarity_score).
        """
        current = extract_char_ngrams(text, self.ngram_size)

        if not self._recent_ngrams:
            self._recent_ngrams.append(current)
            return False, 0.0

        # Confronta con gli ultimi 3 output
        similarities = [
            jaccard_similarity(current, prev)
            for prev in self._recent_ngrams[-3:]
        ]
        max_sim = max(similarities)

        if max_sim >= self.similarity_threshold:
            self._consecutive_similar += 1
        else:
            self._consecutive_similar = 0

        self._recent_ngrams.append(current)

        is_repetition = self._consecutive_similar >= self.consecutive_threshold
        return is_repetition, max_sim

    def classify_pattern(self, similarity: float, token_ratio_reflection: float) -> DeadlockPattern:
        """
        Data la similarity e il ratio token in reflection,
        classifica il pattern di deadlock semantico.
        """
        if token_ratio_reflection > 0.5 and similarity > 0.7:
            return DeadlockPattern.REFLECTION_PARALYSIS
        return DeadlockPattern.CONTEXT_ROT

Sezione 5 - Escalation a Umano

Alcune situazioni non si risolvono automaticamente. Il guardrail loop - dove l'agent cerca di eseguire un'azione che viola le policy, la riformula, e si trova bloccato di nuovo - è il caso tipico. L'agent non ha modo di sapere perché è bloccato: non ha accesso alla logica del guardrail. In questi casi, l'unica recovery sicura è sospendere il task e chiedere input umano.

import asyncio
from enum import Enum
from dataclasses import dataclass
from typing import Callable, Awaitable


class EscalationReason(Enum):
    GUARDRAIL_LOOP = "guardrail_loop"
    MAX_RETRIES_EXHAUSTED = "max_retries_exhausted"
    UNRECOVERABLE_ERROR = "unrecoverable_error"
    AMBIGUOUS_OBJECTIVE = "ambiguous_objective"


@dataclass
class EscalationRequest:
    agent_id: str
    session_id: str
    reason: EscalationReason
    context_summary: str      # Sommario del contesto per l'operatore umano
    last_blocked_action: str  # Ultima azione bloccata
    suggested_resolution: str # Suggerimento automatico per l'operatore


@dataclass
class EscalationResponse:
    approved: bool
    operator_instruction: str | None  # Istruzione specifica dell'operatore
    modified_objective: str | None    # Obiettivo ridefinito, se applicabile


class HumanEscalationHandler:
    """
    Gestisce l'escalation a umano quando le recovery automatiche falliscono.
    Implementa un timeout: se l'operatore non risponde entro N secondi,
    il task viene terminato con stop_reason="escalation_timeout".
    """

    def __init__(
        self,
        notify_fn: Callable[[EscalationRequest], Awaitable[None]],
        response_fn: Callable[[str], Awaitable[EscalationResponse | None]],
        timeout_seconds: int = 300,  # 5 minuti default
    ):
        # notify_fn: invia notifica all'operatore (es. Telegram, email, webhook)
        self._notify = notify_fn
        # response_fn: polling o webhook per la risposta dell'operatore
        self._get_response = response_fn
        self.timeout_seconds = timeout_seconds

    async def escalate(
        self,
        request: EscalationRequest,
    ) -> EscalationResponse:
        """
        Invia la richiesta di escalation e attende la risposta entro il timeout.
        Se timeout scade: ritorna EscalationResponse(approved=False, ...) con
        operator_instruction="timeout" per permettere cleanup ordinato.
        """
        await self._notify(request)

        try:
            response = await asyncio.wait_for(
                self._get_response(request.session_id),
                timeout=self.timeout_seconds,
            )
            if response is None:
                return EscalationResponse(
                    approved=False,
                    operator_instruction="no_response",
                    modified_objective=None,
                )
            return response

        except asyncio.TimeoutError:
            return EscalationResponse(
                approved=False,
                operator_instruction="timeout",
                modified_objective=None,
            )

    def should_escalate(self, events: list[DeadlockEvent]) -> bool:
        """
        Determina se escalare basandosi sugli eventi di deadlock accumulati.
        Escalation quando: guardrail loop critico, o 3+ eventi deadlock diversi.
        """
        if any(
            e.pattern == DeadlockPattern.GUARDRAIL_LOOP and e.severity == "critical"
            for e in events
        ):
            return True

        unique_patterns = {e.pattern for e in events}
        return len(unique_patterns) >= 3

Snippet 6 - Debug Replay Tool

Dopo che un loop è stato bloccato, devi capire cosa è andato storto. Il modo più efficace è il replay deterministico: salvi il trace completo durante l'esecuzione, poi puoi "rileggere" la sessione step by step, iniettando mock responses per i tool, e osservare esattamente quando e perché il loop si è formato.

from dataclasses import dataclass, field, asdict
from typing import Any
import json
from pathlib import Path
import datetime


@dataclass
class TraceEntry:
    """Un entry nel trace completo della sessione agent."""
    iteration: int
    timestamp_iso: str
    thought: str
    tool_name: str | None
    tool_args: dict[str, Any]
    tool_result: str | None
    state_hash: str
    similarity_to_prev: float
    deadlock_events: list[dict]  # Serializzato da DeadlockEvent


class AgentTraceRecorder:
    """
    Registra il trace completo dell'esecuzione agent in formato JSON strutturato.
    Permette replay post-mortem e analisi dei pattern di deadlock.
    """

    def __init__(self, session_id: str, agent_id: str):
        self.session_id = session_id
        self.agent_id = agent_id
        self._entries: list[TraceEntry] = []

    def record(
        self,
        step: AgentStep,
        state_hash: str,
        similarity_to_prev: float,
        deadlock_events: list[DeadlockEvent],
    ) -> None:
        """Registra un singolo step nel trace."""
        entry = TraceEntry(
            iteration=step.iteration,
            timestamp_iso=datetime.datetime.utcnow().isoformat() + "Z",
            thought=step.thought,
            tool_name=step.tool_name,
            tool_args=step.tool_args,
            tool_result=step.tool_result,
            state_hash=state_hash,
            similarity_to_prev=round(similarity_to_prev, 4),
            deadlock_events=[
                {
                    "pattern": e.pattern.name,
                    "iteration": e.iteration,
                    "evidence": e.evidence,
                    "severity": e.severity,
                }
                for e in deadlock_events
            ],
        )
        self._entries.append(entry)

    def dump(self, output_path: Path) -> dict:
        """
        Serializza il trace in JSON strutturato e lo salva su disco.
        Ritorna il dict per logging immediato.
        """
        trace = {
            "session_id": self.session_id,
            "agent_id": self.agent_id,
            "total_iterations": len(self._entries),
            "entries": [asdict(e) for e in self._entries],
        }
        output_path.write_text(json.dumps(trace, indent=2, ensure_ascii=False))
        return trace

    def replay(self, up_to_iteration: int) -> list[TraceEntry]:
        """
        Ritorna le entry fino all'iterazione specificata per debug replay.
        Utile per isolare il momento esatto in cui il loop si è formato.
        """
        return [e for e in self._entries if e.iteration <= up_to_iteration]

Snippet 7 - Structured Trace Dump JSON

Ecco un esempio di output del trace dump prodotto da AgentTraceRecorder.dump(). Questa struttura è progettata per essere ingestita direttamente da un tool di monitoring o analizzata manualmente per post-mortem.

{
  "session_id": "sess-20260608-rag-main-001",
  "agent_id": "rag-main",
  "total_iterations": 8,
  "entries": [
    {
      "iteration": 1,
      "timestamp_iso": "2026-06-08T09:14:02Z",
      "thought": "Devo cercare informazioni sull'articolo richiesto nel vector store.",
      "tool_name": "vector_search",
      "tool_args": {"query": "agent loop debugging", "top_k": 5},
      "tool_result": "5 risultati trovati",
      "state_hash": "a3f9e2c1d84b",
      "similarity_to_prev": 0.0,
      "deadlock_events": []
    },
    {
      "iteration": 6,
      "timestamp_iso": "2026-06-08T09:14:38Z",
      "thought": "I risultati non sembrano corretti. Provo di nuovo con vector_search.",
      "tool_name": "vector_search",
      "tool_args": {"query": "agent loop debugging", "top_k": 5},
      "tool_result": "5 risultati trovati",
      "state_hash": "a3f9e2c1d84b",
      "similarity_to_prev": 0.91,
      "deadlock_events": [
        {
          "pattern": "TOOL_OSCILLATION",
          "iteration": 6,
          "evidence": "State hash duplicato: a3f9e2c1d84b già visto a iter 1",
          "severity": "critical"
        }
      ]
    }
  ]
}

Il campo similarity_to_prev a 0.91 all'iterazione 6 è il segnale chiave: l'agent sta producendo lo stesso pensiero con le stesse parole da quattro iterazioni. Il state_hash identico conferma che non si tratta di una coincidenza semantica, ma di un hard loop esatto.

Recovery Strategies: Tabella Completa

Rilevare il deadlock è metà del problema. L'altra metà è uscirne senza perdere il lavoro parziale dell'agent e senza richiedere sempre intervento umano. Queste sono le strategie di recovery che applico, in ordine crescente di invasività.

La regola che seguo: prova sempre L1, attendi il prossimo step, se il pattern ricompare applica L2. Se dopo L2 il pattern ricompare ancora una volta, valuta L3 solo se l'azione ha conseguenze irreversibili (scrittura su database, invio email, pagamenti). Per task read-only, termina con stop_reason="deadlock" e restituisci il partial_context.

Caso Reale: Agent Loop nel ml-service con Max 10 Iterazioni

Nel ml-service del mio portfolio, l'agent RAG principale usa il guard con limite di 10 iterazioni. Il caso più comune che ho osservato in produzione è la tool oscillation tra vector_search e rerank_results: l'agent chiama vector_search, i risultati non soddisfano la soglia di relevance, chiama rerank, i risultati cambiano leggermente ma non abbastanza, chiama di nuovo vector_search con la stessa query.

La mitigazione che ha funzionato: dopo il secondo vector_search con la stessa query, il guard inietta automaticamente nel contesto: "I risultati di vector_search per questa query sono stati già valutati. Usa i migliori disponibili o riformula l'obiettivo del task." Questo sblocca l'agent nel 90% dei casi senza intervento umano.

Per il context rot, che emerge soprattutto in sessioni di conversazione lunga (oltre 8 scambi), uso un summarization trigger automatico: quando il contesto supera 6.000 token, il sistema comprime gli ultimi 4 messaggi in un summary di 200 token e re-inietta l'obiettivo originale. Il costo extra è circa 400 token per summarization, ma evita di perdere l'agent su sessioni lunghe.

Come Calibrare le Soglie di Rilevamento

Le soglie di default sono un punto di partenza, non valori universali. La calibrazione corretta dipende dal tipo di agent e dalla complessità dei task. Ecco come approcciare la calibrazione:

Per lo state hash: inizia con finestra di 10. Se hai falsi positivi (loop rilevati su task legittimamente ripetitivi come paginate-and-collect), aumenta la finestra a 15 e aggiungi un contatore di pagine nello stato per differenziare gli step.

Per la Jaccard similarity: soglia 0.85 funziona bene per testo generico. Per agenti che producono output molto strutturati (JSON, SQL), abbassa a 0.75 - i template rendono il testo simile per natura. Per agenti su dominio creativo, alza a 0.90 per evitare falsi positivi su variazioni stilistiche legittime.

Per max_iterations: misura il percentile 95 delle iterazioni su task completati con successo nella tua distribuzione reale, poi imposta il limite a 1.5x quel valore. Nel mio caso, il P95 era 6 iterazioni, quindi il limite a 10 lascia margine senza essere troppo permissivo.

FAQ

Cos'è un agent loop deadlock?

Un agent loop deadlock è una condizione in cui un LLM agent entra in un ciclo di esecuzione ripetitiva senza fare progressi reali verso l'obiettivo. L'agent consuma token e risorse continuando a eseguire le stesse azioni o pensieri senza uscire dalla spirale. È distinto da un task difficile che richiede molte iterazioni: in quel caso ogni step aggiunge informazione nuova; in un deadlock no.

Qual è la differenza tra tool oscillation e hard loop?

L'hard loop chiama lo stesso tool con argomenti identici in step consecutivi. La tool oscillation è più sottile: il tool viene chiamato, poi un secondo tool ne annulla l'effetto, poi il primo viene richiamato. Il risultato finale è lo stesso, ma il pattern A→B→A→B è più difficile da rilevare con un semplice hash dello step corrente - richiede una finestra di storia.

Come si rileva automaticamente la reflection paralysis?

Si monitora il rapporto tra token spesi in output di tipo "thought/reflection" rispetto ai token spesi in tool call o azioni concrete. Se oltre il 50% dei token di un'iterazione sono in riflessione meta-cognitiva senza azioni risultanti, e questo si ripete per 2+ iterazioni consecutive, il pattern è classificato come reflection paralysis. Il RepetitionDetector con soglia 0.85 cattura anche la componente semantica.

Qual è il valore sicuro per max_iterations in produzione?

Dipende dalla complessità del task. Per task semplici come lookup o Q&A bastano 5-10 iterazioni. Per task complessi come code generation o ricerca multi-step, 15-25. Il metodo più robusto è misurare il P95 delle iterazioni su task completati con successo nella tua distribuzione reale, poi impostare il limite a 1.5x quel valore. Non usare mai un default generico di libreria senza averlo calibrato sul tuo workload specifico.

Il context rot si può prevenire senza summarization forzata?

Parzialmente. Mantenere lo "objective anchor" esplicito come messaggio fisso all'inizio del contesto (non solo nel system prompt) aiuta significativamente. L'anchor deve contenere: obiettivo del task, criteri di successo misurabili, tool autorizzati. La summarization periodica rimane comunque la mitigazione più robusta per sessioni che superano i 6.000 token di contesto accumulato.

Come distinguo un agent in loop da uno che lavora su un task difficile?

Il segnale chiave è la novità: un agent su task difficili aggiunge nuovi fatti, nuove chiamate a tool diversi, o riduce il problema ad ogni step. Un agent in loop ripete le stesse azioni o produce output semanticamente identici. Misura la distanza embedding (o Jaccard similarity) tra output consecutivi: se rimane alta per 3+ step, è loop. Se decresce progressivamente, è lavoro genuino su problema difficile.

L'escalation a umano è sempre la soluzione giusta per un loop?

No. L'escalation è costosa in latenza e risorse umane. Prima si devono esaurire le strategie automatiche: L1 (state latch, objective anchor, force action), L2 (forced summarization, context reset, temperatura ridotta). L'escalation va usata quando il loop persiste dopo almeno due tentativi automatici e l'azione ha conseguenze irreversibili (scrittura, pagamenti, comunicazioni). Per task read-only, termina il task con partial_context e lascia che il chiamante decida.

Conclusione

I loop negli agent LLM non sono bug casuali: sono pattern strutturali con cause precise e signature rilevabili. I sei pattern che ho descritto - tool oscillation, context rot, reflection paralysis, tool hallucination, guardrail loop, retry storm - coprono la grande maggioranza dei deadlock che incontrerai in produzione. Ognuno ha un fingerprint distinto nel trace e una recovery strategy specifica, dall'automatica (L1) all'escalazione umana (L3).

Gli strumenti chiave da implementare subito sono tre: il CycleDetector con state hashing per loop esatti, il RepetitionDetector con n-gram semantici per loop soft, e il MaxIterationGuard come safety net finale. Il trace recorder strutturato ti permette di fare post-mortem deterministici senza dover ricostruire la sessione a memoria.

Prossimo episodio: Tool design per agent - naming, schema, error handling che riducono errori del 60%. Vedremo come scrivere tool spec che guidano l'agent invece di confonderlo, e come il design del tool previene attivamente la tool hallucination e il guardrail loop che abbiamo analizzato qui.