Most free-tier LLM APIs—often used when experimenting with AI-generated code or building helper tools—are generous enough for personal projects, but they come with a catch. The moment you share your application with a few friends, or host a public demo, you run face-first into rate limits.
Providers like Google, Groq, and Cohere do not just limit requests per minute. They limit tokens per minute and tokens per day. If a single user decides to paste a massive document into your prompt window, they can exhaust your entire daily quota in seconds. The rest of your users get hit with 429 Too Many Requests errors.
To prevent this, you cannot rely on simple IP-based request rate limiters. You need to track the actual resource being consumed: tokens. You need a token ledger.
The Architecture of a Token Ledger
A basic rate limiter counts requests. A token ledger tracks transactions. Think of it like a bank account. Every user starts with a balance of tokens. When they send a prompt, you reserve a portion of their balance. Once the LLM responds, you calculate the exact cost and settle the transaction.
If you only track token usage after a request completes, your system is vulnerable to race conditions. A user could open ten browser tabs and send ten massive requests simultaneously. Because the ledger only updates when the API responds, all ten requests will pass the quota check. By the time the transactions settle, your API key is blocked, and the user has run up a massive negative balance.
To prevent this, we use a two-phase reservation pattern:
- Reserve: Estimate the cost of the prompt and reserve that amount from the user's quota before making the API call. If the user does not have enough tokens, block the request immediately.
- Commit: Once the API call returns, calculate the actual tokens used (input plus output). Update the ledger with the real number and release the excess reservation.
If the API call fails or times out, the reservation expires, and the tokens return to the user's balance.
Choosing the Data Store
For a production system, Redis is the standard choice. It is fast, runs in-memory (for a deeper look at how different environments handle memory, see our guide on practical memory management), and supports atomic operations through Lua scripting. For smaller setups or self-hosted tools, SQLite is more than enough.
We will design a Redis-backed ledger. We need to track three data points for every user:
- Their current token balance.
- Their active reservations.
- The expiration time of those reservations.
Instead of writing complex lock mechanisms in our application code, we can use a Redis transaction to handle the reservation atomically.
Implementing the Ledger
Let us build a Python implementation of this ledger (and if you are learning these concepts, retyping the LLM-generated code yourself is a great way to build a deeper understanding). First, we need to estimate the input tokens before sending the prompt. We can use the tiktoken library for OpenAI models, or the native token counters provided by other SDKs.
import tiktoken
def estimate_prompt_tokens(prompt: str, model: str = "gpt-4o") -> int:
try:
encoding = tiktoken.encoding_for_model(model)
except KeyError:
encoding = tiktoken.get_encoding("cl100k_base")
# Add a safety margin of 20% for system message formatting overhead
return int(len(encoding.encode(prompt)) * 1.2)Now, let us write the core TokenLedger class using Redis. We will store the user's daily quota in a simple key, and their active reservations in a sorted set where the score is the expiration timestamp. This allows us to easily find and clean up orphaned reservations.
import time
import redis
class TokenLedger:
def __init__(self, redis_client: redis.Redis, daily_limit: int = 100000):
self.r = redis_client
self.daily_limit = daily_limit
# Reservations expire after 60 seconds by default
self.reservation_ttl = 60
def _get_user_keys(self, user_id: str):
return {
"balance": f"user:{user_id}:balance",
"reservations": f"user:{user_id}:reservations"
}
def get_available_balance(self, user_id: str) -> int:
keys = self._get_user_keys(user_id)
# Clean up expired reservations first
now = time.time()
self.r.zremrangebyscore(keys["reservations"], "-inf", now)
# Get current balance
balance_str = self.r.get(keys["balance"])
if balance_str is None:
# Initialize new user with daily limit
self.r.set(keys["balance"], self.daily_limit, ex=86400)
balance = self.daily_limit
else:
balance = int(balance_str)
# Subtract active reservations
reserved = 0
active_res = self.r.zrange(keys["reservations"], 0, -1, withscores=True)
for item, _ in active_res:
_, amount = item.decode().split(":")
reserved += int(amount)
return max(0, balance - reserved)The get_available_balance method calculates how many tokens a user can safely spend right now. It removes any reservations that have timed out, reads the base balance, and subtracts the sum of all active reservations.
Next, we need to implement the reservation step. This must be atomic. If two threads check the balance at the same time, they must not double-spend the same tokens.
def reserve_tokens(self, user_id: str, request_id: str, estimated_amount: int) -> bool:
keys = self._get_user_keys(user_id)
# We run this in a pipeline to minimize round trips
pipe = self.r.pipeline()
while True:
try:
# Watch the balance key for changes to prevent race conditions
pipe.watch(keys["balance"])
available = self.get_available_balance(user_id)
if available < estimated_amount:
pipe.unwatch()
return False
# Start transaction
pipe.multi()
expire_at = time.time() + self.reservation_ttl
reservation_value = f"{request_id}:{estimated_amount}"
pipe.zadd(keys["reservations"], {reservation_value: expire_at})
pipe.execute()
return True
except redis.WatchError:
# The balance changed during our check, retry
continueIf the reservation succeeds, we make our call to the LLM API. Once the API returns a response, we get the exact token usage from the response metadata. We then commit the actual usage and remove the reservation.
def commit_tokens(self, user_id: str, request_id: str, estimated_amount: int, actual_amount: int):
keys = self._get_user_keys(user_id)
reservation_value = f"{request_id}:{estimated_amount}"
pipe = self.r.pipeline()
# Remove the reservation and deduct the actual amount from the balance
pipe.zrem(keys["reservations"], reservation_value)
pipe.decrby(keys["balance"], actual_amount)
pipe.execute()If the API call fails, we must release the reservation without deducting anything from the balance.
def release_reservation(self, user_id: str, request_id: str, estimated_amount: int):
keys = self._get_user_keys(user_id)
reservation_value = f"{request_id}:{estimated_amount}"
self.r.zrem(keys["reservations"], reservation_value)Integrating the Ledger with an LLM Call
Let us look at how this fits into a standard API route. We will wrap our call to OpenAI with the ledger check.
import uuid
from openai import OpenAI
openai_client = OpenAI()
redis_conn = redis.Redis(host="localhost", port=6379, db=0)
ledger = TokenLedger(redis_conn)
def generate_response(user_id: str, prompt: str) -> str:
request_id = str(uuid.uuid4())
# 1. Estimate cost
estimated_cost = estimate_prompt_tokens(prompt)
# Add a buffer for the expected output length (e.g., 500 tokens)
estimated_total = estimated_cost + 500
# 2. Reserve tokens
if not ledger.reserve_tokens(user_id, request_id, estimated_total):
raise Exception("Quota exceeded. Please try again later.")
try:
# 3. Call API
response = openai_client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}]
)
# 4. Extract actual usage
usage = response.usage
actual_total = usage.total_tokens
# 5. Commit actual cost
ledger.commit_tokens(user_id, request_id, estimated_total, actual_total)
return response.choices[0].message.content
except Exception as e:
# If the API call failed, clean up the reservation
ledger.release_reservation(user_id, request_id, estimated_total)
raise eThis pattern ensures you never over-allocate tokens. If the LLM provider experiences a service outage, your users do not lose their quota. The reservation is safely rolled back.
Handling the Streaming Edge Case
Streaming responses make token tracking difficult. When you set stream=True, the API does not return a single usage metadata block at the end. Instead, you receive chunks of text over an open connection.
If you do not track streaming usage, users can bypass your ledger by keeping streams open for long periods.
To handle streaming:
- Estimate the maximum tokens you want to allow for the stream (e.g., 1000 tokens).
- Reserve this maximum amount before starting the stream.
- Count the tokens as they arrive in your application. For every chunk of text, decode it and count the tokens.
- If the stream reaches the reserved limit, terminate the connection early.
- Once the stream ends (either naturally or because you cut it off), commit the actual count and release the remaining reservation.
def generate_streaming_response(user_id: str, prompt: str):
request_id = str(uuid.uuid4())
estimated_input = estimate_prompt_tokens(prompt)
max_output_limit = 1000
total_reservation = estimated_input + max_output_limit
if not ledger.reserve_tokens(user_id, request_id, total_reservation):
yield "Error: Quota exceeded."
return
tokens_consumed = estimated_input
try:
stream = openai_client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
stream=True
)
encoding = tiktoken.get_encoding("cl100k_base")
for chunk in stream:
content = chunk.choices[0].delta.content
if content:
# Count tokens in this chunk
chunk_tokens = len(encoding.encode(content))
tokens_consumed += chunk_tokens
if tokens_consumed >= total_reservation:
yield "\n[Stream cut off: Daily token quota reached]"
break
yield content
# Commit what we actually used
ledger.commit_tokens(user_id, request_id, total_reservation, tokens_consumed)
except Exception as e:
ledger.release_reservation(user_id, request_id, total_reservation)
yield f"Error processing request: {str(e)}"This prevents users from running infinite loops that exhaust your keys while keeping the user experience intact.
Smart Key Rotation
If you are running on free tiers, you might have multiple API keys across different accounts to increase your total daily limit. A token ledger can manage this pool of keys.
Instead of hardcoding a single API key, you can track the quota of each key in Redis. When a user makes a request, the ledger checks which key has enough remaining quota, reserves the tokens against that key, and returns the corresponding key string to your API client.
Once the API call returns (and before you pass it to your application where your parser might throw away its best answers during validation), we get the exact token usage from the response metadata. We then commit the actual usage and remove the reservation.
class KeyRotator:
def __init__(self, redis_client: redis.Redis, api_keys: list[str], key_limit: int = 100000):
self.r = redis_client
self.keys = api_keys
self.key_limit = key_limit
def get_available_key(self, required_tokens: int) -> str:
for idx, key in enumerate(self.keys):
redis_key = f"apikey:{idx}:balance"
balance_str = self.r.get(redis_key)
if balance_str is None:
self.r.set(redis_key, self.key_limit, ex=86400)
balance = self.key_limit
else:
balance = int(balance_str)
if balance >= required_tokens:
# Reserve from this key
self.r.decrby(redis_key, required_tokens)
return key
raise Exception("All API keys have exhausted their daily quota.")By combining user-level quotas and key-level quotas, you build a resilient middleware layer. Your users get a fair distribution of resources, and your system dynamically balances the load across your available free keys without dropping requests.
