AI Engineer Portal

Your personal operating system for career transition.

Private mode

Exercise

Implement batch processing with rate limiting

Implement Batch Processing with Rate Limiting

Provider APIs enforce rate limits in two dimensions: requests per minute (RPM) and tokens per minute (TPM). Exceeding either causes 429 errors. A naive batch processor that fires all requests immediately will hit these limits before reaching even 10% of most corpora.

What to build

Implement a RateLimitedBatcher that processes items in batches while respecting both request and token rate limits:

__init__(rpm_limit: int, tpm_limit: int, batch_size: int) — stores limits.
process(items: list[dict], process_fn: Callable[[list[dict]], list[dict]]) -> BatchStats — processes all items in batches. Each item has "tokens": int. Tracks: total requests made, total tokens sent, total elapsed time, and rate limit waits.
Before each batch, check if sending it would exceed either limit within the current 60-second window. If so, sleep until the window resets.
BatchStats dataclass: total_items, total_batches, total_tokens, elapsed_s, rate_limit_waits, items_per_second.

The rate limiting logic: track how many requests and tokens were sent in the last 60 seconds. If the next batch would exceed either limit, compute how long to sleep until the window clears.

Why this matters

Rate limit handling is one of the most common sources of production failures in AI pipelines. Engineers often add naive time.sleep(60) calls that make pipelines 10x slower than necessary. A proper token-aware rate limiter sleeps only when needed and for only as long as required.

Constraints

Use time.time() for rate limit window tracking (not asyncio).
Do not sleep for more than 60 seconds at a time.
Test with a mock process_fn that does not actually call any API.
items_per_second in BatchStats should be rounded to 2 decimal places.

Data Transformation / hard / Step 11 of 16

Practice stage

Transform and summarize data safely

Hint

Think like an evaluation or trace pipeline: group clearly, preserve the data story, and make edge cases visible instead of clever.

Next drill

Build an error classification and retry policy engine

Success criteria

- Handles missing or noisy records predictably
- Produces summary output that is easy to inspect
- Would be safe to rerun in a script workflow

Review checklist

- Do grouped metrics stay readable?
- Would malformed rows be debuggable?
- Did I choose names that explain the transformation?

from __future__ import annotations
import time
import logging
from dataclasses import dataclass, field
from typing import Callable

logger = logging.getLogger(__name__)

@dataclass
class BatchStats:
    total_items: int
    total_batches: int
    total_tokens: int
    elapsed_s: float
    rate_limit_waits: int
    items_per_second: float

class RateLimitedBatcher:
    WINDOW_S = 60.0

def __init__(self, rpm_limit: int, tpm_limit: int, batch_size: int) -> None:
        self.rpm_limit = rpm_limit
        self.tpm_limit = tpm_limit
        self.batch_size = batch_size

def process(
        self,
        items: list[dict],
        process_fn: Callable[[list[dict]], list[dict]],
    ) -> BatchStats:
        # TODO: process items in batches with rate limit enforcement
        raise NotImplementedError

def _check_rate_limits(
        self,
        window_requests: list[float],
        window_tokens: list[tuple[float, int]],
        batch_tokens: int,
        now: float,
    ) -> float:
        # TODO: return seconds to sleep (0.0 if no sleep needed)
        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.