Introduction to Error Recovery Patterns with LlamaIndex
Building production-grade LLM applications means accepting one fundamental truth: things will fail. Network requests time out, models return malformed JSON, vector databases drop connections, and rate limits kick in at the worst possible moments. LlamaIndex, a popular framework for building data-augmented LLM applications, provides several mechanisms to handle these failures gracefully. In this guide, we'll explore error recovery patterns that transform fragile prototypes into resilient production systems.
What Is Error Recovery in LlamaIndex?
Error recovery refers to the set of strategies and techniques used to detect, handle, and recover from failures that occur during the execution of a LlamaIndex pipeline. These failures can originate from multiple sources:
- LLM API failures — rate limits, timeouts, authentication errors, or malformed responses.
- Embedding service failures — connection issues or quota exhaustion when generating vector embeddings.
- Vector store failures — database disconnections, query errors, or index corruption.
- Data loading failures — corrupted documents, unsupported formats, or missing files.
- Parsing failures — invalid JSON output from structured output parsers or function-calling agents.
A robust error recovery strategy ensures that when any of these failures occur, your application can either retry the operation, fall back to an alternative approach, or degrade gracefully without crashing.
Why Error Recovery Matters
In a development environment, a failed request is a minor inconvenience — you simply rerun the script. In production, however, failures have real consequences. A single unhandled exception can break an entire user session, corrupt partially written data, or cause cascading failures across dependent services. For LLM applications specifically, the stakes are even higher because the cost of each request — both in latency and in dollars — is significant. Wasting a successful embedding call because a downstream LLM call failed is both inefficient and avoidable.
Effective error recovery patterns provide three key benefits: they improve reliability by reducing the frequency of total failures, they improve resilience by ensuring the system can recover when failures do occur, and they improve observability by giving you structured insight into what went wrong and how the system responded.
Pattern 1: Retry with Exponential Backoff
The most fundamental error recovery pattern is retrying failed operations with exponential backoff. LlamaIndex integrates with the tenacity library, which provides a clean decorator-based API for defining retry logic. This pattern is especially useful for transient failures like rate limits and temporary network issues.
Basic Retry Configuration
from llama_index.llms.openai import OpenAI
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import openai
# Define retry behavior using tenacity
retry_decorator = retry(
stop=stop_after_attempt(4),
wait=wait_exponential(multiplier=1, min=2, max=30),
retry=retry_if_exception_type((
openai.RateLimitError,
openai.APITimeoutError,
openai.APIConnectionError,
)),
reraise=True,
)
# Wrap the LLM call
@retry_decorator
def call_llm_with_retry(llm: OpenAI, prompt: str) -> str:
return llm.complete(prompt).text
llm = OpenAI(model="gpt-4", temperature=0)
try:
response = call_llm_with_retry(llm, "Explain vector databases in one sentence.")
print(response)
except Exception as e:
print(f"All retry attempts exhausted: {e}")
In this example, the function retries up to four times with exponentially increasing wait times between 2 and 30 seconds. The retry_if_exception_type filter ensures we only retry on specific transient errors rather than retrying on errors that would never succeed, such as authentication failures.
Integrating Retry into Custom Query Engines
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.query_engine import RetrieverQueryEngine
from tenacity import retry, stop_after_attempt, wait_exponential
import logging
logger = logging.getLogger(__name__)
class ResilientQueryEngine:
def __init__(self, index, llm):
self.query_engine = index.as_query_engine(llm=llm)
self.retry_config = retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10),
reraise=True,
)
def query(self, question: str) -> str:
@self.retry_config
def _execute():
return self.query_engine.query(question)
try:
return _execute()
except Exception as e:
logger.error(f"Query failed after retries: {e}")
return "I'm sorry, I encountered an error processing your request. Please try again."
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)
engine = ResilientQueryEngine(index, llm)
result = engine.query("What is in my documents?")
print(result)
Pattern 2: Fallback LLM Providers
Retries help with transient failures, but they won't help if your primary LLM provider experiences an extended outage. A fallback pattern lets you switch to an alternative provider when the primary one is unavailable. This is one of the most powerful resilience patterns for production LLM applications.
Implementing a Fallback LLM Wrapper
from llama_index.core.llms import LLM, CompletionResponse, LLMMetadata
from llama_index.llms.openai import OpenAI
from llama_index.llms.anthropic import Anthropic
from llama_index.llms.gemini import Gemini
from typing import Sequence, Optional
import logging
logger = logging.getLogger(__name__)
class FallbackLLM(LLM):
"""An LLM wrapper that tries multiple providers in order."""
def __init__(self, llms: Sequence[LLM]):
self._llms = list(llms)
super().__init__()
@property
def metadata(self) -> LLMMetadata:
return self._llms[0].metadata
def _complete(self, prompt: str, **kwargs) -> CompletionResponse:
errors = []
for llm in self._llms:
try:
response = llm.complete(prompt, **kwargs)
logger.info(f"Successfully completed using {type(llm).__name__}")
return response
except Exception as e:
logger.warning(f"{type(llm).__name__} failed: {e}")
errors.append(str(e))
raise RuntimeError(f"All LLM providers failed: {errors}")
async def _acomplete(self, prompt: str, **kwargs) -> CompletionResponse:
errors = []
for llm in self._llms:
try:
response = await llm.acomplete(prompt, **kwargs)
return response
except Exception as e:
logger.warning(f"{type(llm).__name__} failed: {e}")
errors.append(str(e))
raise RuntimeError(f"All LLM providers failed: {errors}")
# Configure fallback chain
primary_llm = OpenAI(model="gpt-4", temperature=0)
fallback_llm_1 = Anthropic(model="claude-3-sonnet-20240229", temperature=0)
fallback_llm_2 = Gemini(model="models/gemini-pro", temperature=0)
resilient_llm = FallbackLLM([primary_llm, fallback_llm_1, fallback_llm_2])
Using the Fallback LLM with an Index
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)
# Use the fallback LLM in your query engine
query_engine = index.as_query_engine(llm=resilient_llm)
try:
response = query_engine.query("Summarize the key points in my documents.")
print(response)
except RuntimeError as e:
print(f"Complete failure across all providers: {e}")
Pattern 3: Structured Output Recovery
One of the most common failure modes in LLM applications is when the model returns malformed structured output. LlamaIndex's StructuredLLMOutputParser and Pydantic-based output parsers can fail when the model doesn't follow instructions perfectly. A recovery pattern here involves retrying with corrective prompts.
Self-Correcting Output Parser
from pydantic import BaseModel, ValidationError
from llama_index.core.program import LLMTextCompletionProgram
from llama_index.core.output_parsers import PydanticOutputParser
from llama_index.llms.openai import OpenAI
import json
import logging
logger = logging.getLogger(__name__)
class PersonInfo(BaseModel):
name: str
age: int
occupation: str
city: str
def extract_with_recovery(llm: OpenAI, text: str, max_attempts: int = 3) -> PersonInfo:
parser = PydanticOutputParser(output_cls=PersonInfo)
prompt_template = (
"Extract person information from the following text.\n"
"Return ONLY valid JSON matching this schema:\n{schema}\n\n"
"Text: {text}\n\n"
"Previous error (if any): {error}\n"
"Please ensure the output is valid JSON."
)
last_error = "None"
for attempt in range(max_attempts):
try:
prompt = prompt_template.format(
schema=parser.format_schema(),
text=text,
error=last_error,
)
response = llm.complete(prompt)
parsed = parser.parse(response.text)
logger.info(f"Successfully parsed on attempt {attempt + 1}")
return parsed
except (ValidationError, json.JSONDecodeError) as e:
logger.warning(f"Attempt {attempt + 1} failed: {e}")
last_error = str(e)
except Exception as e:
logger.error(f"Unexpected error on attempt {attempt + 1}: {e}")
last_error = str(e)
raise ValueError(f"Failed to extract structured output after {max_attempts} attempts")
llm = OpenAI(model="gpt-4", temperature=0)
text = "John Smith is a 32-year-old software engineer living in Seattle."
person = extract_with_recovery(llm, text)
print(f"Name: {person.name}, Age: {person.age}, Role: {person.occupation}")
The key insight here is that each retry attempt includes the previous error message in the prompt. This gives the model context about what went wrong, dramatically improving the chances of a successful parse on subsequent attempts.
Pattern 4: Circuit Breaker for External Services
When an external service like a vector database or embedding API is experiencing sustained failures, continuously retrying can make the problem worse. A circuit breaker pattern monitors failure rates and temporarily stops sending requests when a threshold is reached, allowing the service time to recover.
Implementing a Circuit Breaker for Vector Store Operations
import time
import logging
from enum import Enum
from llama_index.core import VectorStoreIndex
from llama_index.core.storage.storage_context import StorageContext
from llama_index.vector_stores.chroma import ChromaVectorStore
logger = logging.getLogger(__name__)
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
class CircuitBreaker:
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: float = 60.0,
half_open_max_calls: int = 3,
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.half_open_max_calls = half_open_max_calls
self._state = CircuitState.CLOSED
self._failure_count = 0
self._last_failure_time = 0
self._half_open_calls = 0
@property
def state(self) -> CircuitState:
if self._state == CircuitState.OPEN:
if time.time() - self._last_failure_time > self.recovery_timeout:
logger.info("Circuit breaker transitioning from OPEN to HALF_OPEN")
self._state = CircuitState.HALF_OPEN
self._half_open_calls = 0
return self._state
def record_success(self):
if self._state == CircuitState.HALF_OPEN:
logger.info("Circuit breaker transitioning from HALF_OPEN to CLOSED")
self._state = CircuitState.CLOSED
self._failure_count = 0
def record_failure(self):
self._failure_count += 1
self._last_failure_time = time.time()
if self._state == CircuitState.HALF_OPEN:
self._state = CircuitState.OPEN
logger.warning("Circuit breaker OPEN (failure during half-open)")
elif self._failure_count >= self.failure_threshold:
self._state = CircuitState.OPEN
logger.warning(f"Circuit breaker OPEN after {self._failure_count} failures")
def can_execute(self) -> bool:
state = self.state
if state == CircuitState.CLOSED:
return True
if state == CircuitState.HALF_OPEN:
if self._half_open_calls < self.half_open_max_calls:
self._half_open_calls += 1
return True
return False
return False
class ResilientVectorStore:
def __init__(self, vector_store, circuit_breaker: CircuitBreaker):
self._vector_store = vector_store
self._circuit_breaker = circuit_breaker
def query(self, query_embedding, top_k: int = 5):
if not self._circuit_breaker.can_execute():
raise RuntimeError("Circuit breaker is OPEN — service unavailable")
try:
result = self._vector_store.query(query_embedding, top_k=top_k)
self._circuit_breaker.record_success()
return result
except Exception as e:
self._circuit_breaker.record_failure()
raise
def add(self, nodes):
if not self._circuit_breaker.can_execute():
raise RuntimeError("Circuit breaker is OPEN — service unavailable")
try:
self._vector_store.add(nodes)
self._circuit_breaker.record_success()
except Exception as e:
self._circuit_breaker.record_failure()
raise
Pattern 5: Graceful Degradation with Cached Responses
Sometimes the best recovery is to serve a cached or approximate response rather than failing entirely. This pattern is particularly valuable for read-heavy applications where users frequently ask similar questions.
Cache-First Query Strategy
import hashlib
import json
import time
from llama_index.core import VectorStoreIndex
from llama_index.core.query_engine import RetrieverQueryEngine
class CachedQueryEngine:
def __init__(self, query_engine: RetrieverQueryEngine, ttl_seconds: int = 3600):
self._query_engine = query_engine
self._cache = {}
self._ttl = ttl_seconds
def _cache_key(self, query: str) -> str:
return hashlib.sha256(query.lower().strip().encode()).hexdigest()
def query(self, question: str) -> dict:
key = self._cache_key(question)
# Check cache first
if key in self._cache:
entry = self._cache[key]
if time.time() - entry["timestamp"] < self._ttl:
print("[CACHE HIT] Returning cached response")
return {
"response": entry["response"],
"source": "cache",
"cached_at": entry["timestamp"],
}
# Attempt fresh query with fallback to stale cache
try:
response = self._query_engine.query(question)
self._cache[key] = {
"response": str(response),
"timestamp": time.time(),
}
return {
"response": str(response),
"source": "fresh",
"sources": [n.node_id for n in response.source_nodes],
}
except Exception as e:
print(f"[ERROR] Fresh query failed: {e}")
# Fall back to stale cache if available
if key in self._cache:
print("[DEGRADED] Returning stale cached response")
return {
"response": self._cache[key]["response"],
"source": "stale_cache",
"warning": "Response may be outdated due to service error",
}
# No cache available — return a graceful error
return {
"response": "I'm unable to process your request at this time. Please try again later.",
"source": "error_fallback",
"error": str(e),
}
# Usage
index = VectorStoreIndex.from_documents(documents)
base_engine = index.as_query_engine(similarity_top_k=5)
cached_engine = CachedQueryEngine(base_engine, ttl_seconds=1800)
result = cached_engine.query("What are the main topics in my documents?")
print(f"Source: {result['source']}")
print(f"Response: {result['response']}")
Pattern 6: Agent-Level Error Recovery
LlamaIndex agents that use tool-calling can encounter errors when tools fail or when the model produces invalid tool calls. Building recovery logic into the agent loop ensures that tool failures don't crash the entire conversation.
Recovery-Aware Agent Loop
from llama_index.core.agent import ReActAgent
from llama_index.core.tools import FunctionTool
from llama_index.llms.openai import OpenAI
import logging
logger = logging.getLogger(__name__)
def search_database(query: str) -> str:
"""Search the internal database for information."""
# Simulate a flaky database
import random
if random.random() < 0.3:
raise ConnectionError("Database connection timed out")
return f"Database results for: {query}"
def search_web(query: str) -> str:
"""Search the web for information."""
return f"Web results for: {query}"
db_tool = FunctionTool.from_defaults(fn=search_database)
web_tool = FunctionTool.from_defaults(fn=search_web)
class ResilientAgent:
def __init__(self, tools, llm, max_retries: int = 2):
self.agent = ReActAgent.from_tools(tools, llm=llm, verbose=True)
self.max_retries = max_retries
def chat(self, message: str) -> str:
for attempt in range(self.max_retries + 1):
try:
response = self.agent.chat(message)
return str(response)
except Exception as e:
logger.warning(f"Agent attempt {attempt + 1} failed: {e}")
if attempt < self.max_retries:
# Inject error context so the agent can adapt
recovery_message = (
f"The previous action failed with error: {e}. "
f"Please try a different approach or tool."
)
self.agent.chat(recovery_message)
else:
return f"I encountered an error and was unable to complete your request: {e}"
llm = OpenAI(model="gpt-4", temperature=0)
agent = ResilientAgent([db_tool, web_tool], llm, max_retries=2)
response = agent.chat("Search for information about renewable energy trends.")
print(response)
Pattern 7: Data Ingestion Error Recovery
When building indexes from large document collections, individual file failures shouldn't abort the entire ingestion process. A batch-processing pattern with per-document error handling ensures partial success.
Resilient Document Ingestion
from llama_index.core import SimpleDirectoryReader, VectorStoreIndex, Document
from llama_index.core.node_parser import SentenceSplitter
import logging
import traceback
logger = logging.getLogger(__name__)
def ingest_documents_resiliently(
directory_path: str,
chunk_size: int = 1024,
chunk_overlap: int = 20,
) -> tuple:
"""Ingest documents with per-file error recovery.
Returns:
tuple: (successful_documents, failed_files)
"""
reader = SimpleDirectoryReader(directory_path, recursive=True)
file_metadata = reader.input_files
successful_docs = []
failed_files = []
for file_path in file_metadata:
try:
# Load individual file
docs = SimpleDirectoryReader(input_files=[str(file_path)]).load_data()
successful_docs.extend(docs)
logger.info(f"Successfully loaded: {file_path}")
except Exception as e:
logger.error(f"Failed to load {file_path}: {e}")
failed_files.append({
"file": str(file_path),
"error": str(e),
"traceback": traceback.format_exc(),
})
# Parse nodes with error handling
parser = SentenceSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap)
all_nodes = []
parse_failures = []
for doc in successful_docs:
try:
nodes = parser.get_nodes_from_documents([doc])
all_nodes.extend(nodes)
except Exception as e:
logger.error(f"Failed to parse document {doc.id_}: {e}")
parse_failures.append({"doc_id": doc.id_, "error": str(e)})
# Build index from successfully parsed nodes
if all_nodes:
index = VectorStoreIndex(nodes=all_nodes)
logger.info(
f"Index built with {len(all_nodes)} nodes from "
f"{len(successful_docs)} documents. "
f"{len(failed_files)} files failed to load, "
f"{len(parse_failures)} documents failed to parse."
)
return index, {"failed_files": failed_files, "parse_failures": parse_failures}
else:
raise RuntimeError("No documents were successfully loaded and parsed")
# Usage
index, failures = ingest_documents_resiliently("./data")
if failures["failed_files"]:
print(f"Warning: {len(failures['failed_files'])} files failed to load")
for f in failures["failed_files"]:
print(f" - {f['file']}: {f['error']}")
Pattern 8: Comprehensive Error Handling with Callbacks
LlamaIndex's callback system provides a centralized way to monitor and respond to events across your entire pipeline. You can use custom callback handlers to log errors, trigger alerts, and collect metrics.
Custom Error Tracking Callback Handler
from llama_index.core.callbacks import CallbackManager, BaseCallbackHandler
from llama_index.core import Settings
from llama_index.core import VectorStoreIndex
from llama_index.llms.openai import OpenAI
import logging
import time
from collections import defaultdict
logger = logging.getLogger(__name__)
class ErrorTrackingHandler(BaseCallbackHandler):
def __init__(self):
self._event_starts = {}
self._errors = defaultdict(list)
self._event_counts = defaultdict(int)
def on_event_start(self, event_type, payload, **kwargs):
self._event_starts[(event_type, id(payload))] = time.time()
self._event_counts[event_type] += 1
def on_event_end(self, event_type, payload, **kwargs):
key = (event_type, id(payload))
if key in self._event_starts:
duration = time.time() - self._event_starts[key]
if "error" in payload:
self._errors[event_type].append({
"error": payload["error"],
"duration": duration,
"timestamp": time.time(),
})
logger.error(f"Event {event_type} failed: {payload['error']}")
del self._event_starts[key]
def get_error_summary(self) -> dict:
return {
"total_events": dict(self._event_counts),
"errors": {k: len(v) for k, v in self._errors.items()},
"error_details": dict(self._errors),
}
# Wire up the callback handler
error_handler = ErrorTrackingHandler()
callback_manager = CallbackManager([error_handler])
llm = OpenAI(model="gpt-4", temperature=0)
Settings.llm = llm
Settings.callback_manager = callback_manager
# Run queries
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
try:
response = query_engine.query("What is in my data?")
print(response)
except Exception as e:
logger.error(f"Query failed: {e}")
finally:
summary = error_handler.get_error_summary()
print(f"Event summary: {summary['total_events']}")
print(f"Errors by type: {summary['errors']}")
Best Practices for Error Recovery with LlamaIndex
1. Classify Your Errors
Not all errors warrant the same recovery strategy. Transient errors (timeouts, rate limits) should be retried. Permanent errors (authentication failures, invalid input) should fail fast. Partial errors (some documents fail to load) should be logged and skipped. Create an error classification system early in your development process.
2. Set Appropriate Timeouts
Always configure explicit timeouts on your LLM and embedding calls. A request that hangs indefinitely is worse than one that fails quickly, because it consumes resources and blocks downstream operations.
from llama_index.llms.openai import OpenAI
from llama_index.core.embeddings import OpenAIEmbedding
# Set explicit timeouts
llm = OpenAI(model="gpt-4", timeout=30.0, max_retries=0)
embed_model = OpenAIEmbedding(timeout=15.0, max_retries=0)
Notice we set max_retries=0 on the LLM client itself — this prevents the underlying SDK from retrying silently, which would interfere with our custom retry logic.
3. Log with Context
When errors occur, log enough context to reproduce and debug the issue. Include the input prompt, the model used, the retry attempt number, and the full error traceback. Structured logging (JSON format) makes it easier to search and aggregate error data in production.
4. Test Your Recovery Logic
Error recovery code is notoriously under-tested because failures are hard to reproduce. Use mocking and fault injection to simulate failures in your test suite:
import pytest
from unittest.mock import patch, MagicMock
from llama_index.llms.openai import OpenAI
def test_fallback_llm_switches_on_failure():
primary = MagicMock(spec=OpenAI)
primary.complete.side_effect = Exception("Primary provider down")
fallback = MagicMock(spec=OpenAI)
fallback.complete.return_value = MagicMock(text="Fallback response")
from myapp.llm_wrapper import FallbackLLM
resilient = FallbackLLM([primary, fallback])
result = resilient.complete("test prompt")
assert result.text == "Fallback response"
assert primary.complete.call_count == 1
assert fallback.complete.call_count == 1
def test_retry_exhaustion():
from tenacity import retry, stop_after_attempt, wait_fixed
call_count = 0
@retry(stop=stop_after_attempt(3), wait=wait_fixed(0), reraise=True)
def flaky_function():
nonlocal call_count
call_count += 1
raise ConnectionError("Always fails")
with pytest.raises(ConnectionError):
flaky_function()
assert call_count == 3
5. Monitor Recovery Metrics
Track key metrics like retry rates, fallback trigger rates, circuit breaker state transitions, and cache hit rates. These metrics help you understand the health of your dependencies and identify when a provider is degrading before it causes a complete outage.
6. Avoid Retry Storms
When multiple components in your system retry simultaneously, you can create a thundering herd problem that overwhelms the recovering service. Add jitter to your retry delays to spread requests over time:
from tenacity import retry, stop_after_attempt, wait_random_exponential
retry_with_jitter = retry(
stop=stop_after_attempt(5),
wait=wait_random_exponential(multiplier=1, max=60),
reraise=True,
)
7. Design for Idempotency
Ensure that retried operations are idempotent — meaning executing them multiple times has the same effect as executing them once. This is especially important for operations that write to vector stores or external databases. Use deterministic IDs for your documents and nodes so that retries don't create duplicates.
from llama_index.core import Document
# Use deterministic doc_id based on content hash
import hashlib
def create_document_with_stable_id(text: str, metadata: dict = None) -> Document:
content_hash = hashlib.sha256(text.encode()).hexdigest()[:16]
return Document(
text=text,
id_=f"doc_{content_hash}",
metadata=metadata or {},
)
Putting It All Together: A Production-Ready Configuration
Here's a comprehensive example that combines multiple recovery patterns into a single production-ready configuration:
import logging
import hashlib
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings, Document
from llama_index.core.callbacks import CallbackManager
from llama_index.llms.openai import OpenAI
from llama_index.core.embeddings import OpenAIEmbedding
from tenacity import retry, stop_after_attempt, wait_random_exponential
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='{"time": "%(asctime)s", "level": "%(levelname)s", "logger": "%(name)s", "message": "%(message)s"}'
)
logger = logging.getLogger(__name__)
# 1. Configure LLM with explicit timeout and no internal retries
llm = OpenAI(model="gpt-4", temperature=0, timeout=30.0, max_retries=0)
# 2. Configure embedding model with timeout
embed_model = OpenAIEmbedding(model="text-embedding-3-small", timeout=15.0, max_retries=0)
# 3. Set global settings
Settings.llm = llm
Settings.embed_model = embed_model
# 4. Define retry decorator for query operations
query_retry = retry(
stop=stop_after_attempt(4),
wait=wait_random_exponential(multiplier=1, max=30),
reraise=True,
)
# 5. Load documents with per-file error handling
def load_documents_safely(directory: str):
reader = SimpleDirectoryReader(directory, recursive=True)
files = reader.input_files
docs = []
failures = []
for f in files:
try:
file_docs = SimpleDirectoryReader(input_files=[str(f)]).load_data()
for doc in file_docs:
content_hash = hashlib.sha256(doc.text.encode()).hexdigest()[:16]
doc.id_ = f"doc_{content_hash}"
docs.extend(file_docs)
except Exception as e:
failures.append({"file": str(f), "error": str(e)})
logger.warning(f"Skipped {f}: {e}")
logger.info(f"Loaded {len(docs)} documents, {len(failures)} failures")
return docs, failures
# 6. Build index
documents, failures = load_documents_safely("./data")
index = VectorStoreIndex.from_documents(documents, show_progress=True)
# 7. Create query engine with retry wrapper
query_engine = index.as_query_engine(similarity_top_k=5)
@query_retry
def safe_query(question: str) -> str:
response = query_engine.query(question)
return str(response)
# 8. Execute with full error handling
try:
answer = safe_query("What are the main themes in my documents?")
print(f"Answer: {answer}")
except Exception as e:
logger.error(f"Query failed after all recovery attempts: {e}")
print("We're experiencing technical difficulties. Please try again later.")
Conclusion
Error recovery is not an afterthought — it is a core architectural concern for any LLM application that needs to operate reliably in production. By combining retry logic with exponential backoff, fallback providers, self-correcting output parsers, circuit breakers, cached degradation, and agent-level recovery, you can build LlamaIndex applications that withstand the inherent unpredictability of LLM APIs and external services. The key is to layer these patterns thoughtfully: use retries for transient failures, fallbacks for provider outages, circuit breakers for sustained issues, and graceful degradation as a last resort. Start by identifying the failure modes most relevant to your application, implement the corresponding patterns, and continuously refine your recovery logic based on real production telemetry. With these patterns in place, your LlamaIndex application will be well-equipped to handle whatever failures come its way while maintaining a smooth experience for your users.