Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.adaptive-ml.com/llms.txt

Use this file to discover all available pages before exploring further.

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.

​
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.

​
batch_process_fallible

Process items in controlled batches with failure handling: 
from adaptive_harmony.core.utils import batch_process_fallible

# Process 100 threads at a time
results = await batch_process_fallible(
    fn=model.generate,
    data=threads,
    batch_size=100,
    stage_notifier=ctx.job.stage_notifier("Inference")
)
Returns list[tuple[int, T]] where the int is the original index in the input data. Use this to track which samples succeeded. When to use:
  • Large datasets where running all at once would exceed memory
  • Want to report progress in batches
  • Need to preserve sample indices

​
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