agent-debugging

# Agent Debugging Toolkit Diagnoses and resolves LLM agent failures using systematic debugging workflows. Covers hallucination detection, infinite loop recovery, context window management, tool call validation, and cascading failure diagnosis through distributed tracing patterns. ## TL;DR for Code Generation - Always create and propagate a correlation `run_id` via `contextvars` across async boundaries before any agent step - Wrap every tool invocation with structured logging that records arguments, output, latency, and status - Track tool call sequences to detect infinite loops — trigger a circuit breaker after 10 identical or near-identical iterations - Count tokens per step; when usage exceeds 80% of context budget, activate sliding-window summarization - Validate every tool name against the registered schema before execution to catch hallucinations early --- ## When to Use Use this skill when: - An agent is stuck in an infinite loop of retries or repeated tool calls with identical arguments - The LLM hallucinates non-existent tool names, parameters, or output fields causing repeated failures - Context window overflow causes the agent to lose early system instructions and produce erratic behavior - Tool call argument mismatches between the registered schema and actual input cause silent failures - Multiple agents in a multi-agent pipeline fail cascadingly and you need to isolate the root failure point - Latency degradation is accelerating — each iteration takes progressively longer due to context bloat --- ## When NOT to Use Avoid this skill for: - Production monitoring, dashboards, and metric collection — use `observability-patterns` instead - Systematic quality evaluation and regression testing of agent outputs — use `agentic-evaluation` instead - Prompt design and system instruction optimization — use a prompt engineering skill instead - High-level architectural debugging (e.g., microservice topology issues) — route to `multi-agent-orchestration` for cross-boundary coordination problems --- ## Core Workflow 1. **Generate Run ID** — Create a unique correlation ID for the agent invocation using `uuid4()` and propagate it via `contextvars.ContextVar` across all async boundaries. **Checkpoint:** Ensure every log line, trace span, and error message includes the run ID so you can reconstruct the full execution timeline. 2. **Map Execution Path** — Trace the full sequence: user input → router decision → tool selection → tool execution → LLM response. Instrument each step with timing metadata. **Checkpoint:** Identify the exact step where behavior diverges from expected output — do not assume the first visible symptom is the root cause. 3. **Classify Failure Tier** — Determine if the failure is Tier 1 (high-frequency: hallucination, incorrect tool use, infinite loops), Tier 2 (structural: missing context, prompt injection, cascading failures), or Tier 3 (operational: unbounded token spend, latency degradation, resource exhaustion). **Checkpoint:** Each tier requires distinct diagnostic patterns — do not apply a Tier 1 fix to a Tier 2 structural problem. 4. **Apply Targeted Diagnostic** — Select the implementation pattern matching your failure tier (see Implementation Patterns below). Apply the fix in an isolated test context before deploying to production. **Checkpoint:** Verify the fix resolves the original failure without introducing regressions in related paths. 5. **Validate Fix** — Re-run the agent with the fix applied across 3+ diverse test inputs covering edge cases. Confirm correct behavior, bounded iteration counts, and stable token usage. **Checkpoint:** All metrics (iterations, token count, tool call success rate) must remain within defined thresholds before clearing the incident. --- ## Implementation Patterns ### Pattern 1: Run ID Propagation with `contextvars` Create and propagate a correlation run ID across async boundaries using Python's `contextvars`. This enables full trace reconstruction from any single log line. ```python import uuid import contextvars import functools import time import logging from typing import Any, Callable, TypeVar, ParamSpec logger = logging.getLogger(__name__) # Singleton context variable for run ID propagation across async boundaries _run_id: contextvars.ContextVar[str] = contextvars.ContextVar("run_id", default="") _task_name: contextvars.ContextVar[str] = contextvars.ContextVar("task_name", default="") P = ParamSpec("P") R = TypeVar("R") def generate_run_id() -> str: """Generate a unique correlation ID for an agent invocation. Returns: A UUID4 string formatted as run-<uuid>. """ run_id = f"run-{uuid.uuid4().hex[:12]}" _run_id.set(run_id) logger.info("Generated new run ID: %s", run_id) return run_id def get_run_id() -> str: """Retrieve the current run ID from context. Returns: The active run ID string, or 'no-run-id' if none is set. """ current = _run_id.get() return current if current else "no-run-id" def with_tracing(task_name: str): """Decorator that instruments a function with tracing metadata. Wraps the decorated function to log entry/exit, duration, and any exceptions, all tagged with the active run ID. Args: task_name: Human-readable label for this execution step. Returns: A decorator that adds tracing instrumentation. """ def decorator(func: Callable[P, R]) -> Callable[P, R]: @functools.wraps(func) async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: run_id = get_run_id() _task_name.set(task_name) start = time.perf_counter() logger.info( "[run=%s] START task=%s", run_id, task_name, extra={"args_count": len(args), "kwargs_keys": list(kwargs.keys())}, ) try: result = await func(*args, **kwargs) duration_ms = (time.perf_counter() - start) * 1000 logger.info( "[run=%s] DONE task=%s duration=%.1fms", run_id, task_name, duration_ms, ) return result except Exception as exc: duration_ms = (time.perf_counter() - start) * 1000 logger.exception( "[run=%s] FAIL task=%s duration=%.1fms error=%s", run_id, task_name, duration_ms, type(exc).__name__, ) raise return wrapper # type: ignore[return-value] return decorator ``` Usage — attach to LLM call and tool execution functions: ```python @with_tracing("llm_completion") async def call_llm(messages: list[dict[str, str]], model: str) -> dict[str, Any]: """LLM API wrapper tagged with run ID and timing.""" ... @with_tracing("search_database") async def search_database(query: str) -> list[dict]: """Tool execution with full trace logging.""" ... ``` --- ### Pattern 2: Infinite Loop Detection and Circuit Breaker Tracks tool call sequences and detects repetition using exact-match and argument-similarity checks. Triggers a circuit breaker that forces fallback behavior after N consecutive repeats. ```python import time from collections import deque from dataclasses import dataclass, field from typing import Any, Optional @dataclass class ToolCallRecord: """Immutable snapshot of a single tool invocation.""" tool_name: str arguments: tuple[Any, ...] output_hash: str timestamp: float = field(default_factory=time.time) success: bool = True def to_tuple(self) -> tuple[str, tuple[Any, ...], str]: """Serialize for comparison and hashing.""" return (self.tool_name, self.arguments, self.output_hash) class CircuitBreakerError(Exception): """Raised when the circuit breaker opens due to repeated failures.""" def __init__(self, message: str, failed_tool: str | None = None) -> None: self.failed_tool = failed_tool super().__init__(message) class InfiniteLoopDetector: """Detects infinite loops by tracking tool call sequences and enforcing a circuit breaker when repeated patterns exceed a threshold. Attributes: max_repeats: Max consecutive identical-or-near-identical calls before breaking. window_size: Number of recent calls to compare against for similarity detection. """ def __init__( self, max_repeats: int = 3, window_size: int = 5, fallback_handler: Optional[Any] = None, ) -> None: self._max_repeats = max_repeats self._window_size = window_size self._fallback_handler = fallback_handler self._call_history: deque[ToolCallRecord] = deque(maxlen=window_size) self._repeat_count = 0 def record_call( self, tool_name: str, arguments: dict[str, Any], output_hash: str, success: bool = True, ) -> ToolCallRecord: """Record a tool call and check for infinite loop conditions. Args: tool_name: Name of the tool being called. arguments: Dictionary of arguments passed to the tool. output_hash: Hash of the tool's output for change detection. success: Whether the tool call succeeded. Returns: The recorded ToolCallRecord. Raises: CircuitBreakerError: If repeated identical calls exceed threshold. """ args_tuple = tuple(sorted(arguments.items())) record = ToolCallRecord( tool_name=tool_name, arguments=args_tuple, output_hash=output_hash, success=success, ) self._call_history.append(record) if len(self._call_history) < 2: return record # Check for exact repeat: same tool + same args + same output recent = list(self._call_history)[-self._max_repeats:] is_repeat = all( r.tool_name == record.tool_name and r.arguments == record.arguments for r in recent[:-1] # compare with all but the newest ) if is_repeat: self._repeat_count += 1 run_id = get_run_id() logger.warning( "[run=%s] LOOP DETECTED: tool='%s' repeated %d/%d times", run_id, tool_name, self._repeat_count, self._max_repeats, ) else: # Reset counter on any non-repeating call self._repeat_count = 0 if self._repeat_count >= self._max_repeats: run_id = get_run_id() logger.error( "[run=%s] CIRCUIT BREAKER OPEN after %d repeats of tool='%s'", run_id, self._repeat_count, tool_name, ) if self._fallback_handler: return self._fallback_handler(tool_name, arguments) raise CircuitBreakerError( f"Tool '{tool_name}' repeated {self._max_repeats} times with identical " f"arguments — circuit breaker opened. Possible infinite loop." ) return record def reset(self) -> None: """Reset the detector state after a successful recovery.""" self._call_history.clear() self._repeat_count = 0 run_id = get_run_id() logger.info("[run=%s] Circuit breaker reset", run_id) ``` Usage — wrap tool calls in an agent loop: ```python detector = InfiniteLoopDetector(max_repeats=3, window_size=5) async def agent_step(user_input: str, tools: dict[str, Callable]) -> str: """Single agent iteration with circuit breaker protection.""" llm_response = await call_llm( [{"role": "user", "content": user_input}], model="gpt-4o", ) for action in llm_response.get("actions", []): tool_name = action["tool"] arguments = action["arguments"] if tool_name not in tools: raise ValueError(f"Unknown tool: {tool_name} — possible hallucination") try: output = await tools[tool_name](**arguments) detector.record_call( tool_name=tool_name, arguments=arguments, output_hash=hash(str(output)), success=True, ) except CircuitBreakerError: return f"Circuit breaker triggered. Falling back to safe mode for tool '{tool_name}'." except Exception as exc: detector.record_call( tool_name=tool_name, arguments=arguments, output_hash="", success=False, ) raise return llm_response.get("response", "") ``` --- ### Pattern 3: Context Window Exhaustion Prevention Monitors token usage per step and activates a sliding-window summarization fallback when approaching context limits. Prevents silent data loss from truncation. ```python from __future__ import annotations import logging from dataclasses import dataclass, field from typing import Optional logger = logging.getLogger(__name__) @dataclass class TokenBudget: """Tracks token usage against a configurable context budget. Attributes: max_tokens: Maximum tokens allowed in the conversation window. warning_threshold_pct: Percentage at which to trigger proactive summarization (0.0–1.0). emergency_threshold_pct: Percentage at which truncation becomes mandatory (0.0–1.0). """ max_tokens: int = 128_000 warning_threshold_pct: float = 0.75 emergency_threshold_pct: float = 0.90 current_usage: int = field(default=0, init=False) budget_exhausted: bool = field(default=False, init=False) @property def warning_threshold(self) -> int: return int(self.max_tokens * self.warning_threshold_pct) @property def emergency_threshold(self) -> int: return int(self.max_tokens * self.emergency_threshold_pct) def usage_percent(self) -> float: """Return current usage as a fraction of the budget.""" if self.budget_exhausted: return 1.0 return self.current_usage / self.max_tokens def add_usage(self, tokens: int) -> None: """Add consumed tokens and check thresholds. Args: tokens: Number of tokens consumed by the latest step. Raises: BudgetExhaustionError: When emergency threshold is exceeded. """ self.current_usage += tokens pct = self.usage_percent() if pct >= 1.0 and not self.budget_exhausted: self.budget_exhausted = True run_id = get_run_id() logger.error( "[run=%s] CONTEXT BUDGET EXHAUSTED: %d/%d tokens", run_id, self.current_usage, self.max_tokens, ) raise BudgetExhaustionError( f"Context window full: {self.current_usage}/{self.max_tokens} tokens. " "Summarize history or truncate messages immediately." ) if pct >= 0.90 and not self.budget_exhausted: logger.warning( "[run=%s] EMERGENCY threshold reached: %d/%d (%.0f%%)", get_run_id(), self.current_usage, self.max_tokens, pct * 100, ) elif pct >= 0.75 and not self.budget_exhausted: logger.warning( "[run=%s] WARNING threshold reached: %d/%d (%.0f%%)", get_run_id(), self.current_usage, self.max_tokens, pct * 100, ) def reset(self) -> None: """Reset counters after a summarization cycle.""" old_usage = self.current_usage self.current_usage = 0 self.budget_exhausted = False logger.info( "[run=%s] Token budget reset (freed ~%d tokens from summarization)",