Asynchronous utilities

Adaptive Harmony provides common utilities for efficient concurrent processing of data. Use these functions to parallelize model inference, grading, and other async operations in your recipes.

`async_map`

Process all items concurrently with automatic progress tracking:

from adaptive_harmony.core.utils import async_map

async def grade_completion(thread: StringThread) -> float:
    return await grader.grade(thread)

# Process all threads in parallel
scores = await async_map(grade_completion, threads)

All tasks run concurrently. Progress is tracked automatically. Parameters:

f — Async function to apply
data — Sequence of items
max_concurrent_samples — Optional int cap on in-flight tasks. Backed by an asyncio.Semaphore. Use this to avoid overwhelming a downstream model or API.
stage_notifier — Optional StageNotifier. When set, progress is reported to the platform as each sample finishes, so the job UI shows live progress.

results = await async_map(
    grader.score_float_value,
    samples,
    max_concurrent_samples=8,
    stage_notifier=ctx.job.stage_notifier("Grading"),
)

`async_map_fallible`

Same as async_map but silently skips failures instead of crashing. Failed samples are excluded from results.

from adaptive_harmony.core.utils import async_map_fallible

# Some threads may fail (e.g., too long for model context)
completions = await async_map_fallible(model.generate, threads)

Use when you expect some samples to fail and want to continue processing the rest. Accepts the same max_concurrent_samples and stage_notifier as async_map. Pass return_indices=True to recover which inputs survived. The result type changes to list[tuple[int, T]] where the int is the index in the original data sequence:

indexed = await async_map_fallible(model.generate, threads, return_indices=True)
for idx, completion in indexed:
    print(f"thread {idx} succeeded")

async_map_batch

Process an iterator in batches with automatic retry on failure:

from adaptive_harmony.core.utils import async_map_batch

# Process batches of 50, retry failed samples up to 50% failure rate
results = await async_map_batch(
    f=model.generate,
    data=iter(threads),
    batch_size=50,
    max_failure_fraction=0.5
)

Parameters:

f - Async function to apply
data - Iterator (not list) of items
batch_size - Number of items per batch
max_failure_fraction - Max fraction of failures before raising exception (default 0.5)

Behavior:

Processes batch_size items concurrently
If a sample fails, pulls next item from iterator and retries
Fails if more than max_failure_fraction * batch_size samples fail
Results are not ordered

Use when:

Working with iterators
Want automatic retry with fresh samples on failure (as in training, where batch size must remain constant)
Don’t need to preserve ordering

Getting Started

Building Blocks

Training & Evaluation

Cookbook

Updates

`async_map`

`async_map_fallible`

async_map_batch

​async_map

​async_map_fallible

​async_map_batch

`async_map`

`async_map_fallible`

async_map_batch