interruptible_sleep

AsyncFunction

This function is defined in the retry_utils module.

Sleep with periodic interruption checks.

Signature

async def interruptible_sleep(seconds: float, check_interval: float, interrupt_fn: Optional[Callable[[], bool]]) -> bool

float

required

Total sleep duration in seconds

float

default:"0.2"

How often to check for interruption (seconds)

Optional

Function that returns True if sleep should be interrupted

bool

True if completed full sleep, False if interrupted

>>> interrupted = await interruptible_sleep(30.0, interrupt_fn=lambda: agent.is_stopped())

praisonaiagents/agent/retry_utils.py at line 70