Introduction/ Issue:
When building LLM applications, we usually focus on prompts, agents, and workflows. However, one challenge that often gets overlooked is handling temporary API failures. Even if our code is correct, the LLM provider may occasionally return errors such as rate limits (429), request timeouts, or temporary service outages.
In many projects, these temporary failures cause the entire application or workflow to fail, even though simply retrying the request after a short delay would have solved the problem. As developers, we often end up writing the same retry logic repeatedly for different projects.
To avoid this repetitive work and make AI applications more reliable, I created a reusable retry wrapper that automatically retries failed LLM requests using exponential backoff and jitter. Since it is built on LangChain’s BaseChatModel, the same component can be used with different LLM providers without changing the retry logic.
Why we need to do / Cause of the issue:
Temporary API failures are a normal part of working with cloud-based AI services. A request may fail because the provider is experiencing high traffic, the network connection is briefly interrupted, or the application has exceeded the API rate limit.
Some common errors include:
- Rate limit errors (HTTP 429)
- Request timeouts
- Temporary connection failures
- Service unavailable (HTTP 503)
- Gateway timeout (HTTP 504)
The important thing is that these errors are usually temporary. In most cases, waiting a few seconds and sending the same request again is enough for it to succeed.
Without a retry mechanism, the impact can be significant:
- AI workflows stop unexpectedly.
- Long-running document processing jobs fail midway.
- Users have to retry operations manually.
- The same retry logic gets implemented repeatedly across different projects.
A reusable retry component solves these problems by handling temporary failures automatically. Instead of failing immediately, it waits for a short period, retries the request, and only reports an error if all retry attempts are exhausted.
To make retries more efficient, the component uses exponential backoff, where the waiting time increases after each failed attempt, along with a small amount of random jitter to prevent multiple requests from retrying at the exact same moment.
How do we solve:
To make our LLM applications more reliable, we’ll create a reusable retry wrapper that can be used with any LangChain chat model. Once created, this component can simply be imported into future projects instead of rewriting the retry logic every time.
Follow the steps below to get started.
Step 1: Install the Required Libraries
Install the required dependencies using pip:
pip install langchain langchain-core langchain-groq python-dotenv
Step 2: Store Your API Key
Create a .env file in your project directory and store your API key securely.
Step 3: Add the Reusable Retry Wrapper
Create a new Python file named ai_retry_wrapper.py and paste the reusable retry wrapper code provided in this blog.
from __future__ import annotations
import random
import time
from typing import Any, Optional
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import BaseMessage, HumanMessage
class RetryExhaustedError(Exception):
“””
Raised when the LLM call still fails after all retry attempts.
The original exception is stored in `last_exception`
so applications can inspect the root cause if needed.
“””
def __init__(self, message: str, last_exception: Exception):
super().__init__(message)
self.last_exception = last_exception
# Common error keywords returned by most LLM providers.
# Instead of importing provider-specific exception classes,
# we detect retryable failures using these markers.
RETRYABLE_MARKERS = (
“rate limit”,
“429”,
“timeout”,
“timed out”,
“connection”,
“temporarily unavailable”,
“service unavailable”,
“502”,
“503”,
“504”,
“overloaded”,
)
def _is_retryable(error: Exception) -> bool:
“””
Determines whether an exception should be retried.
Parameters
———-
error : Exception
Exception raised by the LLM provider.
Returns
——-
bool
True if the exception represents a temporary failure.
“””
message = str(error).lower()
return any(marker in message for marker in RETRYABLE_MARKERS)
class RetryWrapper:
“””
A reusable wrapper that adds automatic retry capability to
any LangChain chat model.
Features
——–
– Automatic retries
– Exponential backoff
– Random jitter
– Supports synchronous and asynchronous calls
Compatible with:
– ChatGroq
– ChatOpenAI
– ChatAnthropic
– AzureChatOpenAI
– Any BaseChatModel implementation
“””
def __init__(
self,
llm: BaseChatModel,
max_retries: int = 4,
base_delay: float = 1.0,
max_delay: float = 30.0,
jitter: float = 0.5,
verbose: bool = False,
):
“””
Initializes the retry wrapper.
Parameters
———-
llm : BaseChatModel
LangChain chat model instance.
max_retries : int
Maximum retry attempts before giving up.
base_delay : float
Initial waiting time before the first retry.
max_delay : float
Maximum delay allowed during exponential backoff.
jitter : float
Random delay added to prevent simultaneous retries.
verbose : bool
Prints retry logs when enabled.
“””
self.llm = llm
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.jitter = jitter
self.verbose = verbose
def _log(self, message: str) -> None:
“””
Prints retry information when verbose mode is enabled.
“””
if self.verbose:
print(f”[RetryWrapper] {message}”)
def _backoff_delay(self, attempt: int) -> float:
“””
Calculates exponential backoff delay.
Formula:
base_delay × 2^(attempt-1)
The delay is capped using max_delay and a small random
jitter is added to avoid retry spikes.
“””
delay = min(
self.base_delay * (2 ** (attempt – 1)),
self.max_delay,
)
delay += random.uniform(0, self.jitter)
return delay
def invoke(
self,
prompt: str | list[BaseMessage],
**kwargs: Any,
) -> BaseMessage:
“””
Sends a synchronous request to the LLM with retry support.
Parameters
———-
prompt : str | list[BaseMessage]
Either a plain prompt string or LangChain messages.
Returns
——-
BaseMessage
Response returned by the LLM.
“””
# Convert plain strings into LangChain HumanMessage objects.
messages = (
[HumanMessage(content=prompt)]
if isinstance(prompt, str)
else prompt
)
last_exception: Optional[Exception] = None
for attempt in range(1, self.max_retries + 1):
try:
self._log(f”Attempt {attempt}/{self.max_retries}”)
return self.llm.invoke(messages, **kwargs)
except Exception as error:
last_exception = error
# Validation errors or authentication issues
# should fail immediately instead of retrying.
if not _is_retryable(error):
self._log(
f”Non-retryable error: {error}”
)
raise
if attempt == self.max_retries:
self._log(
f”Final retry failed: {error}”
)
break
delay = self._backoff_delay(attempt)
self._log(
f”Retryable error detected.”
f” Retrying in {delay:.2f} seconds…”
)
time.sleep(delay)
raise RetryExhaustedError(
f”LLM call failed after {self.max_retries} attempts.”,
last_exception,
)
async def ainvoke(
self,
prompt: str | list[BaseMessage],
**kwargs: Any,
) -> BaseMessage:
“””
Asynchronous version of invoke().
Uses the same retry strategy while awaiting the
underlying async LangChain model.
“””
import asyncio
messages = (
[HumanMessage(content=prompt)]
if isinstance(prompt, str)
else prompt
)
last_exception: Optional[Exception] = None
for attempt in range(1, self.max_retries + 1):
try:
self._log(
f”Async attempt {attempt}/{self.max_retries}”
)
return await self.llm.ainvoke(
messages,
**kwargs,
)
except Exception as error:
last_exception = error
if not _is_retryable(error):
self._log(
f”Non-retryable error: {error}”
)
raise
if attempt == self.max_retries:
self._log(
f”Final retry failed: {error}”
)
break
delay = self._backoff_delay(attempt)
self._log(
f”Retryable error detected.”
f” Retrying in {delay:.2f} seconds…”
)
await asyncio.sleep(delay)
raise RetryExhaustedError(
f”LLM call failed after {self.max_retries} attempts.”,
last_exception,
)
if __name__ == “__main__”:
import os
from dotenv import load_dotenv
from langchain_groq import ChatGroq
# Load environment variables from .env
load_dotenv()
llm = ChatGroq(
model=”llama-3.3-70b-versatile”,
api_key=os.getenv(“GROQ_API_KEY”),
temperature=0,
)
wrapper = RetryWrapper(
llm=llm,
max_retries=4,
base_delay=1,
verbose=True,
)
response = wrapper.invoke(
“Explain why retries are important in AI applications in one sentence.”
)
print(“\nLLM Response:”)
print(response.content)
Step 4: Run the Component
Execute the file to verify that the wrapper is working correctly.
python ai_retry_wrapper.py
Conclusion:
Building an LLM application is not just about generating responses—it is also about making the application reliable enough to handle real-world situations. Temporary API failures are unavoidable, but they should not interrupt the user experience or bring an entire workflow to a halt.
By creating a reusable retry wrapper, we move all retry logic into a single, reusable component that can be plugged into any LangChain-based application. This reduces code duplication, follows industry best practices with exponential backoff and jitter, and makes AI applications more resilient and easier to maintain.
Instead of writing retry logic for every new project, we can simply reuse this component and focus on building better AI solutions, knowing that temporary failures are handled automatically in the background.
