Source code for langchain_core.runnables.retry
from typing import (
    TYPE_CHECKING,
    Any,
    Dict,
    List,
    Optional,
    Tuple,
    Type,
    TypeVar,
    Union,
    cast,
)
from tenacity import (
    AsyncRetrying,
    RetryCallState,
    RetryError,
    Retrying,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential_jitter,
)
from langchain_core.runnables.base import Input, Output, RunnableBindingBase
from langchain_core.runnables.config import RunnableConfig, patch_config
if TYPE_CHECKING:
    from langchain_core.callbacks.manager import (
        AsyncCallbackManagerForChainRun,
        CallbackManagerForChainRun,
    )
    T = TypeVar("T", CallbackManagerForChainRun, AsyncCallbackManagerForChainRun)
U = TypeVar("U")
[docs]class RunnableRetry(RunnableBindingBase[Input, Output]):
    """Retry a Runnable if it fails.
    A RunnableRetry helps can be used to add retry logic to any object
    that subclasses the base Runnable.
    Such retries are especially useful for network calls that may fail
    due to transient errors.
    The RunnableRetry is implemented as a RunnableBinding. The easiest
    way to use it is through the `.with_retry()` method on all Runnables.
    Example:
    Here's an example that uses a RunnableLambda to raise an exception
        .. code-block:: python
            import time
            def foo(input) -> None:
                '''Fake function that raises an exception.'''
                raise ValueError("Invoking foo failed. At time {time.time()}")
            runnable = RunnableLambda(foo)
            runnable_with_retries = runnable.with_retry(