AI Engineer Portal

Your personal operating system for career transition.

Private mode

Exercise

Implement async batch processing for LLM calls

Implement Async Batch Processing for LLM Calls

Calling an LLM API once takes 500ms–3s. Processing 50 prompts serially takes 25–150 seconds. Async batch processing with a concurrency limiter can reduce that to 5–30 seconds while staying within provider rate limits.

What to build

Implement a BatchProcessor class that:

Accepts a list of prompts and processes them concurrently with a bounded semaphore.
Respects a max_concurrent limit — never fires more than N simultaneous requests.
Applies per-item timeout — each call that exceeds timeout_s raises asyncio.TimeoutError.
Retries transient failures — retry up to max_retries times with exponential backoff + jitter before marking an item as failed.
Returns a structured result — a BatchResult with successes: list[ItemResult] and failures: list[ItemFailure]. A partial failure should NOT abort the batch.
Records latency per item — each ItemResult includes latency_ms.

Constraints

Use asyncio.Semaphore for concurrency limiting.
Use asyncio.wait_for for per-call timeouts.
Use asyncio.gather(..., return_exceptions=True) so one failure does not cancel other tasks.
Type-hint every method.

Python Ai / medium / Step 2 of 6

Practice stage

General drill

Hint

Keep the solution explicit and reviewable.

Next drill

Create a streaming document processor with generators

Success criteria

Make the solution explicit, debuggable, and easy to explain.

Review checklist

Review where the boundary is, what gets validated, and what would be hard to debug later.

from __future__ import annotations
import asyncio
import time
from dataclasses import dataclass, field
from typing import Any, Callable, Awaitable

@dataclass
class ItemResult:
    index: int
    prompt: str
    output: Any
    latency_ms: int

@dataclass
class ItemFailure:
    index: int
    prompt: str
    error: str
    attempts: int

@dataclass
class BatchResult:
    successes: list[ItemResult] = field(default_factory=list)
    failures: list[ItemFailure] = field(default_factory=list)

@property
    def success_rate(self) -> float:
        total = len(self.successes) + len(self.failures)
        return len(self.successes) / total if total > 0 else 0.0

class BatchProcessor:
    def __init__(
        self,
        call_fn: Callable[[str], Awaitable[Any]],
        max_concurrent: int = 5,
        timeout_s: float = 30.0,
        max_retries: int = 2,
    ) -> None:
        self._call_fn = call_fn
        self._max_concurrent = max_concurrent
        self._timeout_s = timeout_s
        self._max_retries = max_retries

async def run(self, prompts: list[str]) -> BatchResult:
        # TODO: implement bounded concurrent processing with retry
        raise NotImplementedError

Practice

Generate a variation

Generate a new exercise variation to deepen understanding or practice a related concept.

Attempt history

Recent submissions

Before you submit, decide what a strong answer should make obvious to the reviewer.

No attempts yet.