agentscope.rpc.retry_strategy module

Timeout retry strategies

class agentscope.rpc.retry_strategy.RetryBase

Bases: ABC

The base class for all retry strategies

abstract retry(func: Callable, *args: Any, **kwargs: Any) → Any

Retry the func when any exception occurs

classmethod load_dict(data: dict) → RetryBase

Load the retry strategy from a dict

class agentscope.rpc.retry_strategy.RetryFixedTimes(max_retries: int = 10, delay: float = 5)

Bases: RetryBase

Retry a fixed number of times, and wait a fixed delay time between each attempt.

Init dict format:

• type: ‘fixed’

• max_retries (int): The max retry times

• delay (float): The delay time between each attempt

retry = RetryBase.load_dict({
    "type": "fixed",
    "max_retries": 10,
    "delay": 5,
})

__init__(max_retries: int = 10, delay: float = 5) → None

Initialize the retry strategy

Parameters:

• max_retries (int) – The max retry times

• delay (float) – The delay time between each attempt

retry(func: Callable, *args: Any, **kwargs: Any) → Any

Retry the func when any exception occurs

class agentscope.rpc.retry_strategy.RetryExpential(max_retries: int = 10, base_delay: float = 5, max_delay: float = 300)

Bases: RetryBase

Retry with exponential backoff, which means the delay time will increase exponentially.

Init dict format:

• type: ‘expential’

• max_retries (int): The max retry times

• base_delay (float): The base delay time

• max_delay (float): The max delay time, which will be used if the calculated delay time

• exceeds it.

retry = RetryBase.load_dict({
    "type": "expential",
    "max_retries": 10,
    "base_delay": 5,
    "max_delay": 300,
})

__init__(max_retries: int = 10, base_delay: float = 5, max_delay: float = 300) → None

Initialize the retry strategy

Parameters:

• max_retries (int) – The max retry times

• base_delay (float) – The base delay time

• max_delay (float) – The max delay time

retry(func: Callable, *args: Any, **kwargs: Any) → Any

Retry the func when any exception occurs