run_sequence
Run a list of steps sequentially, piping each output into the next.
from coevolved.base import run_sequence
out = run_sequence([a, b, c], state)
run_first_success
Run steps in order until one succeeds. If all fail, the last error is raised.
from coevolved.base import run_first_success
out = run_first_success([primary, fallback], state)
run_parallel
Run steps concurrently (thread pool). Each step receives the same input state.
from coevolved.base import run_parallel
results = run_parallel([a, b, c], state)
Only use run_parallel with steps that don’t share mutable state and don’t rely on global mutable state.
RetryPolicy
Dataclass describing retry settings:
attemptsbackoff_secondsbackoff_multiplierretry_exceptions
with_retry
Wrap a step with retry logic:
from coevolved.base import RetryPolicy, with_retry
safe = with_retry(unreliable, RetryPolicy(attempts=3))
out = safe(state)
compile_steps
Compile a sequence into a lightweight graph representation (nodes + edges + metadata).
This is useful for debugging and visualization.
Next steps
