HinterBuild logoHinterBuild
Back to blog
AI Systems · 9 min read

Prevent Agent Loops & Runaway Tools: Production Safeguards

Learn prevent agent loops & runaway tools through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.

Muhammad Abdul Sami, author

Muhammad Abdul Sami

· 9 min read

  • AI Agents
  • Tool Calling
  • LangGraph
  • Architecture

Runaway AI agents can exhaust API quotas, rack up costs, and crash systems in minutes. Production agents need circuit breakers, iteration limits, loop detection, and resource monitoring to prevent infinite loops and runaway behavior. This guide covers safeguards from production systems processing millions of agent executions.

Key Takeaways:

  • Treat Prevent Agent Loops & Runaway Tools as a system with an explicit input and output contract.
  • Benchmark a representative baseline before choosing an optimization.
  • Bound retries, queues, concurrency, and total request deadlines.
  • Roll out through offline replay, shadow traffic, and a measurable canary.
  • Keep rollback simple and attach version identifiers to every decision.

Table of Contents:

Why Agents Loop

Agents loop when they fail to make progress toward goals but continue trying the same failed approaches. Common causes:

1. Tool Call Failures

python
for _ in range(infinite):
    result = call_api()  # Always fails
    # No progress made, but agent tries again

Cause: Agent doesn't recognize failure pattern

2. Ambiguous Goals

python
# Vague objective leads to wandering
task = "Make the system better"
# Agent tries random improvements indefinitely

Cause: No clear completion criteria

3. Circular Dependencies

python
# Step 1 needs Step 2, Step 2 needs Step 1
plan = [
    {"step": 1, "needs": [2]},
    {"step": 2, "needs": [1]}
]

Cause: Invalid plan structure

4. Hallucinated Progress

python
# Agent believes it's making progress when it's not
iteration 1: "Making progress..."
iteration 2: "Making progress..."  # Same state
iteration 100: "Making progress..."  # Still same state

Cause: LLM overconfidence

5. Forgotten Context

python
# Context window overflow loses critical info
iteration 50: Agent forgets it already tried this approach
iteration 51: Tries same approach again

Cause: Context management failures

These failures occur in production AI agent systems without proper safeguards.

Iteration Limits

Hard limits prevent infinite execution:

python
from typing import TypedDict, Annotated
import operator

class SafeAgentState(TypedDict):
    """State with iteration tracking."""
    messages: Annotated[list, operator.add]
    iteration: int
    max_iterations: int
    status: str

MAX_ITERATIONS = 20  # Production default

async def agent_with_limit(state: SafeAgentState):
    """Agent with hard iteration limit."""
    
    # Check iteration limit
    if state.get("iteration", 0) >= state.get("max_iterations", MAX_ITERATIONS):
        logger.warning(
            "Max iterations reached",
            extra={
                "iterations": state["iteration"],
                "last_action": state["messages"][-1] if state["messages"] else None
            }
        )
        
        return {
            "status": "max_iterations_reached",
            "messages": [{
                "role": "assistant",
                "content": "I've reached the maximum number of attempts. Please provide more specific guidance or break this into smaller tasks."
            }]
        }
    
    # Normal agent logic
    response = await llm.ainvoke(state["messages"])
    
    return {
        "messages": [response],
        "iteration": state.get("iteration", 0) + 1
    }

Dynamic Iteration Limits

python
def calculate_iteration_limit(task_complexity: str) -> int:
    """Set limits based on task complexity."""
    
    limits = {
        "simple": 5,      # Single-step tasks
        "medium": 15,     # Multi-step tasks
        "complex": 30,    # Research/planning tasks
        "exploratory": 50 # Open-ended tasks
    }
    
    return limits.get(task_complexity, 20)

# Usage
complexity = classify_task_complexity(task)
state = {
    "max_iterations": calculate_iteration_limit(complexity),
    "iteration": 0
}

Soft Limits with Warnings

python
SOFT_LIMIT = 10
HARD_LIMIT = 20

async def agent_with_warnings(state: SafeAgentState):
    """Warn before hitting hard limit."""
    
    iteration = state.get("iteration", 0)
    
    # Soft limit warning
    if iteration == SOFT_LIMIT:
        logger.warning("Approaching iteration limit")
        
        return {
            "messages": [{
                "role": "system",
                "content": f"Warning: You've used {SOFT_LIMIT}/{HARD_LIMIT} iterations. Focus on completing the task efficiently."
            }],
            "iteration": iteration + 1
        }
    
    # Hard limit enforcement
    if iteration >= HARD_LIMIT:
        raise MaxIterationsError(f"Exceeded {HARD_LIMIT} iterations")
    
    # Normal execution
    return await execute_agent_step(state)

Loop Detection

Detect when agents repeat actions:

Action History Tracking

python
from collections import deque
from typing import List, Dict

class LoopDetector:
    """Detect repeated action patterns."""
    
    def __init__(self, window_size: int = 5, threshold: int = 3):
        self.window_size = window_size
        self.threshold = threshold
        self.action_history = deque(maxlen=window_size * 2)
    
    def add_action(self, action: str, args: Dict) -> bool:
        """
        Add action and check for loops.
        Returns True if loop detected.
        """
        # Create action signature
        action_sig = self._create_signature(action, args)
        self.action_history.append(action_sig)
        
        # Check for repeated patterns
        if len(self.action_history) < self.threshold:
            return False
        
        # Count recent occurrences
        recent = list(self.action_history)[-self.window_size:]
        count = recent.count(action_sig)
        
        if count >= self.threshold:
            logger.warning(
                "Loop detected",
                extra={
                    "action": action,
                    "repetitions": count,
                    "history": recent
                }
            )
            return True
        
        return False
    
    def _create_signature(self, action: str, args: Dict) -> str:
        """Create comparable action signature."""
        # Normalize args for comparison
        sorted_args = sorted(args.items())
        return f"{action}:{json.dumps(sorted_args)}"
    
    def get_pattern(self) -> Optional[str]:
        """Detect repeating sequence patterns."""
        if len(self.action_history) < 6:
            return None
        
        recent = list(self.action_history)
        
        # Check for alternating pattern (A, B, A, B, A, B)
        if len(set(recent[-6:])) == 2:
            if recent[-1] == recent[-3] == recent[-5]:
                return "alternating"
        
        # Check for cycle pattern (A, B, C, A, B, C)
        for cycle_length in [2, 3, 4]:
            if self._has_cycle(recent, cycle_length):
                return f"cycle_{cycle_length}"
        
        return None
    
    def _has_cycle(self, actions: List[str], length: int) -> bool:
        """Check for repeating cycle of given length."""
        if len(actions) < length * 2:
            return False
        
        recent = actions[-length * 2:]
        first_cycle = recent[:length]
        second_cycle = recent[length:]
        
        return first_cycle == second_cycle

# Usage
loop_detector = LoopDetector(window_size=5, threshold=3)

async def safe_agent_with_loop_detection(state: SafeAgentState):
    """Agent with loop detection."""
    
    # Get next action
    action, args = await get_next_action(state)
    
    # Check for loops
    if loop_detector.add_action(action, args):
        # Loop detected - break out
        pattern = loop_detector.get_pattern()
        
        logger.error(
            "Agent stuck in loop",
            extra={"pattern": pattern, "action": action}
        )
        
        return {
            "status": "loop_detected",
            "messages": [{
                "role": "system",
                "content": f"""Loop detected: You're repeating the same action.
                Pattern: {pattern}
                Try a completely different approach or ask for human help."""
            }]
        }
    
    # Execute action
    result = await execute_action(action, args)
    
    return {"messages": [result]}

State-Based Loop Detection

python
class StateLoopDetector:
    """Detect when agent state stops changing."""
    
    def __init__(self, window_size: int = 3):
        self.window_size = window_size
        self.state_history = deque(maxlen=window_size)
    
    def add_state(self, state: Dict) -> bool:
        """
        Add state and check if agent is making progress.
        Returns True if stuck (state not changing).
        """
        # Create state fingerprint
        fingerprint = self._create_fingerprint(state)
        self.state_history.append(fingerprint)
        
        # Check if recent states are identical
        if len(self.state_history) >= self.window_size:
            unique_states = len(set(self.state_history))
            
            if unique_states == 1:
                logger.warning("Agent state not changing - stuck")
                return True
        
        return False
    
    def _create_fingerprint(self, state: Dict) -> str:
        """Create comparable state fingerprint."""
        # Extract relevant state components
        key_state = {
            "tool_results": state.get("tool_results"),
            "current_goal": state.get("current_goal"),
            "completed_steps": len(state.get("completed_steps", []))
        }
        
        return json.dumps(key_state, sort_keys=True)

This integrates with reliable tool calling.

Circuit Breakers

Prevent cascading failures from external services:

python
from enum import Enum
from datetime import datetime, timedelta
from collections import defaultdict

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Blocking requests
    HALF_OPEN = "half_open"  # Testing recovery

class CircuitBreaker:
    """Prevent runaway tool calls to failing services."""
    
    def __init__(
        self,
        failure_threshold: int = 5,
        success_threshold: int = 2,
        timeout_seconds: int = 60
    ):
        self.failure_threshold = failure_threshold
        self.success_threshold = success_threshold
        self.timeout_seconds = timeout_seconds
        
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time: Optional[datetime] = None
        self.opened_at: Optional[datetime] = None
    
    async def call(self, func: Callable, *args, **kwargs):
        """Execute function with circuit breaker protection."""
        
        # Check if circuit is open
        if self.state == CircuitState.OPEN:
            elapsed = (datetime.now() - self.opened_at).total_seconds()
            
            if elapsed < self.timeout_seconds:
                raise CircuitBreakerOpenError(
                    f"Circuit breaker open. Try again in {self.timeout_seconds - elapsed:.0f}s"
                )
            else:
                # Timeout elapsed - move to half-open
                self.state = CircuitState.HALF_OPEN
                self.success_count = 0
                logger.info("Circuit breaker half-open - testing recovery")
        
        # Attempt call
        try:
            result = await func(*args, **kwargs)
            self._record_success()
            return result
        
        except Exception as e:
            self._record_failure()
            raise
    
    def _record_success(self):
        """Record successful call."""
        self.failure_count = 0
        
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            
            if self.success_count >= self.success_threshold:
                self.state = CircuitState.CLOSED
                logger.info("Circuit breaker closed - service recovered")
        
        elif self.state == CircuitState.OPEN:
            # Shouldn't reach here, but handle gracefully
            self.state = CircuitState.HALF_OPEN
    
    def _record_failure(self):
        """Record failed call."""
        self.failure_count += 1
        self.last_failure_time = datetime.now()
        
        if self.failure_count >= self.failure_threshold:
            if self.state != CircuitState.OPEN:
                self.state = CircuitState.OPEN
                self.opened_at = datetime.now()
                logger.error(
                    "Circuit breaker opened",
                    extra={"failures": self.failure_count}
                )

class CircuitBreakerOpenError(Exception):
    """Circuit breaker is open."""
    pass

# Usage
breakers = {
    "database": CircuitBreaker(failure_threshold=5, timeout_seconds=60),
    "external_api": CircuitBreaker(failure_threshold=3, timeout_seconds=120),
    "email_service": CircuitBreaker(failure_threshold=10, timeout_seconds=300)
}

async def protected_tool_call(tool_name: str, tool_func: Callable, args: Dict):
    """Call tool with circuit breaker protection."""
    
    breaker = breakers.get(tool_name)
    
    if breaker:
        try:
            return await breaker.call(tool_func, **args)
        except CircuitBreakerOpenError as e:
            logger.warning(f"Circuit breaker prevented call to {tool_name}")
            return {
                "success": False,
                "error": "Service temporarily unavailable",
                "retry_after": breaker.timeout_seconds
            }
    else:
        # No circuit breaker - direct call
        return await tool_func(**args)

Resource Monitoring

Monitor and limit resource consumption:

python
import psutil
import asyncio
from dataclasses import dataclass

@dataclass
class ResourceLimits:
    """Resource consumption limits."""
    max_memory_mb: int = 500
    max_cpu_percent: float = 80.0
    max_execution_time_seconds: int = 300
    max_tool_calls: int = 50
    max_llm_calls: int = 30
    max_tokens: int = 100000

class ResourceMonitor:
    """Monitor agent resource usage."""
    
    def __init__(self, limits: ResourceLimits):
        self.limits = limits
        self.start_time = datetime.now()
        self.tool_call_count = 0
        self.llm_call_count = 0
        self.token_count = 0
        self.process = psutil.Process()
    
    def check_limits(self) -> Optional[str]:
        """
        Check if any limits exceeded.
        Returns error message if limit exceeded, None otherwise.
        """
        # Memory check
        memory_mb = self.process.memory_info().rss / 1024 / 1024
        if memory_mb > self.limits.max_memory_mb:
            return f"Memory limit exceeded: {memory_mb:.0f}MB > {self.limits.max_memory_mb}MB"
        
        # CPU check
        cpu_percent = self.process.cpu_percent(interval=0.1)
        if cpu_percent > self.limits.max_cpu_percent:
            return f"CPU limit exceeded: {cpu_percent:.1f}% > {self.limits.max_cpu_percent}%"
        
        # Execution time check
        elapsed = (datetime.now() - self.start_time).total_seconds()
        if elapsed > self.limits.max_execution_time_seconds:
            return f"Execution time limit exceeded: {elapsed:.0f}s > {self.limits.max_execution_time_seconds}s"
        
        # Tool call limit
        if self.tool_call_count > self.limits.max_tool_calls:
            return f"Tool call limit exceeded: {self.tool_call_count} > {self.limits.max_tool_calls}"
        
        # LLM call limit
        if self.llm_call_count > self.limits.max_llm_calls:
            return f"LLM call limit exceeded: {self.llm_call_count} > {self.limits.max_llm_calls}"
        
        # Token limit
        if self.token_count > self.limits.max_tokens:
            return f"Token limit exceeded: {self.token_count} > {self.limits.max_tokens}"
        
        return None
    
    def record_tool_call(self):
        """Increment tool call counter."""
        self.tool_call_count += 1
    
    def record_llm_call(self, tokens: int):
        """Increment LLM call counter and token usage."""
        self.llm_call_count += 1
        self.token_count += tokens
    
    def get_stats(self) -> Dict:
        """Get current resource usage statistics."""
        return {
            "memory_mb": self.process.memory_info().rss / 1024 / 1024,
            "cpu_percent": self.process.cpu_percent(interval=0),
            "elapsed_seconds": (datetime.now() - self.start_time).total_seconds(),
            "tool_calls": self.tool_call_count,
            "llm_calls": self.llm_call_count,
            "tokens": self.token_count
        }

# Usage
monitor = ResourceMonitor(ResourceLimits(
    max_memory_mb=500,
    max_execution_time_seconds=300,
    max_tool_calls=50
))

async def monitored_agent(state: SafeAgentState):
    """Agent with resource monitoring."""
    
    # Check resource limits
    limit_error = monitor.check_limits()
    if limit_error:
        logger.error("Resource limit exceeded", extra={"error": limit_error})
        
        return {
            "status": "resource_limit_exceeded",
            "error": limit_error,
            "stats": monitor.get_stats()
        }
    
    # Record LLM call
    response = await llm.ainvoke(state["messages"])
    monitor.record_llm_call(tokens=count_tokens(response))
    
    # Execute tools
    for tool_call in extract_tool_calls(response):
        monitor.record_tool_call()
        result = await execute_tool(tool_call)
    
    return {"messages": [response]}

Timeout Strategies

Prevent indefinite execution:

python
import asyncio
from typing import Optional

async def execute_with_timeout(
    coro,
    timeout_seconds: int,
    timeout_message: str = "Operation timed out"
) -> Dict:
    """Execute coroutine with timeout."""
    
    try:
        result = await asyncio.wait_for(coro, timeout=timeout_seconds)
        return {"success": True, "result": result}
    
    except asyncio.TimeoutError:
        logger.warning(
            "Operation timeout",
            extra={"timeout": timeout_seconds}
        )
        return {
            "success": False,
            "error": timeout_message,
            "timeout_seconds": timeout_seconds
        }

# Hierarchical timeouts
async def agent_with_hierarchical_timeouts(task: str):
    """Agent with timeouts at multiple levels."""
    
    # Overall agent timeout: 5 minutes
    async def run_agent():
        # Tool-level timeout: 30 seconds each
        async def execute_tool_with_timeout(tool_call):
            return await execute_with_timeout(
                execute_tool(tool_call),
                timeout_seconds=30,
                timeout_message=f"Tool {tool_call['name']} timed out"
            )
        
        # LLM call timeout: 60 seconds
        async def llm_with_timeout(messages):
            return await execute_with_timeout(
                llm.ainvoke(messages),
                timeout_seconds=60,
                timeout_message="LLM call timed out"
            )
        
        # Agent execution loop
        for iteration in range(MAX_ITERATIONS):
            response = await llm_with_timeout(messages)
            
            if not response["success"]:
                return response  # LLM timeout
            
            tool_results = await asyncio.gather(*[
                execute_tool_with_timeout(call)
                for call in extract_tool_calls(response["result"])
            ])
            
            # Check for tool timeouts
            if any(not r["success"] for r in tool_results):
                return {"success": False, "error": "Tool execution failed"}
        
        return {"success": True}
    
    # Overall timeout
    return await execute_with_timeout(
        run_agent(),
        timeout_seconds=300,
        timeout_message="Agent execution timed out after 5 minutes"
    )

Cost Guards

Prevent runaway costs:

python
class CostGuard:
    """Monitor and limit agent costs."""
    
    def __init__(self, max_cost_usd: float = 10.0):
        self.max_cost_usd = max_cost_usd
        self.current_cost = 0.0
        
        # Token costs (GPT-4 pricing)
        self.cost_per_1k_input = 0.03
        self.cost_per_1k_output = 0.06
    
    def record_llm_call(self, input_tokens: int, output_tokens: int):
        """Record LLM call cost."""
        cost = (
            (input_tokens / 1000) * self.cost_per_1k_input +
            (output_tokens / 1000) * self.cost_per_1k_output
        )
        
        self.current_cost += cost
        
        logger.info(
            "LLM call cost",
            extra={
                "call_cost": f"${cost:.4f}",
                "total_cost": f"${self.current_cost:.4f}"
            }
        )
    
    def check_budget(self) -> bool:
        """Check if budget exceeded."""
        if self.current_cost >= self.max_cost_usd:
            logger.error(
                "Cost limit exceeded",
                extra={
                    "current_cost": f"${self.current_cost:.2f}",
                    "max_cost": f"${self.max_cost_usd:.2f}"
                }
            )
            return False
        
        # Warn at 80% budget
        if self.current_cost >= 0.8 * self.max_cost_usd:
            logger.warning(
                "Approaching cost limit",
                extra={
                    "current_cost": f"${self.current_cost:.2f}",
                    "remaining": f"${self.max_cost_usd - self.current_cost:.2f}"
                }
            )
        
        return True
    
    def get_remaining_budget(self) -> float:
        """Get remaining budget."""
        return max(0, self.max_cost_usd - self.current_cost)

# Usage
cost_guard = CostGuard(max_cost_usd=5.0)

async def cost_aware_agent(state: SafeAgentState):
    """Agent with cost monitoring."""
    
    # Check budget before LLM call
    if not cost_guard.check_budget():
        return {
            "status": "budget_exceeded",
            "error": f"Exceeded budget of ${cost_guard.max_cost_usd}",
            "total_cost": f"${cost_guard.current_cost:.2f}"
        }
    
    # Make LLM call
    response = await llm.ainvoke(state["messages"])
    
    # Record cost
    cost_guard.record_llm_call(
        input_tokens=count_tokens(state["messages"]),
        output_tokens=count_tokens(response)
    )
    
    return {"messages": [response]}

Graceful Degradation

Degrade gracefully when limits approached:

python
class GracefulDegradation:
    """Adjust agent behavior as limits approached."""
    
    def __init__(self, monitor: ResourceMonitor):
        self.monitor = monitor
    
    def get_adjustment_strategy(self) -> Dict:
        """Determine how to adjust agent behavior."""
        stats = self.monitor.get_stats()
        limits = self.monitor.limits
        
        # Calculate usage percentages
        tool_usage = stats["tool_calls"] / limits.max_tool_calls
        llm_usage = stats["llm_calls"] / limits.max_llm_calls
        time_usage = stats["elapsed_seconds"] / limits.max_execution_time_seconds
        
        strategy = {
            "reduce_tool_calls": tool_usage > 0.7,
            "simplify_reasoning": llm_usage > 0.7,
            "prioritize_completion": time_usage > 0.7,
            "switch_to_cheaper_model": llm_usage > 0.8
        }
        
        return strategy

async def adaptive_agent(state: SafeAgentState):
    """Agent that adapts behavior based on resource usage."""
    
    strategy = degradation.get_adjustment_strategy()
    
    # Build system message with constraints
    constraints = []
    
    if strategy["reduce_tool_calls"]:
        constraints.append("Minimize tool calls - only essential operations")
    
    if strategy["simplify_reasoning"]:
        constraints.append("Be concise - avoid lengthy reasoning")
    
    if strategy["prioritize_completion"]:
        constraints.append("Focus on completing core task - skip optional steps")
    
    if strategy["switch_to_cheaper_model"]:
        # Use faster, cheaper model
        model = ChatOpenAI(model="gpt-3.5-turbo")
    else:
        model = llm
    
    # Add constraints to system message
    if constraints:
        messages = state["messages"].copy()
        messages.insert(0, {
            "role": "system",
            "content": f"RESOURCE CONSTRAINTS:\n" + "\n".join(f"- {c}" for c in constraints)
        })
    else:
        messages = state["messages"]
    
    response = await model.ainvoke(messages)
    return {"messages": [response]}

LangGraph Implementation

Comprehensive safeguards in LangGraph:

python
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class SafeLangGraphState(TypedDict):
    """State with all safety tracking."""
    messages: Annotated[list, operator.add]
    iteration: int
    max_iterations: int
    tool_calls: int
    cost_usd: float
    status: str

def create_safe_agent():
    """Production agent with all safeguards."""
    
    workflow = StateGraph(SafeLangGraphState)
    
    # Initialize monitors
    loop_detector = LoopDetector(window_size=5, threshold=3)
    monitor = ResourceMonitor(ResourceLimits())
    cost_guard = CostGuard(max_cost_usd=5.0)
    
    def agent_node(state: SafeLangGraphState):
        """Agent with comprehensive safety checks."""
        
        # Check iteration limit
        if state["iteration"] >= state["max_iterations"]:
            return {
                "status": "max_iterations",
                "messages": [{
                    "role": "assistant",
                    "content": "Maximum iterations reached."
                }]
            }
        
        # Check resource limits
        limit_error = monitor.check_limits()
        if limit_error:
            return {"status": "resource_limit", "error": limit_error}
        
        # Check budget
        if not cost_guard.check_budget():
            return {"status": "budget_exceeded"}
        
        # Make LLM call
        response = await llm.ainvoke(state["messages"])
        
        # Record cost
        cost_guard.record_llm_call(
            input_tokens=count_tokens(state["messages"]),
            output_tokens=count_tokens(response)
        )
        monitor.record_llm_call(count_tokens(response))
        
        return {
            "messages": [response],
            "iteration": state["iteration"] + 1,
            "cost_usd": cost_guard.current_cost
        }
    
    def tool_node(state: SafeLangGraphState):
        """Tool execution with safety checks."""
        
        tool_calls = extract_tool_calls(state["messages"][-1])
        
        results = []
        for call in tool_calls:
            # Check for loops
            if loop_detector.add_action(call["name"], call["args"]):
                return {
                    "status": "loop_detected",
                    "messages": [{
                        "role": "system",
                        "content": "Loop detected - try different approach"
                    }]
                }
            
            # Execute with circuit breaker
            result = await protected_tool_call(
                call["name"],
                get_tool(call["name"]),
                call["args"]
            )
            
            monitor.record_tool_call()
            results.append(result)
        
        return {
            "messages": results,
            "tool_calls": state["tool_calls"] + len(tool_calls)
        }
    
    def should_continue(state: SafeLangGraphState) -> str:
        """Routing with safety checks."""
        
        # Check status
        if state.get("status") in ["max_iterations", "resource_limit", "budget_exceeded", "loop_detected"]:
            return "end"
        
        # Normal routing
        last_message = state["messages"][-1]
        if has_tool_calls(last_message):
            return "tools"
        
        return "end"
    
    workflow.add_node("agent", agent_node)
    workflow.add_node("tools", tool_node)
    
    workflow.add_conditional_edges(
        "agent",
        should_continue,
        {"tools": "tools", "end": END}
    )
    
    workflow.add_edge("tools", "agent")
    workflow.set_entry_point("agent")
    
    return workflow.compile()

# Usage
safe_agent = create_safe_agent()

result = safe_agent.invoke({
    "messages": [{"role": "user", "content": task}],
    "iteration": 0,
    "max_iterations": 20,
    "tool_calls": 0,
    "cost_usd": 0.0
})

This integrates with stateful checkpoints.

Testing Loop Prevention

python
import pytest

@pytest.mark.asyncio
async def test_iteration_limit_enforced():
    """Test agent stops at iteration limit."""
    state = {
        "messages": [],
        "iteration": 0,
        "max_iterations": 5
    }
    
    # Run agent until it stops
    while state["iteration"] < 10:  # Try more than limit
        state = await agent_with_limit(state)
        
        if state.get("status") == "max_iterations_reached":
            break
    
    # Should stop at limit
    assert state["iteration"] == 5
    assert state["status"] == "max_iterations_reached"

@pytest.mark.asyncio
async def test_loop_detection_triggers():
    """Test loop detector catches repeated actions."""
    detector = LoopDetector(window_size=5, threshold=3)
    
    # Repeat same action
    for _ in range(3):
        loop_detected = detector.add_action("search", {"query": "test"})
    
    # Should detect loop on 3rd repetition
    assert loop_detected is True

@pytest.mark.asyncio
async def test_circuit_breaker_opens():
    """Test circuit breaker opens after failures."""
    breaker = CircuitBreaker(failure_threshold=3)
    
    async def failing_function():
        raise Exception("Service unavailable")
    
    # Trigger failures
    for _ in range(3):
        with pytest.raises(Exception):
            await breaker.call(failing_function)
    
    # Circuit should be open
    assert breaker.state == CircuitState.OPEN
    
    # Subsequent calls should fail immediately
    with pytest.raises(CircuitBreakerOpenError):
        await breaker.call(failing_function)

@pytest.mark.asyncio
async def test_resource_limits_enforced():
    """Test resource monitor enforces limits."""
    monitor = ResourceMonitor(ResourceLimits(max_tool_calls=10))
    
    # Record tool calls
    for _ in range(10):
        monitor.record_tool_call()
    
    # Should not exceed limit
    error = monitor.check_limits()
    assert error is None
    
    # One more should trigger limit
    monitor.record_tool_call()
    error = monitor.check_limits()
    assert error is not None
    assert "Tool call limit exceeded" in error

@pytest.mark.asyncio
async def test_cost_guard_prevents_overruns():
    """Test cost guard stops execution at budget."""
    guard = CostGuard(max_cost_usd=1.0)
    
    # Simulate expensive calls
    for _ in range(100):
        guard.record_llm_call(input_tokens=1000, output_tokens=1000)
    
    # Should exceed budget
    assert not guard.check_budget()
    assert guard.current_cost >= 1.0

Learn comprehensive agent testing strategies.

Monitoring and Alerts

python
from prometheus_client import Counter, Histogram, Gauge

# Metrics
agent_iterations = Histogram(
    'agent_iterations_total',
    'Number of iterations per agent execution',
    buckets=[1, 5, 10, 20, 50, 100]
)

agent_loops_detected = Counter(
    'agent_loops_detected_total',
    'Number of loops detected',
    ['pattern_type']
)

circuit_breaker_trips = Counter(
    'circuit_breaker_trips_total',
    'Number of times circuit breakers opened',
    ['service']
)

agent_resource_usage = Gauge(
    'agent_resource_usage',
    'Current resource usage',
    ['resource_type']
)

async def monitored_agent_execution(task: str):
    """Execute agent with comprehensive monitoring."""
    
    monitor = ResourceMonitor(ResourceLimits())
    start_time = datetime.now()
    
    try:
        result = await safe_agent.invoke({"task": task})
        
        # Record metrics
        agent_iterations.observe(result.get("iteration", 0))
        
        # Record final resource usage
        stats = monitor.get_stats()
        agent_resource_usage.labels(resource_type="memory_mb").set(stats["memory_mb"])
        agent_resource_usage.labels(resource_type="tool_calls").set(stats["tool_calls"])
        
        # Alert on high iterations
        if result.get("iteration", 0) > 15:
            await send_alert(
                title="High agent iteration count",
                details=f"Agent used {result['iteration']} iterations for task: {task}"
            )
        
        return result
    
    except Exception as e:
        logger.error("Agent execution failed", extra={"error": str(e)})
        raise
    
    finally:
        duration = (datetime.now() - start_time).total_seconds()
        logger.info(
            "Agent execution completed",
            extra={
                "duration_seconds": duration,
                "stats": monitor.get_stats()
            }
        )

See agent observability guide.

Prevent Agent Loops & Runaway Tools Decision Table

DecisionPrefer the simpler path whenAdd operational complexity when
ArchitectureOne component can own the contract and stateIndependent scaling or fault isolation is required
RolloutOffline replay covers the meaningful casesLive behavior requires shadow traffic and a canary
RecoveryA failed operation is safe to repeatPartial effects require idempotency or reconciliation
MeasurementOne service objective represents user impactQuality, latency, and cost need separate gates

Operating Prevent Agent Loops & Runaway Tools as a System

The implementation is only one part of Prevent Agent Loops & Runaway Tools. A production design also needs an explicit contract for inputs, outputs, ownership, and failure behavior. Write that contract before selecting a library. It should identify which component validates input, where state lives, what may be retried, and which result is authoritative when two components disagree. This prevents a convenient prototype boundary from silently becoming the long-term architecture.

Start with a representative baseline. Capture request shape, traffic distribution, dependency latency, error classes, and the quality signal users actually care about. Averages hide the cases that cause incidents, so keep percentiles and segment measurements by workload type. Record the configuration and dataset version beside every result. Without that context, a faster or more accurate run cannot be reproduced and should not be used to approve a rollout.

Define the failure model

List failures by where they originate: invalid input, capacity exhaustion, dependency timeout, partial state change, malformed output, and semantically wrong output. Each class needs a different response. Validation errors should fail immediately. Transient dependency failures may be retried with a budget and jitter. An operation that may have committed must use an idempotency key or reconciliation step before retrying. A syntactically valid but incorrect result belongs in evaluation and review, not a blind retry loop.

Set a deadline for the complete operation and derive smaller budgets for each dependency. Local timeouts that add up to more than the caller's deadline merely create abandoned work. Propagate cancellation where the protocol supports it. Bound every queue, retry loop, context buffer, and concurrency pool; an unbounded safety mechanism becomes a second outage during overload.

Design a degraded mode before it is needed. Depending on the workload, that can mean returning a cached answer, selecting a simpler path, placing work in a durable queue, or asking for human review. The degraded response must be visible in telemetry and, where it changes meaning, visible to the caller. Silent fallback makes quality regressions almost impossible to diagnose.

Measure the decision, not just the component

Use three layers of signals. System metrics cover latency, throughput, saturation, and errors. Correctness metrics measure whether the result satisfies its contract. Business or user metrics show whether the system solved the intended problem. Improving only one layer can move the others backward, so release criteria should name acceptable movement for all three.

Attach a reason code to every route, rejection, fallback, and retry. Include version identifiers for configuration, code, model, schema, and data when relevant. Logs should let an engineer reconstruct a decision without storing secrets or raw personal data. Traces should cross process boundaries, while metrics should remain low-cardinality enough to operate reliably.

Alert on symptoms that require action, not every internal anomaly. A useful alert names the affected service objective, links to a runbook, and distinguishes a customer-visible incident from exhausted headroom. Dashboards serve a different purpose: they support diagnosis and capacity planning. Treating a dashboard as an alerting strategy leaves failures undiscovered until someone happens to look.

Roll out with reversible steps

Ship Prevent Agent Loops & Runaway Tools behind a versioned interface and a kill switch. Begin with offline replay using production-shaped, privacy-safe samples. Then use shadow execution when duplicate work has acceptable cost and side effects can be suppressed. A small canary should exercise the real dependency graph before traffic expands. Compare the canary with the baseline by cohort rather than mixing both populations into one aggregate.

Promotion gates should be written before the rollout. Include a minimum sample size or observation window, maximum regression in tail latency and error rate, and a correctness threshold. Roll back automatically when a hard safety boundary is crossed; use manual review for ambiguous quality movement. Preserve enough evidence from both paths to explain why the gate passed or failed.

Configuration deserves the same discipline as code. Review changes, validate them before activation, keep an immutable history, and make rollback a single operation. If a deployment changes code and configuration together, record both versions. Otherwise an incident responder may roll back the binary while leaving the triggering configuration active.

Capacity and cost controls

Model capacity in units the bottleneck understands: concurrent connections, tokens, queue jobs, database transactions, GPU memory, or bytes in flight. Convert the expected traffic distribution into those units and include burst behavior. Then load-test the first constrained dependency, not merely the public endpoint. A system that accepts more work than it can finish within its deadline is overloaded even if CPU utilization looks comfortable.

Cost is also a reliability limit. Add per-request attribution, tenant or workflow budgets, and a global circuit breaker for unexpectedly expensive paths. Review unit economics at the same granularity as performance; a cheap median can conceal a small class of requests responsible for most spend. Optimize only after measuring, because reducing context, replicas, validation, or redundancy can trade visible cost for less visible risk.

Production readiness review

Before launch, ask an engineer who did not build the feature to follow the runbook through one simulated failure. Verify backups or checkpoints by restoring them, not by checking that a job reported success. Exercise credential rotation, dependency unavailability, bad configuration, and rollback. Assign an owner for each alarm and a date for reviewing thresholds after real traffic arrives.

The final architecture document should be short enough to remain current. Keep the decision, rejected alternatives, invariants, dependency contracts, dashboards, and rollback procedure. Link detailed experiments rather than pasting them into the document. Teams that need help turning this review into an operable service can use our Prevent Agent Loops & Runaway Tools engineering support.

Frequently Asked Questions

How many iterations should I allow?

10-20 iterations for most production agents. Simple tasks complete in 3-5 iterations. Complex research tasks may need 30-50. Set limits based on task type:

  • Simple: 5-10 iterations
  • Medium: 10-20 iterations
  • Complex: 20-30 iterations
  • Exploratory: 30-50 iterations

Start conservative and adjust based on production data.

How do I detect loops early?

Track action history and state changes:

  1. Store last 5-10 actions
  2. Count repetitions of same action
  3. Alert after 3 repetitions
  4. Check if state is changing between iterations
  5. Detect alternating patterns (A→B→A→B)

When should circuit breakers open?

Open after 3-5 consecutive failures to the same service. Keep open for 30-60 seconds before testing recovery. For critical services, open more aggressively (2 failures). For less critical, allow more failures (10+).

How do I prevent cost runaway?

Implement multi-level cost guards:

  1. Per-task budget limits
  2. Per-user budget limits
  3. System-wide rate limits
  4. Alert at 80% budget consumption
  5. Hard stop at 100% budget

Track both token usage and dollar cost.

Should I kill the agent or degrade gracefully?

Degrade gracefully when possible:

  • Approaching limits: Simplify behavior, use cheaper models
  • At limits: Complete current task with reduced scope
  • Exceeded limits: Stop immediately, return partial results

Hard kills lose all progress. Graceful degradation preserves completed work.

How do I test loop prevention?

Test scenarios:

  1. Repeated tool calls - Same action 3+ times
  2. Alternating actions - A→B→A→B pattern
  3. State stagnation - No progress across iterations
  4. Resource exhaustion - Hit memory/CPU limits
  5. Timeout handling - Execution exceeds time limit

Mock failing tools and verify loop detection triggers.

What metrics should I monitor?

Critical metrics:

  • Iterations per task (p50, p95, p99)
  • Loop detection rate (% of tasks)
  • Circuit breaker trips (by service)
  • Resource usage (memory, CPU, tokens)
  • Cost per task (actual spend)
  • Timeout rate (% exceeding limits)

Alert on anomalies.

How do I handle loops in multi-agent systems?

In multi-agent systems:

  • Track loops per agent and system-wide
  • Supervisor detects when worker agents loop
  • Implement collective iteration limits (all agents combined)
  • Use delegation quotas to prevent circular delegation

Can loops be legitimate?

Yes, some patterns are valid:

  • Retry logic - 2-3 attempts for transient failures acceptable
  • Iterative refinement - Multiple passes for quality improvement
  • User clarification - Back-and-forth in conversational agents

Distinguish productive iteration from unproductive looping by checking state progress.

How do I debug agent loops in production?

  1. Examine action history - What actions repeated?
  2. Check state evolution - Did state change between iterations?
  3. Review tool results - Were tools returning errors?
  4. Analyze reasoning - What was agent thinking?
  5. Check external factors - API failures causing loops?

Enable detailed logging for loop incidents.

Conclusion

Production AI agents need comprehensive safeguards against infinite loops, runaway tool calls, and resource exhaustion. Iteration limits, loop detection, circuit breakers, and resource monitoring prevent agents from spiraling out of control.

Essential safeguards:

  • Iteration limits: Hard caps on execution steps (10-20 for most tasks)
  • Loop detection: Track action patterns and state changes
  • Circuit breakers: Prevent cascading failures from external services
  • Resource monitoring: Track memory, CPU, tokens, cost
  • Timeouts: Hierarchical timeouts at tool, LLM, and agent levels
  • Cost guards: Budget limits with early warnings
  • Graceful degradation: Adjust behavior as limits approached

Safety is not optional. Uncontrolled agents can exhaust API quotas, run up thousands in costs, or crash systems in minutes. Implement these safeguards from day one, not after the first incident.

Start with aggressive limits, then relax based on production evidence. Better to be overly cautious than to debug a $10,000 runaway agent bill.

Ready to build safe, production-grade AI agents? Contact our team or explore our agent safety case studies.

Free consultation

Book a free consultation call on agent loop prevention & safeguards

30-minute call with the HinterBuild team. Discuss your project, architecture questions, or next steps — no obligation.

Book a meeting

Resources:

Keep reading

Related articles

AI Systems · 10 min read

Reliable Tool Calling: Production AI Agent Error Handling &

Reliable Tool Calling guidance for engineers: compare architecture choices, avoid failure modes, and ship a measurable, reliable production implementation.

Read post

AI Systems · 12 min read

ReAct vs Plan-and-Execute: Agent Reasoning Patterns Compared

Learn react vs plan-and-execute through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.

Read post

AI Systems · 9 min read

Multi-Agent Orchestration Patterns: Production Guide for AI

Learn multi-agent orchestration patterns through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.

Read post

AI Systems · 13 min read

LangGraph vs CrewAI vs AutoGen: Which AI Agent Framework to

Learn langgraph vs crewai vs autogen through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.

Read post