Ragas Ragas
stable · 中文译文
中文译文 · 原文：https://docs.ragas.io/en/stable/references/run_config/ · 许可证 Apache-2.0

RunConfig

RunConfig(timeout: int = 180, max_retries: int = 10, max_wait: int = 60, max_workers: int = 16, exception_types: Union[Type[BaseException], Tuple[Type[BaseException], ...]] = (Exception,), log_tenacity: bool = False, seed: int = 42)

Ragas 操作的超时、重试与随机种子配置。

参数：

名称 类型 说明 默认值
timeout int 单次操作的最长等待时间（秒），默认 180。 180
max_retries int 最大重试次数，默认 10。 10
max_wait int 两次重试之间的最长等待时间（秒），默认 60。 60
max_workers int 最大并发 worker 数，默认 16。 16
exception_types Union[Type[BaseException], Tuple[Type[BaseException], ...]] 需要捕获并重试的异常类型，默认 (Exception,)。 (Exception,)
log_tenacity bool 是否使用 tenacity 记录重试尝试，默认 False。 False
seed int 用于可复现性的随机种子，默认 42。 42

属性：

名称 类型 说明
rng Generator 使用指定种子初始化的随机数生成器。

说明

__post_init__ 方法会使用指定种子将 rng 属性初始化为 numpy 随机数生成器。

add_retry

add_retry(fn: WrappedFn, run_config: RunConfig) -> WrappedFn

使用给定的 RunConfig，为函数添加重试能力。

该函数用 tenacity 库包装输入函数，并根据 RunConfig 中的设置配置重试行为。

说明

  • 若 RunConfig 中启用了 log_tenacity，则会为重试尝试配置日志。
  • 重试逻辑使用带随机抖动的指数退避来计算等待时间。
  • 重试次数以及要重试的异常类型根据 RunConfig 配置。

源代码位于 src/ragas/run_config.py

def add_retry(fn: WrappedFn, run_config: RunConfig) -> WrappedFn:
    """
    Adds retry functionality to a given function using the provided RunConfig.

    This function wraps the input function with retry logic using the tenacity library.
    It configures the retry behavior based on the settings in the RunConfig.

    Notes
    -----
    - If log_tenacity is enabled in the RunConfig, it sets up logging for retry attempts.
    - The retry logic uses exponential backoff with random jitter for wait times.
    - The number of retry attempts and exception types to retry on are configured
      based on the RunConfig.
    """
    # configure tenacity's after section wtih logger
    if run_config.log_tenacity is not None:
        logger = logging.getLogger(f"ragas.retry.{fn.__name__}")
        tenacity_logger = after_log(logger, logging.DEBUG)
    else:
        tenacity_logger = after_nothing

    r = Retrying(
        wait=wait_random_exponential(multiplier=1, max=run_config.max_wait),
        stop=stop_after_attempt(run_config.max_retries),
        retry=retry_if_exception_type(run_config.exception_types),
        reraise=True,
        after=tenacity_logger,
    )
    return r.wraps(fn)

add_async_retry

add_async_retry(fn: WrappedFn, run_config: RunConfig) -> WrappedFn

在函数失败时进行重试的装饰器。

源代码位于 src/ragas/run_config.py

def add_async_retry(fn: WrappedFn, run_config: RunConfig) -> WrappedFn:
    """
    Decorator for retrying a function if it fails.
    """
    # configure tenacity's after section wtih logger
    if run_config.log_tenacity is not None:
        logger = logging.getLogger(f"TENACITYRetry[{fn.__name__}]")
        tenacity_logger = after_log(logger, logging.DEBUG)
    else:
        tenacity_logger = after_nothing

    r = AsyncRetrying(
        wait=wait_random_exponential(multiplier=1, max=run_config.max_wait),
        stop=stop_after_attempt(run_config.max_retries),
        retry=retry_if_exception_type(run_config.exception_types),
        reraise=True,
        after=tenacity_logger,
    )
    return r.wraps(fn)
← TokenizersExecutor →