Tutorial 5: Handle Agent Failures

Contributor
4 days ago
3 min read

Agents fail in ways that simpler systems don't. They can loop, give up too early, call wrong tools, or get confused mid-task. This tutorial covers handling each.

What You'll Build

A production-safe agent with handling for the common failure modes.

Step 1: Identify the Failure Modes (10 min)

Common agent failures:

Loop: same tool called repeatedly with no progress
Stuck: agent reaches max steps without completing
Tool error: underlying tool failed
API error: LLM call failed
Confused: agent's response doesn't match intent
Overshooting: agent does too much, including unintended actions

Each needs different handling.

Step 2: Detect Loops (15 min)

def detect_loop(trace, lookback=3):
    if len(trace) < lookback + 1:
        return False
    
    recent = trace[-lookback:]
    return all(
        step["tool"] == recent[0]["tool"] and 
        step["input"] == recent[0]["input"]
        for step in recent
    )

# In the agent loop
if detect_loop(trace):
    # Insert a message that breaks the loop
    messages.append({
        "role": "user",
        "content": "You're repeating the same action. Try a different approach or give your best answer with what you have."
    })

Same input → same output → loop. Detect and break.

Step 3: Set Multiple Timeouts (10 min)

def agent_with_timeout(task, max_seconds=60, max_steps=10):
    start = time.time()
    
    for step in range(max_steps):
        # Check time
        if time.time() - start > max_seconds:
            return wrap_up(messages, "Time limit reached")
        
        # Check step
        if step == max_steps - 1:
            # Last step; force final answer
            messages.append({"role": "user", "content": "Give your best final answer."})
        
        # Check API call timeout
        response = client.messages.create(timeout=20, ...)
        # ...

Multiple boundaries prevent runaway.

Step 4: Handle Tool Failures (15 min)

def execute_tool_safely(tool_name, input):
    try:
        result = TOOL_FN[tool_name](**input)
        return {"success": True, "result": result}
    except KeyError:
        return {"success": False, "error": f"Unknown tool: {tool_name}"}
    except TypeError as e:
        return {"success": False, "error": f"Bad input: {e}"}
    except Exception as e:
        log_tool_error(tool_name, input, e)
        return {"success": False, "error": "Internal error; try a different approach"}

Return errors to the agent. It can recover.

Step 5: Handle LLM API Failures (15 min)

def llm_call_with_retry(messages, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.messages.create(messages=messages, ...)
        except anthropic.RateLimitError:
            wait = 2 ** attempt
            time.sleep(wait)
        except anthropic.APIError as e:
            if e.status_code >= 500:
                # Retry on server errors
                time.sleep(2 ** attempt)
            else:
                raise  # Don't retry client errors
    
    raise Exception("API call failed after retries")

Transient errors get retried; permanent errors fail clearly.

Step 6: Detect Confused Output (15 min)

After the agent completes:

def validate_output(task, output, trace):
    # Check if output addresses the task
    relevance_check = call_llm(f"""
        Task: {task}
        Agent output: {output}
        
        Does the output address the task? Respond with yes/no/partial.
    """)
    
    if "no" in relevance_check.lower():
        # Agent got confused; either retry or escalate
        return False, "Output didn't address the task"
    
    return True, None

Quality check before returning to user. Optional but useful for high-stakes.

Step 7: Build Fallback Paths (varies)

For each potential failure, what's the fallback?

def task_with_fallback(task):
    try:
        result = agent(task)
        if not validate_output(task, result.output):
            return fallback_response(task)
        return result.output
    except Exception as e:
        log_error(e)
        return fallback_response(task)

def fallback_response(task):
    return "I'm having trouble with that. Could you rephrase, or I can connect you with support?"

Graceful degradation. User experience preserved.

Step 8: Save Traces of Failures (15 min)

def save_failure(task, trace, error):
    db.insert("agent_failures", {
        "task": task,
        "trace": json.dumps(trace),
        "error": str(error),
        "timestamp": datetime.now(),
    })

The failures are how you improve.

Step 9: Periodic Review (weekly)

SELECT error, COUNT(*) as count
FROM agent_failures
WHERE timestamp > NOW() - INTERVAL '7 days'
GROUP BY error
ORDER BY count DESC;

Top failure modes are where to invest.

Step 10: Test Failure Handling (varies)

Intentionally cause failures:

Make a tool throw
Make the LLM API rate-limited
Submit a task that would loop
Submit a task the agent can't handle

For each, the system should fail gracefully — clear error to user; logged for the team; no infinite loop or crash.

What You Just Did

You built failure handling into your agent. Loops break; tools recover; users get sensible responses even when things go wrong.

Common Failure Modes

Optimistic loops. No max_steps; no time budget. Agent runs forever.

Silent tool failures. Tool throws; agent treats as if success.

No retries. Transient API error fails the whole task.

No fallback. Agent fails → user sees error.

No failure tracking. Can't improve what you don't measure.

ShiftQuality