Source code for langchain.schema.runnable.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.schema.runnable.base import Input, Output, RunnableBinding
from langchain.schema.runnable.config import RunnableConfig, patch_config
if TYPE_CHECKING:
    from langchain.callbacks.manager import (
        AsyncCallbackManagerForChainRun,
        CallbackManagerForChainRun,
    )
    T = TypeVar("T", CallbackManagerForChainRun, AsyncCallbackManagerForChainRun)
U = TypeVar("U")
[docs]class RunnableRetry(RunnableBinding[Input, Output]):
    """Retry a Runnable if it fails."""
    retry_exception_types: Tuple[Type[BaseException], ...] = (Exception,)
    wait_exponential_jitter: bool = True
    max_attempt_number: int = 3
    @property
    def _kwargs_retrying(self) -> Dict[str, Any]:
        kwargs: Dict[str, Any] = dict()
        if self.max_attempt_number:
            kwargs["stop"] = stop_after_attempt(self.max_attempt_number)
        if self.wait_exponential_jitter:
            kwargs["wait"] = wait_exponential_jitter()
        if self.retry_exception_types:
            kwargs["retry"] = retry_if_exception_type(self.retry_exception_types)
        return kwargs
    def _sync_retrying(self, **kwargs: Any) -> Retrying: