跳到内容

RunConfig

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

最大并发工作进程数，默认为 16。

 16
exception_types Union[Type[BaseException], Tuple[Type[BaseException], ...]]

需要捕获并重试的异常类型，默认为 (Exception,)。

 (Exception,)
log_tenacity bool

是否使用 tenacity 记录重试尝试，默认为 False。

 False
seed int

用于可重现性的随机种子，默认为 42。

 42

属性

名称 类型 描述
rng 生成器

使用指定种子初始化的随机数生成器。
备注

__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 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
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)