Tool failures are inevitable in production AI agents — the question is whether your system handles them gracefully or catastrophically. This lesson builds a complete fallback chain framework from error classification through LLM-aware partial result handling, with real code you can deploy today.
You've built an AI agent that can call tools — maybe it queries a live pricing API, runs SQL against your data warehouse, or fetches documents from a vector store. In your development environment, everything works beautifully. Then you deploy to production, and within 48 hours you're staring at a wall of errors: the pricing API timed out, the SQL query returned a partial result set, a third-party service returned a 503, and your agent — with no strategy for any of these situations — either crashed outright or hallucinated its way to a confident-sounding wrong answer.
This is the unglamorous reality of production agent systems. Tool calling is the mechanism that gives agents their power, but tools are external systems, and external systems fail. What separates a toy demo from a system you'd actually trust with a business-critical workflow is how gracefully it handles failure. Fallback chains — ordered sequences of alternative strategies that activate when a primary tool fails — are the core pattern you need. But implementing them well requires thinking carefully about error classification, timeout budgets, partial result handling, and how to keep the LLM informed about what happened without derailing the conversation.
By the end of this lesson, you'll have a production-ready fallback framework you can drop into any agent architecture.
What you'll learn:
This lesson assumes you're comfortable with AI agent fundamentals — specifically function/tool calling, the ReAct loop, and agent state management. You should also have some familiarity with building multi-step agents with planning and memory and ideally have already connected at least one external tool to an agent. We'll use Python with OpenAI's API throughout, but the patterns apply to any LLM framework.
The single biggest mistake practitioners make when implementing fallbacks is treating all tool failures identically. They catch Exception, log it, and move to the next option. This is better than nothing, but it's sloppy in ways that bite you in production.
Tool failures fall into distinct categories, and your fallback strategy should depend on which category you're in:
Transient failures are temporary. The API is overloaded right now but will recover. A database connection dropped. A rate limit window just closed. For transient failures, the right first response is often a retry with exponential backoff — not an immediate switch to a fallback.
Permanent failures won't resolve on retry. The API key is invalid. The endpoint doesn't exist. The user doesn't have permissions for this resource. Retrying wastes time and budget; you should fall through to the next strategy immediately.
Partial failures are the trickiest. The tool returned something, but it's incomplete. Your SQL query returned 47 of 50 records before hitting a row limit. Your vector search returned results but with low confidence scores. The API returned data for 8 of 10 requested items. Partial results require a different response than total failures — you may be able to use what you got, supplement with a fallback, or surface the incompleteness to the LLM explicitly.
Timeout failures are a special case of transient failure, but they deserve separate treatment because of the latency implications. If you wait the full 30-second timeout before activating a fallback, your user experience is already destroyed. You need timeout budgets that are much shorter than the maximum theoretical wait time.
Here's the classification logic we'll build on throughout this lesson:
import httpx
import asyncio
from enum import Enum
from dataclasses import dataclass
from typing import Any, Optional
class FailureMode(Enum):
TRANSIENT = "transient" # Retry is worthwhile
PERMANENT = "permanent" # Don't retry, fall through immediately
PARTIAL = "partial" # Got some data, may be usable
TIMEOUT = "timeout" # Time budget exceeded
RATE_LIMITED = "rate_limited" # Wait before retry or fall through
@dataclass
class ToolResult:
success: bool
data: Any
failure_mode: Optional[FailureMode] = None
error_message: Optional[str] = None
completeness_score: float = 1.0 # 1.0 = complete, 0.0 = nothing useful
latency_ms: float = 0.0
tool_name: str = ""
def classify_http_error(status_code: int, response_body: str = "") -> FailureMode:
"""Classify an HTTP error into a failure mode."""
if status_code == 429:
return FailureMode.RATE_LIMITED
elif status_code in (500, 502, 503, 504):
return FailureMode.TRANSIENT
elif status_code in (400, 401, 403, 404, 422):
return FailureMode.PERMANENT
elif status_code == 206: # HTTP Partial Content
return FailureMode.PARTIAL
else:
return FailureMode.TRANSIENT # Default to transient for unknown errors
def classify_exception(exc: Exception) -> FailureMode:
"""Classify a Python exception into a failure mode."""
if isinstance(exc, asyncio.TimeoutError):
return FailureMode.TIMEOUT
elif isinstance(exc, httpx.ConnectError):
return FailureMode.TRANSIENT
elif isinstance(exc, httpx.TimeoutException):
return FailureMode.TIMEOUT
elif isinstance(exc, PermissionError):
return FailureMode.PERMANENT
elif isinstance(exc, (ValueError, TypeError)):
return FailureMode.PERMANENT # Bad input, won't improve on retry
else:
return FailureMode.TRANSIENT
Key insight: The
completeness_scorefield onToolResultis the thing most implementations forget. You need a way to represent the difference between "this tool completely failed" and "this tool gave me 60% of what I asked for." That distinction changes how you compose your response.
A fallback chain is an ordered list of tool strategies, where each strategy has its own timeout budget, retry policy, and activation conditions. When the first strategy fails, you evaluate whether to retry it, then move to the next strategy if appropriate, continuing until either one succeeds or you've exhausted all options.
Let's build this as a composable system:
import time
import asyncio
import logging
from typing import Callable, List, Optional
from dataclasses import dataclass, field
logger = logging.getLogger(__name__)
@dataclass
class RetryPolicy:
max_attempts: int = 3
base_delay_seconds: float = 1.0
max_delay_seconds: float = 30.0
backoff_multiplier: float = 2.0
retry_on: List[FailureMode] = field(
default_factory=lambda: [FailureMode.TRANSIENT, FailureMode.RATE_LIMITED]
)
@dataclass
class FallbackStrategy:
name: str
tool_fn: Callable
timeout_seconds: float
retry_policy: RetryPolicy = field(default_factory=RetryPolicy)
min_completeness_to_accept: float = 0.5 # Accept result if >= 50% complete
# If True, this strategy can supplement a partial result from a previous one
can_supplement: bool = False
class FallbackChain:
def __init__(self, strategies: List[FallbackStrategy], chain_name: str = "unnamed"):
self.strategies = strategies
self.chain_name = chain_name
self._execution_log: List[dict] = []
async def execute(self, **tool_kwargs) -> ToolResult:
"""
Execute the fallback chain, trying each strategy in order.
Returns the best result achieved, even if imperfect.
"""
best_partial_result: Optional[ToolResult] = None
for strategy in self.strategies:
result = await self._execute_strategy_with_retries(
strategy, **tool_kwargs
)
self._execution_log.append({
"strategy": strategy.name,
"success": result.success,
"completeness": result.completeness_score,
"failure_mode": result.failure_mode.value if result.failure_mode else None,
"latency_ms": result.latency_ms,
})
if result.success and result.completeness_score >= strategy.min_completeness_to_accept:
logger.info(
f"[{self.chain_name}] Strategy '{strategy.name}' succeeded "
f"(completeness={result.completeness_score:.2f})"
)
return result
if result.completeness_score > 0:
# Store as best partial result seen so far
if (best_partial_result is None or
result.completeness_score > best_partial_result.completeness_score):
best_partial_result = result
# Permanent failures: skip remaining retries for THIS strategy
# (already handled in _execute_strategy_with_retries), fall through
if result.failure_mode == FailureMode.PERMANENT:
logger.warning(
f"[{self.chain_name}] Strategy '{strategy.name}' permanently failed. "
f"Moving to next strategy."
)
# All strategies exhausted
if best_partial_result is not None:
logger.warning(
f"[{self.chain_name}] All strategies exhausted. Returning best partial result "
f"(completeness={best_partial_result.completeness_score:.2f})"
)
return best_partial_result
return ToolResult(
success=False,
data=None,
failure_mode=FailureMode.PERMANENT,
error_message=f"All fallback strategies in chain '{self.chain_name}' failed.",
completeness_score=0.0,
tool_name=self.chain_name,
)
async def _execute_strategy_with_retries(
self, strategy: FallbackStrategy, **tool_kwargs
) -> ToolResult:
"""Execute a single strategy with its retry policy."""
policy = strategy.retry_policy
attempt = 0
last_result = None
while attempt < policy.max_attempts:
attempt += 1
start_time = time.monotonic()
try:
result = await asyncio.wait_for(
strategy.tool_fn(**tool_kwargs),
timeout=strategy.timeout_seconds,
)
result.latency_ms = (time.monotonic() - start_time) * 1000
result.tool_name = strategy.name
if result.success:
return result
# Check if this failure mode is retryable
if result.failure_mode not in policy.retry_on:
return result # Not retryable, return immediately
last_result = result
except asyncio.TimeoutError:
latency_ms = (time.monotonic() - start_time) * 1000
last_result = ToolResult(
success=False,
data=None,
failure_mode=FailureMode.TIMEOUT,
error_message=f"Strategy '{strategy.name}' timed out after {strategy.timeout_seconds}s",
completeness_score=0.0,
latency_ms=latency_ms,
tool_name=strategy.name,
)
if FailureMode.TIMEOUT not in policy.retry_on:
return last_result
except Exception as exc:
latency_ms = (time.monotonic() - start_time) * 1000
failure_mode = classify_exception(exc)
last_result = ToolResult(
success=False,
data=None,
failure_mode=failure_mode,
error_message=str(exc),
completeness_score=0.0,
latency_ms=latency_ms,
tool_name=strategy.name,
)
if failure_mode not in policy.retry_on:
return last_result
# Calculate backoff delay
if attempt < policy.max_attempts:
delay = min(
policy.base_delay_seconds * (policy.backoff_multiplier ** (attempt - 1)),
policy.max_delay_seconds,
)
logger.info(
f"[{strategy.name}] Attempt {attempt} failed, retrying in {delay:.1f}s"
)
await asyncio.sleep(delay)
return last_result
Warning: Notice that
_execute_strategy_with_retriesusesasyncio.wait_forrather than relying on the tool's own timeout. This is critical. Never trust that an external tool will respect your deadline — always impose it from the outside. Many HTTP client libraries have their own timeout parameters, but those don't protect you from slow response bodies that stream indefinitely.
Let's make this concrete. You're building an agent for a retail analytics platform. One of its tools fetches current product pricing. Here's the landscape:
This is a textbook fallback chain. Let's implement the actual tool functions:
import httpx
import json
from datetime import datetime, timedelta
from typing import Optional
import redis
# Assume these are configured elsewhere
PRICING_API_BASE = "https://pricing.internal.company.com"
AGGREGATOR_API_BASE = "https://api.pricingaggregator.io"
REDIS_CLIENT = redis.Redis(host="localhost", port=6379, db=0)
async def fetch_from_internal_api(product_ids: list[str]) -> ToolResult:
"""Primary tool: Internal pricing API"""
async with httpx.AsyncClient() as client:
try:
response = await client.post(
f"{PRICING_API_BASE}/v2/prices/batch",
json={"product_ids": product_ids},
headers={"Authorization": f"Bearer {INTERNAL_API_KEY}"},
)
if response.status_code == 200:
data = response.json()
prices = data.get("prices", [])
# Check for partial results — API may not return all requested IDs
completeness = len(prices) / len(product_ids) if product_ids else 1.0
return ToolResult(
success=True,
data={"prices": prices, "source": "internal_api", "as_of": datetime.utcnow().isoformat()},
completeness_score=completeness,
)
else:
failure_mode = classify_http_error(response.status_code)
return ToolResult(
success=False,
data=None,
failure_mode=failure_mode,
error_message=f"Internal API returned {response.status_code}: {response.text[:200]}",
)
except Exception as exc:
return ToolResult(
success=False,
data=None,
failure_mode=classify_exception(exc),
error_message=str(exc),
)
async def fetch_from_aggregator(product_ids: list[str]) -> ToolResult:
"""Secondary tool: Third-party pricing aggregator"""
async with httpx.AsyncClient() as client:
try:
response = await client.get(
f"{AGGREGATOR_API_BASE}/prices",
params={"ids": ",".join(product_ids), "currency": "USD"},
headers={"X-API-Key": AGGREGATOR_API_KEY},
)
if response.status_code == 200:
data = response.json()
prices = data.get("results", [])
completeness = len(prices) / len(product_ids) if product_ids else 1.0
return ToolResult(
success=True,
data={"prices": prices, "source": "aggregator", "as_of": datetime.utcnow().isoformat()},
completeness_score=completeness,
)
elif response.status_code == 429:
return ToolResult(
success=False,
data=None,
failure_mode=FailureMode.RATE_LIMITED,
error_message="Aggregator rate limit exceeded",
)
else:
return ToolResult(
success=False,
data=None,
failure_mode=classify_http_error(response.status_code),
error_message=f"Aggregator API error: {response.status_code}",
)
except Exception as exc:
return ToolResult(
success=False,
data=None,
failure_mode=classify_exception(exc),
error_message=str(exc),
)
async def fetch_from_warehouse(product_ids: list[str]) -> ToolResult:
"""Tertiary tool: Data warehouse (synchronous query wrapped in async)"""
import asyncio
def _query_warehouse(ids):
# This would be your actual DB query
import psycopg2
conn = psycopg2.connect(WAREHOUSE_DSN)
cursor = conn.cursor()
cursor.execute(
"""
SELECT product_id, current_price, updated_at
FROM product_pricing_snapshot
WHERE product_id = ANY(%s)
AND updated_at > NOW() - INTERVAL '4 hours'
""",
(ids,)
)
rows = cursor.fetchall()
conn.close()
return [
{"product_id": r[0], "price": float(r[1]), "updated_at": r[2].isoformat()}
for r in rows
]
try:
# Run blocking DB call in a thread pool
loop = asyncio.get_event_loop()
prices = await loop.run_in_executor(None, _query_warehouse, product_ids)
completeness = len(prices) / len(product_ids) if product_ids else 1.0
return ToolResult(
success=True,
data={"prices": prices, "source": "warehouse_snapshot", "as_of": datetime.utcnow().isoformat(), "may_be_stale": True},
completeness_score=completeness,
)
except Exception as exc:
return ToolResult(
success=False,
data=None,
failure_mode=classify_exception(exc),
error_message=str(exc),
)
async def fetch_from_cache(product_ids: list[str]) -> ToolResult:
"""Last resort: Redis cache"""
prices = []
missing = []
for pid in product_ids:
cached = REDIS_CLIENT.get(f"price:{pid}")
if cached:
prices.append(json.loads(cached))
else:
missing.append(pid)
if not prices:
return ToolResult(
success=False,
data=None,
failure_mode=FailureMode.PERMANENT,
error_message="No cached prices available",
)
completeness = len(prices) / len(product_ids)
return ToolResult(
success=True,
data={"prices": prices, "source": "redis_cache", "missing_ids": missing, "is_cached": True},
completeness_score=completeness,
)
Now wire it all together into a chain:
def build_pricing_fallback_chain() -> FallbackChain:
return FallbackChain(
chain_name="product_pricing",
strategies=[
FallbackStrategy(
name="internal_pricing_api",
tool_fn=fetch_from_internal_api,
timeout_seconds=3.0, # Aggressive timeout — we expect this to be fast
retry_policy=RetryPolicy(
max_attempts=2,
base_delay_seconds=0.5,
retry_on=[FailureMode.TRANSIENT],
),
min_completeness_to_accept=0.9, # Need 90%+ of prices to accept this result
),
FallbackStrategy(
name="pricing_aggregator",
tool_fn=fetch_from_aggregator,
timeout_seconds=8.0,
retry_policy=RetryPolicy(
max_attempts=1, # Don't retry aggregator — charges per call
retry_on=[],
),
min_completeness_to_accept=0.7,
),
FallbackStrategy(
name="warehouse_snapshot",
tool_fn=fetch_from_warehouse,
timeout_seconds=15.0, # Warehouse queries can be slow
retry_policy=RetryPolicy(
max_attempts=2,
retry_on=[FailureMode.TRANSIENT],
),
min_completeness_to_accept=0.5,
),
FallbackStrategy(
name="redis_cache",
tool_fn=fetch_from_cache,
timeout_seconds=1.0,
retry_policy=RetryPolicy(max_attempts=1, retry_on=[]),
min_completeness_to_accept=0.0, # Accept whatever cache has
),
],
)
Tip: Notice the
min_completeness_to_acceptthresholds decrease as you go down the chain. Your primary source needs to be highly complete before you accept it (because you could fall through to a better partial from a later source). Your cache of last resort accepts anything, because something is better than nothing.
Here's where most implementations get it wrong: they return a partial or degraded result to the LLM and let the LLM respond as if it got perfect data. The agent then produces a confident answer based on incomplete information. That's a failure of honesty with downstream consequences.
The right approach is to enrich the tool result with context about its reliability, and inject that context into the system prompt or tool response in a way the LLM can use. Let's build that layer:
def format_tool_result_for_llm(result: ToolResult, original_request: dict) -> dict:
"""
Format a ToolResult into a structured tool response that communicates
data quality and completeness to the LLM.
"""
if result.success and result.completeness_score >= 0.95:
# Clean result, no caveats needed
return {
"status": "success",
"data": result.data,
}
elif result.success and result.completeness_score > 0:
# Partial success — communicate what's missing
missing_count = round((1 - result.completeness_score) * len(original_request.get("product_ids", [])))
staleness_warning = ""
if result.data and result.data.get("may_be_stale"):
staleness_warning = " Note: this data comes from a warehouse snapshot and may be up to 4 hours old."
if result.data and result.data.get("is_cached"):
staleness_warning = " Note: this data comes from cache and may be significantly out of date."
return {
"status": "partial_success",
"data": result.data,
"data_quality_warning": (
f"Retrieved prices for {round(result.completeness_score * 100)}% of requested products. "
f"Approximately {missing_count} product(s) could not be priced.{staleness_warning} "
f"Inform the user of this limitation in your response."
),
}
else:
# Total failure
return {
"status": "error",
"data": None,
"error_details": (
f"The pricing tool failed completely ({result.error_message}). "
f"Do not attempt to provide pricing information. "
f"Tell the user the pricing system is temporarily unavailable and suggest they check back shortly."
),
}
This formatted response gets returned as the tool's result to the LLM. The key insight is that you're not just returning data — you're returning instructions about how to use that data. A well-prompted LLM will follow those instructions and produce appropriately hedged responses.
Note: This connects directly to the broader challenge of prompt engineering for RAG systems. The same principle applies here: your system prompt should explicitly instruct the agent to communicate data quality issues to users rather than papering over them with confident-sounding prose.
Let's wire this into an actual agent that uses OpenAI's tool calling API:
from openai import AsyncOpenAI
import json
client = AsyncOpenAI()
# Define the tool schema for OpenAI
PRICING_TOOL_SCHEMA = {
"type": "function",
"function": {
"name": "get_product_prices",
"description": "Retrieve current pricing for one or more products by their product IDs.",
"parameters": {
"type": "object",
"properties": {
"product_ids": {
"type": "array",
"items": {"type": "string"},
"description": "List of product IDs to retrieve prices for",
},
},
"required": ["product_ids"],
},
},
}
async def execute_tool_call(tool_name: str, tool_args: dict) -> str:
"""Execute a tool call, routing through the appropriate fallback chain."""
if tool_name == "get_product_prices":
chain = build_pricing_fallback_chain()
result = await chain.execute(product_ids=tool_args["product_ids"])
formatted = format_tool_result_for_llm(result, tool_args)
return json.dumps(formatted)
else:
return json.dumps({"error": f"Unknown tool: {tool_name}"})
async def run_pricing_agent(user_message: str) -> str:
"""Run the pricing agent with full fallback chain support."""
messages = [
{
"role": "system",
"content": (
"You are a retail analytics assistant. When pricing data is incomplete or stale, "
"always communicate this clearly to the user. Never present partial data as complete. "
"If the pricing system is unavailable, say so directly and suggest alternatives."
),
},
{"role": "user", "content": user_message},
]
while True:
response = await client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=[PRICING_TOOL_SCHEMA],
tool_choice="auto",
)
message = response.choices[0].message
if message.tool_calls:
# Process tool calls
messages.append(message)
for tool_call in message.tool_calls:
tool_args = json.loads(tool_call.function.arguments)
tool_result = await execute_tool_call(
tool_call.function.name,
tool_args,
)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": tool_result,
})
else:
# No more tool calls — return the final response
return message.content
One of the subtler problems with fallback chains is that naively chaining timeouts can produce terrible user experience. If your primary tool times out after 3 seconds, your secondary after 8 seconds, and your tertiary after 15 seconds, and all three fail, you've kept the user waiting 26 seconds before admitting defeat. That's unacceptable.
The solution is a wall-clock budget — a single timer for the entire chain that overrides individual strategy timeouts:
import asyncio
import time
class BudgetedFallbackChain(FallbackChain):
def __init__(self, strategies: List[FallbackStrategy], chain_name: str = "unnamed",
total_budget_seconds: float = 10.0):
super().__init__(strategies, chain_name)
self.total_budget_seconds = total_budget_seconds
async def execute(self, **tool_kwargs) -> ToolResult:
"""Execute with a hard wall-clock deadline across the entire chain."""
deadline = time.monotonic() + self.total_budget_seconds
best_partial: Optional[ToolResult] = None
for strategy in self.strategies:
remaining = deadline - time.monotonic()
if remaining <= 0.2: # Less than 200ms left — not worth trying
logger.warning(
f"[{self.chain_name}] Budget exhausted before trying '{strategy.name}'"
)
break
# Cap this strategy's timeout at remaining budget
effective_timeout = min(strategy.timeout_seconds, remaining - 0.1)
adjusted_strategy = FallbackStrategy(
name=strategy.name,
tool_fn=strategy.tool_fn,
timeout_seconds=effective_timeout,
retry_policy=strategy.retry_policy,
min_completeness_to_accept=strategy.min_completeness_to_accept,
)
result = await self._execute_strategy_with_retries(
adjusted_strategy, **tool_kwargs
)
self._execution_log.append({
"strategy": strategy.name,
"success": result.success,
"completeness": result.completeness_score,
"failure_mode": result.failure_mode.value if result.failure_mode else None,
"latency_ms": result.latency_ms,
"budget_remaining_ms": (deadline - time.monotonic()) * 1000,
})
if result.success and result.completeness_score >= strategy.min_completeness_to_accept:
return result
if result.completeness_score > 0:
if best_partial is None or result.completeness_score > best_partial.completeness_score:
best_partial = result
return best_partial or ToolResult(
success=False,
data=None,
failure_mode=FailureMode.TIMEOUT,
error_message=f"Chain '{self.chain_name}' exhausted total budget of {self.total_budget_seconds}s",
completeness_score=0.0,
)
Key insight: For interactive agents where a human is waiting, a 10-second total budget is often the right call — even if that means you never try the warehouse and go straight to cache. A fast, honest degraded answer is almost always better than a slow perfect answer. Reserve long budgets for background or batch agent workflows where latency doesn't matter.
A fallback chain you can't observe is a fallback chain you can't improve. In production, you need to track:
Here's a simple Prometheus-compatible instrumentation layer:
from prometheus_client import Counter, Histogram, Gauge
# Metrics
tool_executions_total = Counter(
"agent_tool_executions_total",
"Total tool execution attempts",
["chain_name", "strategy_name", "outcome"]
)
tool_completeness = Histogram(
"agent_tool_completeness_score",
"Distribution of tool result completeness scores",
["chain_name", "strategy_name"],
buckets=[0.0, 0.25, 0.5, 0.75, 0.9, 0.95, 1.0]
)
chain_latency_seconds = Histogram(
"agent_chain_latency_seconds",
"Total chain execution time",
["chain_name", "final_outcome"],
buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0]
)
fallback_activations = Counter(
"agent_fallback_activations_total",
"Number of times a non-primary strategy was used",
["chain_name", "strategy_name"]
)
class InstrumentedFallbackChain(BudgetedFallbackChain):
async def execute(self, **tool_kwargs) -> ToolResult:
start = time.monotonic()
result = await super().execute(**tool_kwargs)
elapsed = time.monotonic() - start
# Record per-strategy metrics from execution log
for i, log_entry in enumerate(self._execution_log):
outcome = "success" if log_entry["success"] else log_entry.get("failure_mode", "unknown")
tool_executions_total.labels(
chain_name=self.chain_name,
strategy_name=log_entry["strategy"],
outcome=outcome,
).inc()
tool_completeness.labels(
chain_name=self.chain_name,
strategy_name=log_entry["strategy"],
).observe(log_entry["completeness"])
if i > 0 and log_entry["success"]:
# This is a fallback activation
fallback_activations.labels(
chain_name=self.chain_name,
strategy_name=log_entry["strategy"],
).inc()
# Record chain-level metrics
final_outcome = "success" if result.success else "failure"
if result.success and result.completeness_score < 0.95:
final_outcome = "partial_success"
chain_latency_seconds.labels(
chain_name=self.chain_name,
final_outcome=final_outcome,
).observe(elapsed)
return result
The patterns here integrate well with production RAG monitoring practices — the same infrastructure that watches your retrieval pipeline should watch your tool call chains.
Build a fallback chain for a news article retrieval tool used by a research agent. The agent needs to retrieve full article text given a URL or article ID. Here's your tool landscape:
Your tasks:
Implement the four tool functions following the ToolResult pattern above. For the purposes of the exercise, you can mock the actual HTTP calls but implement the error handling and classification logic for real.
Wire them into a BudgetedFallbackChain with a 12-second total budget. Think carefully about timeout allocation — the Wayback Machine is slow, so if it activates, how much budget should it get?
Implement format_tool_result_for_llm for this use case. What should the LLM say when it only has the title and URL?
Add one metric that isn't in the example above: track the reason that each fallback strategy was skipped (timeout, rate limit, etc.). This is valuable for diagnosing chain failures.
Stretch goal: Implement a simple cache-warming strategy. After a successful retrieval from the Wayback Machine (slow), write the result to the primary cache so that subsequent requests for the same article are fast.
Mistake 1: Timeouts that are too generous for the UX context
Setting a 30-second timeout on your primary tool because that's what the SLA says sounds reasonable. But if the primary tool regularly takes 15 seconds during peak hours, your fallback chain will often incur that 15-second wait before even starting. Profile your tools' actual latency distributions and set timeouts based on the P95, not the theoretical maximum.
Mistake 2: Retrying permanent failures
If your API key is invalid, retrying three times with backoff just means you fail three times more slowly. Always check for 401/403 status codes and classify them as PERMANENT so you fall through immediately.
Mistake 3: Not clearing the execution log between calls
If you're reusing a FallbackChain instance (which you should, for efficiency), make sure to clear self._execution_log at the start of each execute() call. Stale log entries from previous executions will corrupt your metrics.
Mistake 4: Ignoring partial results from early strategies
If your primary tool returns 80% of the requested products and then your secondary returns the other 20%, you should merge those results rather than returning them separately. Implement a result-merging layer for your specific data types.
Mistake 5: Treating the LLM as a black box for error handling
Some practitioners let the LLM decide what to do with an error message. This works sometimes but is fragile. Be explicit in your tool response format about what the LLM should do. "Tell the user pricing is unavailable" is better guidance than just {"error": "503"}.
Mistake 6: Not accounting for cascading failures
If all your data sources depend on the same upstream service (say, a common authentication provider), all your fallback strategies can fail simultaneously and for the same reason. Design your fallback chain so that strategies at different tiers have different dependencies wherever possible.
Warning: Fallback chains can mask systemic problems. If your primary tool has a bug that causes it to always return empty results (completeness_score = 0.0), your chain will silently fall through to the secondary every time. Without monitoring, you might not notice for days. Always alert when fallback activation rate exceeds a baseline threshold.
You now have a complete framework for handling tool failures in production agent systems. Let's recap what we built:
FallbackChain with per-strategy timeouts, retry policies, and completeness thresholds that correctly cascade through alternatives.This is hard-won production knowledge. Most agent frameworks give you tool calling; none of them give you a robust failure handling strategy. You have to build that yourself, and now you have the foundation.
Where to go next: