Error Handling in AI Agents: Circuit Breakers, Retry & Recovery

In deterministic software, errors are exceptions -- clearly defined failure states with predictable stack traces. In agentic AI systems, "errors" include hallucinations that return HTTP 200, tool calls that succeed technically but fail semantically, and reasoning chains that produce confident nonsense. Traditional try-catch blocks don't protect against these failure modes.

The challenge compounds in multi-step agentic workflows. A single agent interaction might involve parsing user intent, retrieving context from a vector database, calling an external API tool, generating a response with an LLM, validating that response against grounding sources, and formatting the final output. Each step introduces distinct failure modes, and an error at any stage can cascade through the entire pipeline. Without deliberate resilience engineering, even a brief network hiccup can bring down an otherwise well-designed agent.

For NCP-AAI certification candidates, mastering error handling and resilience patterns is critical for building production-grade agentic AI systems. Error handling and recovery spans across two NCP-AAI exam domains -- Agent Development (15%) and Agent Design (15%) -- and accounts for roughly 10-12% of exam questions. This guide covers every pattern you need to know, from basic retry logic with exponential backoff to sophisticated circuit breakers, semantic fallback strategies, NVIDIA-specific error handling tools, and human-in-the-loop escalation workflows.

The Unique Error Landscape of Agentic AI

Production AI agents must handle a wide range of failures gracefully. Unlike traditional web applications that deal primarily with HTTP errors and database connection issues, agentic AI systems face failures at every layer of the stack: API timeouts from external services, invalid tool parameters from malformed LLM outputs, authentication failures from expired tokens, resource exhaustion from rate limits and quotas, and -- most insidiously -- unexpected outputs like hallucinations and reasoning errors that look perfectly normal at the protocol level.

Traditional vs Agentic Error Taxonomy

The last category -- semantic errors -- represents the hardest challenge. An agent can execute perfectly, consume 5,000 tokens, invoke three tools successfully, and still produce a response that is completely wrong. This is unique to AI systems and has no direct analog in traditional software engineering.

Error Classification: Transient vs Client vs Semantic

Before choosing a recovery strategy, you must correctly classify the error. The NCP-AAI exam tests this classification heavily because each error type demands a fundamentally different response.

Transient errors are temporary failures where the same request will likely succeed if retried after a delay. These include HTTP 503 (Service Unavailable), network timeouts, and HTTP 429 (Rate Limit Exceeded). The correct recovery strategy is exponential backoff retry.

Client errors are failures caused by something wrong with the request itself. Retrying the identical request will never fix the problem. These include HTTP 400 (Bad Request) where parameters are malformed, HTTP 401 (Unauthorized) where the authentication token is expired or invalid, and HTTP 404 (Not Found) where the requested resource does not exist. Each requires a targeted fix before retrying: correct the parameters for 400, re-authenticate for 401, or update the resource path for 404.

Semantic errors are the most deceptive. The request succeeds at every technical layer -- valid HTTP 200, well-formed JSON, no exceptions -- but the content is factually wrong, logically inconsistent, or hallucinatory. These require validation-layer defenses like LLM-as-judge grounding checks, not retry or circuit breaker patterns.

HTTP Error Code Handling Reference

Understanding how to handle specific HTTP error codes is essential for the NCP-AAI exam. Here is a reference for the most commonly tested codes:

429 Too Many Requests: The server is rate-limiting your agent. Check for a Retry-After header in the response -- if present, wait exactly that many seconds before retrying. If no header is present, fall back to exponential backoff. The exam specifically tests whether you know to respect the Retry-After header rather than using a fixed delay or retrying immediately.

401 Unauthorized: The agent's authentication token is expired or invalid. The correct action is to refresh the token or re-authenticate, then retry the request with the new credentials. Simply retrying with the same expired token will produce the same 401 error indefinitely.

503 Service Unavailable: The backend service is temporarily down. This is the textbook case for exponential backoff retry. The service is expected to recover, so waiting and retrying is the correct approach.

400 Bad Request: Something is wrong with the request parameters. The agent must parse the error message, identify the malformed field, correct it, and then retry with the fixed parameters. Blind retries will never resolve a 400 error.

def handle_http_error(error, request_func, request_params):
    """Route HTTP errors to the correct recovery strategy."""
    if error.status_code == 429:
        # Respect Retry-After header if present
        retry_after = error.headers.get("Retry-After")
        if retry_after:
            time.sleep(int(retry_after))
            return request_func(**request_params)
        else:
            # Fall back to exponential backoff
            return retry_with_backoff(request_func, **request_params)

    elif error.status_code == 401:
        # Re-authenticate before retrying
        refresh_token()
        return request_func(**request_params)

    elif error.status_code == 400:
        # Parse error, fix parameters, retry once
        corrected_params = fix_parameters(error.message, request_params)
        return request_func(**corrected_params)

    elif error.status_code == 503:
        # Transient -- exponential backoff
        return retry_with_backoff(request_func, **request_params)

    elif error.status_code == 404:
        # Resource does not exist -- fail gracefully
        raise ResourceNotFoundError(f"Resource not found: {error.url}")

    else:
        raise UnhandledHTTPError(f"Unexpected error {error.status_code}")

Validation Errors

A separate category of errors occurs when the LLM generates tool call parameters that fail schema validation before the tool is even executed. These are distinct from both transient and semantic errors -- they are structural problems in the agent's output that can be caught deterministically without any network calls or LLM evaluation.

Common validation errors include type mismatches (expected integer, got string), out-of-range values (price = -$100), missing required fields (a flight search without a destination), and format violations (a date field containing free-form text instead of ISO 8601 format). In production agentic systems, validation errors are surprisingly common because LLMs generate structured output probabilistically rather than deterministically, and even well-prompted models occasionally produce malformed JSON, extra fields, or values outside the expected range.

The correct approach is to validate inputs before tool execution. This "validate-first" pattern prevents cascading failures downstream and avoids wasting resources on API calls that will inevitably fail. When validation fails, the error message should be fed back to the LLM so it can self-correct and generate valid parameters on the next attempt. This creates a tight feedback loop that resolves most validation errors within one or two retries.

def validate_tool_params(tool_name, params):
    """Validate tool parameters against schema before execution."""
    schema = get_tool_schema(tool_name)
    errors = []

    for field, rules in schema.items():
        if rules.get("required") and field not in params:
            errors.append(f"Missing required field: {field}")
        if field in params:
            value = params[field]
            if rules["type"] == "integer" and not isinstance(value, int):
                errors.append(f"{field} must be integer, got {type(value).__name__}")
            if "min" in rules and value < rules["min"]:
                errors.append(f"{field} value {value} below minimum {rules['min']}")
            if "max" in rules and value > rules["max"]:
                errors.append(f"{field} value {value} above maximum {rules['max']}")

    if errors:
        # Feed errors back to LLM for self-correction
        raise ToolValidationError(
            f"Invalid parameters for {tool_name}: {'; '.join(errors)}"
        )

    return True

Pattern 1: Retry with Exponential Backoff

Use Case: Transient failures (network timeouts, rate limits, temporary service outages)

Retry with exponential backoff is the most fundamental resilience pattern and the one most frequently tested on the NCP-AAI exam. The core idea is simple: when a transient error occurs, wait for an increasing amount of time before each subsequent retry attempt. This gives the failing service time to recover while avoiding the aggressive retry behavior that can make outages worse.

The pattern is appropriate only for transient errors -- errors where the same request is expected to succeed if retried after a delay. Applying it to client errors (400, 401) or semantic errors (hallucinations) is a common mistake that the exam specifically tests.

The Exponential Backoff Formula

Pattern 2: Circuit Breaker

Use Case: Prevent cascading failures when external services (APIs, databases, vector stores) become unhealthy

The circuit breaker pattern is borrowed from electrical engineering. Just as a physical circuit breaker trips to prevent an overloaded wire from starting a fire, a software circuit breaker stops sending requests to a failing service to prevent cascading failures across the system.

In agentic AI architectures, cascading failures are especially dangerous because agents typically depend on multiple external services -- vector databases for RAG retrieval, external APIs for tool execution, LLM inference endpoints for reasoning, and databases for state persistence. If the vector database goes down and the agent keeps hammering it with retry attempts, the agent's response latency spikes, which causes upstream timeouts, which causes the orchestrator to spawn more agent instances, which creates even more load on the already-struggling vector database. A circuit breaker interrupts this vicious cycle by quickly failing requests to the unhealthy service, giving it time to recover.

The circuit breaker pattern differs from retry logic in a critical way: retries handle individual transient failures, while circuit breakers detect sustained outages and protect the entire system from wasting resources on a service that is consistently failing. The NCP-AAI exam tests this distinction -- retries are for occasional hiccups, circuit breakers are for prolonged failures.

Circuit Breaker State Transitions

The circuit breaker has three states, and the NCP-AAI exam tests your understanding of the transitions between them:

State: CLOSED (normal operation -- all requests flow through)
  -> failure_count reaches threshold (e.g., 5 failures in 60s)
  -> Transition to OPEN

State: OPEN (all requests immediately rejected -- no calls to failing service)
  -> recovery_timeout elapses (e.g., 30 seconds)
  -> Transition to HALF_OPEN

State: HALF_OPEN (allow exactly one test request through)
  -> If test request SUCCEEDS -> Transition to CLOSED (reset failure count)
  -> If test request FAILS -> Transition back to OPEN (restart recovery timer)

Implementation

from enum import Enum
from datetime import datetime, timedelta
from threading import Lock

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

class CircuitBreaker:
    """Circuit breaker pattern for external dependencies."""

    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: int = 60,
        expected_exception: type = Exception
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.expected_exception = expected_exception

        self.failure_count = 0
        self.last_failure_time = None
        self.state = CircuitState.CLOSED
        self._lock = Lock()

    def call(self, func, *args, **kwargs):
        """Execute function with circuit breaker protection."""
        with self._lock:
            if self.state == CircuitState.OPEN:
                if self._should_attempt_reset():
                    self.state = CircuitState.HALF_OPEN
                else:
                    raise CircuitBreakerOpenError(
                        f"Circuit breaker OPEN. Retry after {self.recovery_timeout}s"
                    )

        try:
            result = func(*args, **kwargs)

            # Success -- reset if in half-open state
            with self._lock:
                if self.state == CircuitState.HALF_OPEN:
                    self.state = CircuitState.CLOSED
                    self.failure_count = 0

            return result

        except self.expected_exception as e:
            with self._lock:
                self.failure_count += 1
                self.last_failure_time = datetime.now()

                if self.failure_count >= self.failure_threshold:
                    self.state = CircuitState.OPEN

            raise

    def _should_attempt_reset(self) -> bool:
        """Check if enough time has passed to attempt recovery."""
        return (
            self.last_failure_time is not None and
            datetime.now() - self.last_failure_time >= timedelta(
                seconds=self.recovery_timeout
            )
        )

class CircuitBreakerOpenError(Exception):
    """Raised when circuit breaker is open."""
    pass

# Usage with agent tools
vector_db_breaker = CircuitBreaker(
    failure_threshold=3,
    recovery_timeout=30,
    expected_exception=(ConnectionError, TimeoutError)
)

def retrieve_context(query: str) -> list[str]:
    """Retrieve context from vector DB with circuit breaker."""
    return vector_db_breaker.call(
        vector_db.search,
        query_embedding=embed(query),
        top_k=5
    )

Tool-Specific Circuit Breakers

In production agentic systems, different tools have different reliability profiles and recovery characteristics. Configure separate circuit breakers for each dependency:

# Configure different breakers for different dependencies
tool_breakers = {
    "vector_search": CircuitBreaker(
        failure_threshold=3, recovery_timeout=30
    ),
    "api_external": CircuitBreaker(
        failure_threshold=5, recovery_timeout=60
    ),
    "database_query": CircuitBreaker(
        failure_threshold=2, recovery_timeout=20
    ),
    "payment_api": CircuitBreaker(
        failure_threshold=1, recovery_timeout=120  # Critical -- trip fast
    ),
}

def execute_tool_with_protection(tool_name: str, *args, **kwargs):
    """Execute tool with appropriate circuit breaker."""
    breaker = tool_breakers.get(tool_name)
    if breaker:
        return breaker.call(tools[tool_name].execute, *args, **kwargs)
    else:
        return tools[tool_name].execute(*args, **kwargs)

Pattern 3: Graceful Degradation with Fallback Strategies

Use Case: Maintain service availability when primary capabilities fail

Fallback strategies trade quality for availability. When the primary method fails, the system falls back to progressively simpler alternatives rather than returning a complete failure. This is one of the most important design principles for production agentic systems: a degraded response is almost always better than no response at all.

Consider a RAG-based customer support agent. Its primary strategy uses vector search to retrieve relevant documentation and a large, high-quality LLM to generate the response. If the vector database is temporarily unavailable, the agent could fall back to keyword-based search. If both search methods fail, it could fall back to the LLM's parametric knowledge without any retrieved context. If even the primary LLM is unavailable, it could use a smaller, faster model. Each fallback step reduces response quality, but the user still gets an answer.

The NCP-AAI exam tests your ability to design appropriate fallback hierarchies and to understand the tradeoffs at each level. A well-designed fallback chain degrades gracefully along multiple dimensions: model quality, context richness, response latency, and cost.

Fallback Chain Implementation

from typing import Optional, Callable, List, Any
from dataclasses import dataclass

@dataclass
class FallbackStrategy:
    """Defines a fallback option."""
    name: str
    executor: Callable
    max_attempts: int = 1
    cost_multiplier: float = 1.0  # Relative cost vs primary

class FallbackChain:
    """Execute strategies in order until one succeeds."""

    def __init__(self, strategies: List[FallbackStrategy]):
        self.strategies = strategies

    def execute(self, *args, **kwargs) -> Any:
        """Try each strategy until success."""
        last_error = None

        for strategy in self.strategies:
            for attempt in range(strategy.max_attempts):
                try:
                    result = strategy.executor(*args, **kwargs)
                    return result

                except Exception as e:
                    last_error = e

        # All strategies exhausted
        raise FallbackExhaustedError(
            f"All fallback strategies failed. Last error: {last_error}"
        )

# Example: RAG with multiple fallback strategies
def rag_primary(query: str) -> str:
    """Primary RAG: Vector search + large model."""
    context = vector_db.search(embed(query), top_k=5)
    return llm_large.generate(query, context)

def rag_fallback_cheaper_model(query: str) -> str:
    """Fallback 1: Same vector search, cheaper model."""
    context = vector_db.search(embed(query), top_k=5)
    return llm_small.generate(query, context)

def rag_fallback_keyword_search(query: str) -> str:
    """Fallback 2: Keyword search instead of vector."""
    context = keyword_search(query, top_k=5)
    return llm_large.generate(query, context)

def rag_fallback_no_context(query: str) -> str:
    """Fallback 3: Pure LLM, no retrieval."""
    return llm_large.generate(query, context=[])

# Define fallback chain
rag_chain = FallbackChain([
    FallbackStrategy("primary_rag", rag_primary,
                     max_attempts=2),
    FallbackStrategy("cheaper_model", rag_fallback_cheaper_model,
                     max_attempts=2, cost_multiplier=0.1),
    FallbackStrategy("keyword_search", rag_fallback_keyword_search,
                     max_attempts=1, cost_multiplier=0.8),
    FallbackStrategy("no_context", rag_fallback_no_context,
                     max_attempts=1, cost_multiplier=0.3),
])

# Usage
response = rag_chain.execute(user_query)

Graceful Degradation: Partial Results

Sometimes the best fallback is not a complete alternative strategy but partial results. If one component of a multi-tool workflow fails, the agent should deliver what it can rather than failing entirely.

class PartialResultBuilder:
    """Collect partial results from multi-tool workflows."""

    def __init__(self):
        self.results = {}
        self.failures = {}

    def execute_tool(self, tool_name: str, func: Callable, *args, **kwargs):
        """Execute tool, capturing both successes and failures."""
        try:
            self.results[tool_name] = func(*args, **kwargs)
        except Exception as e:
            self.failures[tool_name] = str(e)

    def build_response(self, query: str) -> str:
        """Build response from available partial results."""
        available_data = ", ".join(self.results.keys())
        unavailable_data = ", ".join(self.failures.keys())

        prompt = f"""
        User query: {query}
        Available data: {self.results}
        Unavailable services: {unavailable_data}

        Provide the best answer using available data.
        Clearly state which information is unavailable.
        """
        return llm.generate(prompt)

# Example: Travel assistant with partial degradation
builder = PartialResultBuilder()

builder.execute_tool("flights", search_flights, destination="Paris")
builder.execute_tool("weather", get_weather_forecast, city="Paris")
builder.execute_tool("hotels", search_hotels, city="Paris")

# If weather API fails:
# "Found 3 flights and 12 hotels in Paris. Weather data is
#  currently unavailable -- check back shortly."
response = builder.build_response("Show me travel options for Paris")

The NCP-AAI exam tests whether you understand that partial results are better than complete failure. If a question describes a multi-tool workflow where one tool fails, the correct answer almost always involves returning partial results with a clear indication of what is missing.

Pattern 4: Semantic Validation and Self-Correction

Use Case: Detect and recover from hallucinations, reasoning errors, invalid outputs

This pattern addresses the unique agentic AI challenge where the system returns technically successful but semantically incorrect responses. Unlike all the patterns discussed so far -- which deal with infrastructure-level failures that are detectable through error codes and exceptions -- semantic validation must evaluate the content of the response itself.

Semantic errors are particularly dangerous because they pass every technical check. The HTTP status is 200. The JSON is well-formed. The Pydantic model validates successfully. The response is fluent, confident, and detailed. But the information is wrong. In a customer support agent, this might mean confidently telling a customer the wrong return policy. In a medical information agent, it could mean citing a study that does not exist. In a financial agent, it could mean recommending an investment based on hallucinated performance data.

Production agentic systems require two layers of semantic defense: structural output validation (ensuring the response conforms to expected schemas, contains required fields, and passes basic sanity checks) and content grounding validation (ensuring the factual claims in the response are supported by the retrieved source documents).

Step 1: Output Validation with Structured Schemas

from pydantic import BaseModel, Field, validator
from typing import Literal

class AgentOutput(BaseModel):
    """Validated agent response."""
    answer: str = Field(..., min_length=10, max_length=2000)
    confidence: float = Field(..., ge=0.0, le=1.0)
    sources: list[str] = Field(default_factory=list)
    safety_check: Literal["safe", "unsafe"] = "safe"

    @validator("answer")
    def answer_not_refusal(cls, v):
        """Detect refusals disguised as answers."""
        refusal_patterns = [
            "I cannot", "I don't have access", "I'm unable to",
            "As an AI", "I don't know", "I cannot provide"
        ]
        if any(pattern in v for pattern in refusal_patterns):
            raise ValueError("Agent refused to answer")
        return v

    @validator("sources")
    def sources_not_empty_if_factual(cls, v, values):
        """Require sources for factual claims."""
        answer = values.get("answer", "")
        if len(answer) > 200 and len(v) == 0:
            raise ValueError("Long answer requires sources")
        return v

def validated_agent_call(query: str) -> AgentOutput:
    """Call agent with output validation."""
    raw_response = agent.run(query)

    try:
        validated = AgentOutput(**raw_response)
        return validated
    except ValueError as e:
        raise ValidationError(f"Agent output validation failed: {e}")

Step 2: Hallucination Detection

def detect_hallucination(
    answer: str,
    sources: list[str]
) -> tuple[bool, float]:
    """Detect if answer is grounded in sources."""

    # Method 1: Semantic similarity check
    answer_embedding = embed(answer)
    source_embeddings = [embed(s) for s in sources]

    max_similarity = max(
        cosine_similarity(answer_embedding, source_emb)
        for source_emb in source_embeddings
    )

    # Method 2: LLM-as-judge
    judge_prompt = f"""
    Evaluate if the ANSWER is fully supported by the SOURCES.

    ANSWER: {answer}

    SOURCES:
    {chr(10).join(f"[{i+1}] {s}" for i, s in enumerate(sources))}

    Is the answer supported? Reply with:
    - "YES" if fully supported
    - "PARTIAL" if partially supported
    - "NO" if not supported or hallucinated

    Confidence (0.0-1.0):
    """

    judge_response = llm_judge.complete(judge_prompt)

    is_hallucination = (
        max_similarity < 0.6 or
        "NO" in judge_response.upper()
    )

    confidence = extract_confidence(judge_response)

    return is_hallucination, confidence

# Usage with auto-retry and progressive grounding
def agent_with_hallucination_guard(
    query: str,
    max_attempts: int = 3
) -> str:
    """Run agent with hallucination detection and retry."""

    for attempt in range(max_attempts):
        response = agent.run(query)

        is_hallucination, confidence = detect_hallucination(
            response["answer"],
            response["sources"]
        )

        if not is_hallucination:
            return response["answer"]

        # Retry with stronger grounding instruction
        agent.update_system_prompt(
            "You MUST cite sources for every factual claim. "
            "If unsure, say 'I don't have enough information.'"
        )

    raise HallucinationError(
        f"Agent hallucinated after {max_attempts} attempts"
    )

Pattern 5: Token Budget Management

Use Case: Prevent context window overflow, control costs

Token budget management prevents a subtle but critical failure mode: silently exceeding the model's context window, which causes either truncation (lost information) or API errors. In agentic systems where context grows dynamically through tool calls and conversation history, budget management is essential.

Unlike the other patterns in this guide which handle failures after they occur, token budget management is a preventive pattern -- it avoids the failure entirely by ensuring the prompt fits within limits before making the inference call. This makes it the most cost-effective resilience pattern since it prevents wasted inference calls on prompts that would fail or produce degraded results due to truncation.

The core challenge is allocating a fixed token budget across competing demands: system prompt (fixed), user query (variable), retrieved context documents (variable, potentially very large), conversation history (grows over time), and a reserve for the model's response. A good token budget manager prioritizes the most recent and most relevant content when truncation is necessary, discarding older conversation history before discarding retrieved context.

import tiktoken

class TokenBudgetManager:
    """Manage token budgets for agent interactions."""

    def __init__(
        self,
        model: str,
        max_prompt_tokens: int = 6000,
        max_completion_tokens: int = 2000,
        reserve_tokens: int = 500  # Safety margin
    ):
        self.encoding = tiktoken.encoding_for_model(model)
        self.max_prompt_tokens = max_prompt_tokens
        self.max_completion_tokens = max_completion_tokens
        self.reserve_tokens = reserve_tokens

    def count_tokens(self, text: str) -> int:
        """Count tokens in text."""
        return len(self.encoding.encode(text))

    def truncate_context(
        self,
        system_prompt: str,
        user_query: str,
        context_docs: list[str],
        conversation_history: list[dict]
    ) -> dict:
        """Truncate inputs to fit budget."""

        # Fixed costs (always included)
        system_tokens = self.count_tokens(system_prompt)
        query_tokens = self.count_tokens(user_query)
        fixed_tokens = system_tokens + query_tokens

        # Available budget for dynamic content
        available_budget = (
            self.max_prompt_tokens -
            fixed_tokens -
            self.reserve_tokens
        )

        if available_budget < 0:
            raise TokenBudgetError("Query exceeds maximum prompt size")

        # Allocate budget: 60% context, 40% history
        context_budget = int(available_budget * 0.6)
        history_budget = int(available_budget * 0.4)

        # Truncate context documents
        truncated_context = self._truncate_docs(
            context_docs, context_budget
        )

        # Truncate conversation history (keep recent messages)
        truncated_history = self._truncate_history(
            conversation_history, history_budget
        )

        return {
            "system_prompt": system_prompt,
            "user_query": user_query,
            "context": truncated_context,
            "history": truncated_history,
            "tokens_used": fixed_tokens + context_budget + history_budget
        }

    def _truncate_docs(self, docs: list[str], budget: int) -> list[str]:
        """Truncate document list to fit budget."""
        truncated = []
        tokens_used = 0

        for doc in docs:
            doc_tokens = self.count_tokens(doc)
            if tokens_used + doc_tokens <= budget:
                truncated.append(doc)
                tokens_used += doc_tokens
            else:
                # Partial doc inclusion
                remaining_budget = budget - tokens_used
                if remaining_budget > 100:  # Minimum useful size
                    partial_doc = self.encoding.decode(
                        self.encoding.encode(doc)[:remaining_budget]
                    )
                    truncated.append(partial_doc + "...")
                break

        return truncated

# Usage in RAG agent
budget_manager = TokenBudgetManager(
    model="gpt-4-turbo",
    max_prompt_tokens=8000
)

def rag_agent_with_budget(query: str) -> str:
    """RAG agent with automatic token budget management."""

    # Retrieve more documents than we can use
    candidate_docs = vector_db.search(query, top_k=20)

    # Truncate to fit budget
    truncated_inputs = budget_manager.truncate_context(
        system_prompt=AGENT_SYSTEM_PROMPT,
        user_query=query,
        context_docs=candidate_docs,
        conversation_history=get_recent_history(limit=10)
    )

    # Generate with guaranteed fit
    response = llm.generate(
        system=truncated_inputs["system_prompt"],
        messages=truncated_inputs["history"],
        context=truncated_inputs["context"],
        query=truncated_inputs["user_query"],
        max_tokens=budget_manager.max_completion_tokens
    )

    return response

Pattern 6: Multi-Agent Consensus for Critical Decisions

Use Case: High-stakes decisions where errors are costly (medical diagnosis, financial advice, legal analysis)

When a single agent's error could have severe consequences, running multiple independent agents and requiring consensus provides an additional safety layer. This pattern prioritizes correctness over speed and cost efficiency.

The intuition behind multi-agent consensus is the same as having multiple doctors review a difficult diagnosis or multiple engineers review a critical design. Each agent processes the query independently, potentially using different models, different prompting strategies, or different RAG configurations. If a majority of agents agree on the answer, the system has high confidence in its correctness. If agents disagree significantly, the system escalates to human review rather than guessing.

This pattern is expensive -- it multiplies inference costs by the number of agents -- so it is reserved for high-stakes domains where the cost of an error far exceeds the cost of additional compute. The NCP-AAI exam tests whether you can identify scenarios where consensus is appropriate versus scenarios where a simple fallback chain is sufficient. The key differentiator is the cost and reversibility of errors: if an incorrect answer can be easily corrected (customer support), use fallback chains; if an incorrect answer has irreversible consequences (financial transactions, medical decisions), use consensus.

from collections import Counter
from typing import List

def multi_agent_consensus(
    query: str,
    agents: List[Agent],
    min_agreement: float = 0.7
) -> str:
    """Run multiple agents and require consensus."""

    responses = []
    for agent in agents:
        try:
            response = agent.run(query)
            responses.append(response)
        except Exception as e:
            print(f"Agent {agent.name} failed: {e}")

    if len(responses) < 2:
        raise InsufficientResponsesError(
            "Need at least 2 agent responses"
        )

    # Check for consensus using semantic similarity
    response_hashes = [hash_response(r) for r in responses]
    most_common = Counter(response_hashes).most_common(1)[0]
    agreement_rate = most_common[1] / len(responses)

    if agreement_rate >= min_agreement:
        # Consensus reached
        consensus_response = next(
            r for r in responses
            if hash_response(r) == most_common[0]
        )
        return consensus_response
    else:
        # No consensus -- escalate to human
        raise ConsensusFailureError(
            f"Agents disagree ({agreement_rate:.1%} agreement). "
            f"Escalating to human review."
        )

def hash_response(response: str) -> int:
    """Hash response for consensus checking."""
    # In production: use embedding similarity instead of exact match
    return hash(response.lower().strip())

Pattern 7: Human-in-the-Loop Escalation

Use Case: When automated recovery fails, compliance-critical operations, high-stakes decisions

Some failures cannot be resolved by automated patterns. When retries are exhausted, circuit breakers are open, and fallbacks have failed, the system must escalate to a human operator. This pattern is especially important for compliance-critical operations in finance, healthcare, and legal domains.

class EscalationManager:
    """Manage human-in-the-loop escalation for unrecoverable failures."""

    def __init__(self, notification_service, max_auto_retries: int = 3):
        self.notification_service = notification_service
        self.max_auto_retries = max_auto_retries

    def execute_with_escalation(
        self,
        func: Callable,
        *args,
        escalation_reason: str = "",
        priority: str = "normal",
        **kwargs
    ) -> Any:
        """Execute with automatic escalation on repeated failure."""

        for attempt in range(self.max_auto_retries):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                if attempt == self.max_auto_retries - 1:
                    # All automated recovery exhausted -- escalate
                    ticket_id = self.notification_service.create_ticket(
                        title=f"Agent escalation: {escalation_reason}",
                        description=(
                            f"Automated recovery failed after "
                            f"{self.max_auto_retries} attempts.\n"
                            f"Last error: {e}\n"
                            f"Function: {func.__name__}\n"
                            f"Args: {args}"
                        ),
                        priority=priority,
                    )
                    raise EscalationError(
                        f"Escalated to human review. "
                        f"Ticket: {ticket_id}"
                    )
                time.sleep(2 ** attempt)

# Usage: Payment processing with escalation
escalation_mgr = EscalationManager(
    notification_service=pager_duty_client,
    max_auto_retries=3
)

def process_payment(order_id: str, amount: float):
    """Process payment with human escalation on failure."""
    return escalation_mgr.execute_with_escalation(
        payment_gateway.charge,
        order_id=order_id,
        amount=amount,
        escalation_reason=f"Payment failed for order {order_id}",
        priority="high"
    )

The NCP-AAI exam tests human-in-the-loop escalation in the context of compliance-critical operations. The key concepts to remember:

Escalation is the last resort. It should only trigger after all automated recovery strategies (retries, circuit breakers, fallbacks) have been exhausted. Escalating too early wastes human operator time; escalating too late (or not at all) risks silent failures in critical workflows.

Escalation requires active notification, not just logging. Writing an error to a log file is not escalation. Production systems must use active notification channels -- PagerDuty alerts, Slack notifications, email tickets, or SMS -- to ensure a human operator is aware of the failure and can take action.

Context is essential. The escalation notification must include enough information for the human operator to diagnose and resolve the issue without having to reproduce it: the original query, the error type and message, the number of retry attempts, which fallback strategies were tried, timestamps, and relevant identifiers (order IDs, session IDs, etc.).

The workflow should be resumable. After human intervention resolves the underlying issue, the agent workflow should be able to resume from where it left off rather than requiring the user to start over. This means persisting workflow state at checkpoints throughout the execution pipeline.

NVIDIA Error Handling Tools

Understanding NVIDIA's specific tools and frameworks for error handling is directly tested on the NCP-AAI exam. These tools implement many of the patterns described above in production-ready, enterprise-grade packages that integrate with the NVIDIA AI Enterprise ecosystem.

NeMo Guardrails for Input and Output Validation

NVIDIA NeMo Guardrails provides a declarative framework for validating both inputs to and outputs from LLM-based agents. Guardrails operate as a middleware layer that intercepts requests before they reach the LLM and validates responses before they reach the user. This is the NVIDIA-native solution for the semantic validation pattern (Pattern 4) described earlier.

NeMo Guardrails uses a Colang-based configuration language to define rails -- validation rules that trigger on specific conditions. Rails can check for prompt injection attempts, PII leakage, hallucinated content, off-topic responses, and harmful output. When a rail is violated, NeMo Guardrails can block the request, modify the response, or trigger a fallback action.

# NeMo Guardrails configuration for error handling
rails:
  input:
    - type: validation
      check: no_malicious_code
    - type: validation
      check: no_pii_data
    - type: validation
      check: input_length_limit
  output:
    - type: validation
      check: no_hallucinations
    - type: validation
      check: fact_verification
    - type: validation
      check: no_harmful_content

The key exam concept is that guardrails operate before and after LLM processing -- they form a bidirectional validation layer. Input rails prevent malicious or malformed queries from reaching the model, while output rails catch hallucinations, harmful content, and factual errors before they reach the user.

NeMo Agent Toolkit Error Policies

The NVIDIA NeMo Agent Toolkit provides built-in error recovery configuration that integrates retry logic, backoff strategies, and fallback responses into the agent lifecycle:

from nemo_agent import Agent, ErrorPolicy

agent = Agent(
    model="nvidia/llama-3-70b-nemo",
    error_policy=ErrorPolicy(
        max_retries=3,
        backoff_strategy="exponential",
        fallback_response="I encountered an error. Please try again.",
        on_tool_error="retry_with_different_params",
        on_llm_error="fallback_to_smaller_model",
    )
)

For the NCP-AAI exam, know the difference between retry (same tool, same or corrected parameters) and fallback (alternative tool or model). The ErrorPolicy configuration distinguishes between tool-level errors and LLM-level errors because they require different recovery strategies.

NVIDIA Triton Inference Server Error Handling

When deploying agent LLMs via NVIDIA Triton Inference Server, additional error handling considerations come into play. Triton provides built-in health check endpoints that can be integrated with circuit breaker patterns to detect model loading failures, GPU memory exhaustion, and inference queue saturation before they cascade into agent-level failures.

Triton's gRPC and HTTP endpoints return specific error codes that map to different recovery strategies: model not loaded (requires admin intervention or model warm-up), inference queue full (retry after delay), and GPU out-of-memory (requires model scaling or request batching adjustments). Understanding these Triton-specific error modes is relevant for NCP-AAI questions about deploying agents on NVIDIA infrastructure.

Observability and Error Monitoring

No error handling strategy is complete without observability. Production agentic systems must track error rates, fallback usage, circuit breaker state changes, and escalation frequency. The key metrics to monitor include:

• Error rate by type: Percentage of requests resulting in transient errors, client errors, and semantic errors, tracked separately. A spike in any category triggers different investigation paths.
• Fallback usage rate: If more than 20% of requests are hitting fallback strategies, the primary strategy has a systemic issue that needs investigation, not just resilience.
• Circuit breaker state changes: Every transition from CLOSED to OPEN should trigger an alert. Frequent state oscillation (rapid CLOSED -> OPEN -> HALF_OPEN -> CLOSED cycles) indicates an unstable dependency.
• Mean time to recovery (MTTR): How long circuit breakers stay in OPEN state before successfully transitioning back to CLOSED. Increasing MTTR trends indicate degrading dependency health.
• Escalation frequency: The number of human-in-the-loop escalations per time period. If escalation rate increases, automated recovery patterns may need tuning.

These monitoring practices bridge the gap between implementing resilience patterns and operating them effectively in production. The NCP-AAI exam may present monitoring scenarios and ask which metric indicates a specific type of failure or which alert threshold is appropriate.

Combining Patterns: Production Architecture

In production, these patterns are never used in isolation. A robust agentic AI system layers multiple resilience patterns into a defense-in-depth architecture where each layer catches failures that slip through the layer above it.

The typical production stack looks like this, from outermost to innermost layer:

1. Input validation (NeMo Guardrails input rails) -- blocks malformed, malicious, or off-topic queries before any processing begins
2. Token budget management -- ensures the prompt fits within the model's context window before making expensive inference calls
3. Circuit breakers -- checks whether dependent services are healthy before attempting calls, failing fast if they are in OPEN state
4. Retry with exponential backoff -- handles transient failures within individual service calls
5. Fallback chains -- provides alternative strategies when primary approaches fail repeatedly
6. Semantic validation (NeMo Guardrails output rails, LLM-as-judge) -- catches hallucinations and factual errors in the generated response
7. Human-in-the-loop escalation -- final safety net when all automated recovery is exhausted

Each layer serves a distinct purpose, and removing any single layer creates a gap that certain failure modes will exploit. The NCP-AAI exam tests your ability to identify which layer addresses which failure type and to design architectures that cover the full spectrum of agentic AI errors.

class ResilientAgentPipeline:
    """Production pipeline combining all error handling patterns."""

    def __init__(self):
        self.circuit_breakers = {
            "llm": CircuitBreaker(failure_threshold=5, recovery_timeout=60),
            "vector_db": CircuitBreaker(failure_threshold=3, recovery_timeout=30),
            "tools": CircuitBreaker(failure_threshold=3, recovery_timeout=45),
        }
        self.token_manager = TokenBudgetManager(
            model="llama-3-70b",
            max_prompt_tokens=8000
        )
        self.escalation_mgr = EscalationManager(
            notification_service=alerting_client,
            max_auto_retries=3
        )

    def execute(self, query: str) -> str:
        """Execute query through full resilience pipeline."""

        # Layer 1: Input validation (NeMo Guardrails)
        validated_query = self.validate_input(query)

        # Layer 2: Token budget management
        context = self.retrieve_with_budget(validated_query)

        # Layer 3: LLM call with circuit breaker + retry
        try:
            response = self.circuit_breakers["llm"].call(
                self.call_llm_with_retry,
                validated_query,
                context
            )
        except CircuitBreakerOpenError:
            # Layer 4: Fallback to smaller model
            response = self.fallback_llm(validated_query, context)

        # Layer 5: Output validation (hallucination detection)
        validated_response = self.validate_output(
            response, context
        )

        return validated_response

Production Checklist: Error Resilience

Common NCP-AAI Exam Traps for Error Handling

The NCP-AAI exam includes several recurring trap patterns in error handling questions. Knowing these in advance can prevent costly mistakes:

Trap 1: Retrying client errors. If an agent receives a 401 Unauthorized or 400 Bad Request, the exam will offer "retry with exponential backoff" as a tempting answer. This is always wrong for client errors. The correct action is to fix the underlying issue (re-authenticate, correct parameters) before retrying.

Trap 2: Confusing semantic errors with runtime errors. When the question describes an agent that returns HTTP 200 with a confident but incorrect answer, do not select retry or circuit breaker as the solution. The correct answer involves output validation, hallucination detection, or grounding checks.

Trap 3: Circuit breaker in HALF_OPEN state. The exam may ask what happens when a test request fails in HALF_OPEN state. The answer is that the circuit breaker transitions back to OPEN (not CLOSED), and the recovery timer restarts.

Trap 4: Ignoring the Retry-After header. When the question specifies that a 429 response includes a Retry-After: 60 header, the correct action is to wait exactly 60 seconds -- not to use exponential backoff, not to retry immediately, and not to fail permanently. The Retry-After header is an explicit contract from the server.

Trap 5: Full failure vs partial results. When a multi-tool workflow has one tool failure out of three, the correct approach is graceful degradation with partial results, not complete failure or silent omission of the missing data.

Practice for NCP-AAI Exam

Test your error handling knowledge with Preporato's NCP-AAI Practice Tests:

• Retry logic and exponential backoff timing calculations
• Circuit breaker state transition scenarios
• HTTP error code classification (transient vs client)
• Fallback strategy design challenges
• Hallucination detection technique selection
• NeMo Guardrails configuration questions
• Human-in-the-loop escalation scenarios
• Production resilience pattern combinations

Start practicing today and master production-grade error handling for agentic AI systems.

Error Handling Decision Matrix

When facing an NCP-AAI exam question about error handling, use this decision tree to select the correct pattern:

Step 1: Classify the error.

• Did the request fail with an HTTP error code? Go to Step 2.
• Did the request succeed (HTTP 200) but the content is wrong? Apply semantic validation (Pattern 4).
• Did the request fail due to token/context limits? Apply token budget management (Pattern 5).

Step 2: Is it a transient or client error?

• Transient (503, timeout, 429): Go to Step 3.
• Client (400, 401, 404): Fix the root cause first (re-authenticate, correct parameters, update path), then retry once.

Step 3: Is it a single failure or a sustained outage?

• Single failure or occasional hiccup: Apply retry with exponential backoff (Pattern 1).
• Multiple consecutive failures across requests: Apply circuit breaker (Pattern 2).

Step 4: Have automated retries and circuit breakers been exhausted?

• No: Continue with retry/circuit breaker patterns.
• Yes: Apply fallback chain (Pattern 3) or graceful degradation (partial results).

Step 5: Is the operation high-stakes with irreversible consequences?

• No: Use the fallback chain result.
• Yes: Apply multi-agent consensus (Pattern 6) for additional validation.

Step 6: Have all automated strategies been exhausted?

• Apply human-in-the-loop escalation (Pattern 7) as the final safety net.

This decision tree maps directly to the exam's scenario-based questions. The exam will describe a specific failure situation and ask which pattern to apply. Using this classification framework eliminates the most common trap answers.

Pattern Selection Quick Reference

Conclusion

Error handling in agentic AI systems requires a fundamental shift from traditional software engineering patterns. The error landscape is broader -- spanning transient network failures, client request errors, validation failures, semantic hallucinations, and resource exhaustion -- and each category demands a fundamentally different recovery strategy. Applying the wrong pattern to the wrong error type (retrying a 401, circuit-breaking a hallucination) is not just ineffective -- it can make the problem worse.

The seven core patterns covered in this guide form a complete resilience framework:

1. Retry with exponential backoff (2^attempt seconds with jitter) handles transient failures while avoiding thundering herd problems
2. Circuit breakers (CLOSED -> OPEN -> HALF_OPEN state machine) prevent cascading failures when services are in sustained outage
3. Fallback chains and graceful degradation maintain availability by trading quality for reliability, returning partial results when full results are unavailable
4. Semantic validation and self-correction (LLM-as-judge, grounding checks) catch hallucinations and reasoning errors that pass all technical checks
5. Token budget management prevents context window overflow and controls costs in dynamic agent interactions
6. Multi-agent consensus provides correctness guarantees for high-stakes decisions where errors have irreversible consequences
7. Human-in-the-loop escalation serves as the final safety net when all automated recovery is exhausted

NVIDIA's NeMo Guardrails and Agent Toolkit provide production-ready implementations of these patterns, with declarative Colang-based configuration for bidirectional input/output validation and built-in error policies that integrate retry, backoff, and fallback strategies directly into the agent lifecycle.

For NCP-AAI certification, the exam tests three things: your ability to correctly classify error types (transient, client, semantic), your knowledge of which pattern addresses each error type, and your understanding of how these patterns combine in a layered production architecture. Mastering these concepts ensures both exam success and the ability to build resilient, production-grade agentic AI systems.