api rate limit
**API Rate Limiting**
**Why Rate Limiting?**
Protect services from abuse, ensure fair usage, manage costs, and maintain system stability.
**Rate Limiting Strategies**
**Token Bucket**
```python
class TokenBucket:
def __init__(self, capacity, refill_rate):
self.capacity = capacity
self.tokens = capacity
self.refill_rate = refill_rate # tokens per second
self.last_refill = time.time()
def consume(self, tokens=1):
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
def _refill(self):
now = time.time()
refill = (now - self.last_refill) * self.refill_rate
self.tokens = min(self.capacity, self.tokens + refill)
self.last_refill = now
```
**Sliding Window**
```python
class SlidingWindowRateLimiter:
def __init__(self, max_requests, window_seconds):
self.max_requests = max_requests
self.window = window_seconds
self.requests = {} # user_id -> list of timestamps
def is_allowed(self, user_id):
now = time.time()
cutoff = now - self.window
# Remove old requests
self.requests[user_id] = [
t for t in self.requests.get(user_id, [])
if t > cutoff
]
if len(self.requests[user_id]) < self.max_requests:
self.requests[user_id].append(now)
return True
return False
```
**Comparison**
| Algorithm | Burst Handling | Memory | Accuracy |
|-----------|----------------|--------|----------|
| Fixed window | Poor | Low | Low |
| Sliding window | Good | Medium | High |
| Token bucket | Good | Low | High |
| Leaky bucket | Smooth | Low | High |
**Implementation Levels**
| Level | Location | Scope |
|-------|----------|-------|
| API Gateway | Infrastructure | Global |
| Application | Code | Per-endpoint |
| Database | Connection pool | Resource |
**LLM API Specific Limits**
| Limit Type | Example |
|------------|---------|
| Requests per minute | 60 RPM |
| Tokens per minute | 100,000 TPM |
| Concurrent requests | 10 |
| Daily quota | 1M tokens/day |
**Handling Rate Limits**
```python
async def call_with_retry(request):
for attempt in range(max_retries):
try:
return await api.call(request)
except RateLimitError as e:
wait_time = e.retry_after or (2 ** attempt)
await asyncio.sleep(wait_time)
raise MaxRetriesExceeded()
```
**Best Practices**
- Use exponential backoff for retries
- Show remaining quota in response headers
- Implement tiered limits (free vs paid)
- Queue requests during limit