create_base_retry_decorator() — langchain Function Reference

Architecture documentation for the create_base_retry_decorator() function in llms.py from the langchain codebase.

Function python LangChainCore LanguageModelBase calls 1

Entity Profile

LangChainCore→ LanguageModelBase→ create_base_retry_decorator() — langchain Function Reference

Dependency Diagram

graph TD
  eca3f969_601a_ea54_6966_ed46d1d6b461["create_base_retry_decorator()"]
  a4692bf1_369d_4673_b1eb_6b9a8cbb9994["llms.py"]
  eca3f969_601a_ea54_6966_ed46d1d6b461 -->|defined in| a4692bf1_369d_4673_b1eb_6b9a8cbb9994
  b2f4f0d8_9586_a5f0_0e59_fb3a00a9c3ca["_log_error_once()"]
  eca3f969_601a_ea54_6966_ed46d1d6b461 -->|calls| b2f4f0d8_9586_a5f0_0e59_fb3a00a9c3ca
  style eca3f969_601a_ea54_6966_ed46d1d6b461 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/language_models/llms.py lines 73–128

def create_base_retry_decorator(
    error_types: list[type[BaseException]],
    max_retries: int = 1,
    run_manager: AsyncCallbackManagerForLLMRun | CallbackManagerForLLMRun | None = None,
) -> Callable[[Any], Any]:
    """Create a retry decorator for a given LLM and provided a list of error types.

    Args:
        error_types: List of error types to retry on.
        max_retries: Number of retries.
        run_manager: Callback manager for the run.

    Returns:
        A retry decorator.

    Raises:
        ValueError: If the cache is not set and cache is True.
    """
    logging_ = before_sleep_log(logger, logging.WARNING)

    def _before_sleep(retry_state: RetryCallState) -> None:
        logging_(retry_state)
        if run_manager:
            if isinstance(run_manager, AsyncCallbackManagerForLLMRun):
                coro = run_manager.on_retry(retry_state)
                try:
                    try:
                        loop = asyncio.get_event_loop()
                    except RuntimeError:
                        asyncio.run(coro)
                    else:
                        if loop.is_running():
                            task = loop.create_task(coro)
                            _background_tasks.add(task)
                            task.add_done_callback(_background_tasks.discard)
                        else:
                            asyncio.run(coro)
                except Exception as e:
                    _log_error_once(f"Error in on_retry: {e}")
            else:
                run_manager.on_retry(retry_state)

    min_seconds = 4
    max_seconds = 10
    # Wait 2^x * 1 second between each retry starting with
    # 4 seconds, then up to 10 seconds, then 10 seconds afterwards
    retry_instance: retry_base = retry_if_exception_type(error_types[0])
    for error in error_types[1:]:
        retry_instance |= retry_if_exception_type(error)
    return retry(
        reraise=True,
        stop=stop_after_attempt(max_retries),
        wait=wait_exponential(multiplier=1, min=min_seconds, max=max_seconds),
        retry=retry_instance,
        before_sleep=_before_sleep,
    )