Utils Module

Utility helpers for the research pipeline.

This module currently provides a small set of runtime helpers that are reused across pipeline stages. The public API is intentionally minimal and stable.

Example:

from agent.pipeline.utils import retry_async

async def fetch():
    # some flaky network call
    return 42

result = await retry_async(lambda: fetch(), attempts=3, base_delay=1.0)
assert result == 42
async agent.pipeline.utils.retry_async(func, *, attempts=3, base_delay=5.0, factor=2.0)[source]

Retry an async operation with exponential backoff.

Parameters:

  • func (Callable[[], Awaitable[TypeVar(T)]]) – Zero-argument coroutine factory to call on each attempt. Using a factory defers creation of the coroutine until it is awaited, avoiding “already awaited” errors on retries.

  • attempts (int) – Total attempts including the first call (>= 1). Default 3.

  • base_delay (float) – Initial delay in seconds before the next attempt. Default 5.0.

  • factor (float) – Multiplicative backoff factor after each failure. Default 2.0.

Return type:

TypeVar(T)

Returns:

The value returned by the successful call to func.

Raises:

Exception – Re-raises the last exception encountered if all attempts fail.

Example:

async def get_value() -> int:
    return 7

value = await retry_async(lambda: get_value(), attempts=5, base_delay=0.2)
assert value == 7