with_retry() — langchain Function Reference

Architecture documentation for the with_retry() function in base.py from the langchain codebase.

Function python LangChainCore Runnables calls 1 called by 1

Entity Profile

LangChainCore→ Runnables→ with_retry() — langchain Function Reference

Dependency Diagram

graph TD
  3a6d006a_a650_9272_b916_709c48148361["with_retry()"]
  4a62481c_02cb_a5de_1833_50669d5351a6["Runnable"]
  3a6d006a_a650_9272_b916_709c48148361 -->|defined in| 4a62481c_02cb_a5de_1833_50669d5351a6
  bd48bfb2_deef_cf9e_749e_3028362225a1["with_retry()"]
  bd48bfb2_deef_cf9e_749e_3028362225a1 -->|calls| 3a6d006a_a650_9272_b916_709c48148361
  bd48bfb2_deef_cf9e_749e_3028362225a1["with_retry()"]
  3a6d006a_a650_9272_b916_709c48148361 -->|calls| bd48bfb2_deef_cf9e_749e_3028362225a1
  style 3a6d006a_a650_9272_b916_709c48148361 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/base.py lines 1860–1922

    def with_retry(
        self,
        *,
        retry_if_exception_type: tuple[type[BaseException], ...] = (Exception,),
        wait_exponential_jitter: bool = True,
        exponential_jitter_params: ExponentialJitterParams | None = None,
        stop_after_attempt: int = 3,
    ) -> Runnable[Input, Output]:
        """Create a new `Runnable` that retries the original `Runnable` on exceptions.

        Args:
            retry_if_exception_type: A tuple of exception types to retry on.
            wait_exponential_jitter: Whether to add jitter to the wait
                time between retries.
            stop_after_attempt: The maximum number of attempts to make before
                giving up.
            exponential_jitter_params: Parameters for
                `tenacity.wait_exponential_jitter`. Namely: `initial`, `max`,
                `exp_base`, and `jitter` (all `float` values).

        Returns:
            A new `Runnable` that retries the original `Runnable` on exceptions.

        Example:
            ```python
            from langchain_core.runnables import RunnableLambda

            count = 0


            def _lambda(x: int) -> None:
                global count
                count = count + 1
                if x == 1:
                    raise ValueError("x is 1")
                else:
                    pass


            runnable = RunnableLambda(_lambda)
            try:
                runnable.with_retry(
                    stop_after_attempt=2,
                    retry_if_exception_type=(ValueError,),
                ).invoke(1)
            except ValueError:
                pass

            assert count == 2
            ```
        """
        # Import locally to prevent circular import
        from langchain_core.runnables.retry import RunnableRetry  # noqa: PLC0415

        return RunnableRetry(
            bound=self,
            kwargs={},
            config={},
            retry_exception_types=retry_if_exception_type,
            wait_exponential_jitter=wait_exponential_jitter,
            max_attempt_number=stop_after_attempt,
            exponential_jitter_params=exponential_jitter_params,
        )