AI Engineer Portal

Your personal operating system for career transition.

Private mode

Exercise

Build a rate limiter for AI API endpoints

Build a Rate Limiter for AI API Endpoints

LLM API calls are expensive and slow. Without rate limiting at your application layer, a single misbehaving client can exhaust your provider token budget or trigger provider-side rate limits that degrade service for everyone.

You need two rate limiting strategies working together:

Per-user RPM (requests per minute): limit how many requests a single user can make per minute.
Global TPM (tokens per minute): limit the total tokens consumed across all users per minute to stay within your provider quota.

What to build

Implement a TwoLayerRateLimiter that:

check_user(user_id: str) -> RateLimitResult — checks if the user is within their per-user RPM limit.
check_global(estimated_tokens: int) -> RateLimitResult — checks if the global TPM budget has headroom.
record_usage(user_id: str, tokens_used: int) -> None — records actual usage after a successful call.
Uses a sliding window algorithm (1-minute window) for both limits.
RateLimitResult has: allowed: bool, reason: str, retry_after_seconds: float.

Constraints

Standard library only (use time.monotonic()).
Per-user default limit: 10 RPM.
Global default: 100,000 TPM.
Sliding window: deque of timestamps for RPM, deque of (timestamp, tokens) for TPM.

Api Async / medium / Step 11 of 23

Practice stage

Async and provider control

Hint

Make waiting behavior explicit. Timeouts, retries, and concurrency limits matter more than squeezing everything into one helper.

Next drill

Implement graceful degradation with fallback tiers

Success criteria

- Uses async boundaries coherently
- Makes timeout and retry decisions legible
- Would be maintainable under provider instability

Review checklist

- Is timeout behavior explicit?
- Is retryable failure separate from terminal failure?
- Would logs reveal what actually timed out?

from __future__ import annotations
import time
from collections import defaultdict, deque
from dataclasses import dataclass

@dataclass
class RateLimitResult:
    allowed: bool
    reason: str = ""
    retry_after_seconds: float = 0.0

class TwoLayerRateLimiter:
    WINDOW_SECONDS = 60.0

def __init__(self, user_rpm_limit: int = 10, global_tpm_limit: int = 100_000):
        self.user_rpm_limit = user_rpm_limit
        self.global_tpm_limit = global_tpm_limit
        # TODO: initialize per-user request windows and global token window
        raise NotImplementedError

def check_user(self, user_id: str) -> RateLimitResult:
        # TODO: sliding window check for user RPM
        raise NotImplementedError

def check_global(self, estimated_tokens: int) -> RateLimitResult:
        # TODO: sliding window check for global TPM
        raise NotImplementedError

def record_usage(self, user_id: str, tokens_used: int) -> None:
        # TODO: add request timestamp to user window and tokens to global window
        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.