You've built something that works. Your internal LLM-powered tool performs beautifully — it summarizes contracts, answers customer questions, generates reports. Now leadership wants to productize it. You'll serve thirty enterprise customers, each with their own data, their own users, their own expectations about privacy, and their own budget ceilings. Congratulations: you're now building a multi-tenant LLM platform, and the architecture decisions you make in the next few weeks will either scale gracefully to a hundred customers or quietly collapse under the weight of the second one.
The naive approach is to stand up one API endpoint, share one system prompt, and send everyone's requests through the same pipeline. This breaks almost immediately. Tenant A's conversation history starts bleeding context into Tenant B's completions. A single customer running a batch job eats the entire organization's rate limit. Your monthly OpenAI bill arrives and you have no idea which customer cost you $4,000 and which cost $12. Multi-tenancy for LLM platforms is a genuinely hard systems problem, and most of the existing literature treats it as an afterthought or assumes you're already running Kubernetes at scale. This lesson meets you in the middle: production-grade patterns, real code, and honest discussion of trade-offs.
By the end of this lesson, you'll be able to design and implement a multi-tenant LLM platform that isolates tenant context at the prompt and session level, enforces per-tenant token quotas with graceful degradation, tracks costs with enough granularity to produce customer-facing invoices, and handles the failure modes that kill multi-tenant systems in production.
What you'll learn:
This lesson assumes you are comfortable with:
You do not need prior experience with multi-tenancy specifically. That's the point of this lesson.
Multi-tenancy in a traditional web application is largely a data isolation problem. You add a tenant_id column to your tables, filter every query, and you're most of the way there. LLMs introduce three dimensions that make this substantially harder.
First, the context is stateful in ways databases aren't. A database row is inert. A conversation context is cumulative and deterministic in its influence — every token in the prompt window shapes the model's output. If Tenant A's system prompt or conversation history leaks into a request made on behalf of Tenant B, the contamination is semantic, not just a data exposure. Tenant B's outputs become wrong in ways that may not be immediately obvious.
Second, consumption is non-deterministic and bursty. You cannot predict in advance how many tokens a user's query will consume because the model's response length is probabilistic. A customer asking a simple question might get a one-sentence answer or, if something in the prompt encourages verbosity, a five-paragraph essay. This makes quota enforcement fundamentally different from, say, rate-limiting API calls, where you know exactly how much resource each call consumes.
Third, the cost model is external and you pay it before you can bill for it. You get charged by your LLM provider immediately when you make an API call. If your quota enforcement has a race condition and lets ten concurrent requests through simultaneously for a customer who's hit their limit, you eat those costs. The billing problem is also attribution: a single API call might serve a workflow that spans multiple customers' data (think: a summarization job that touches a shared knowledge base), and allocating that cost requires deliberate design.
Let's build a system that handles all three dimensions cleanly.
Before writing any request-routing code, get the data model right. You need to represent four things: tenants, their quota configuration, their usage state, and their cost records.
# models.py
from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal
from enum import Enum
from typing import Optional
import uuid
class QuotaPeriod(Enum):
HOURLY = "hourly"
DAILY = "daily"
MONTHLY = "monthly"
class QuotaAction(Enum):
REJECT = "reject" # Hard stop — return 429
THROTTLE = "throttle" # Slow requests via queue
NOTIFY = "notify" # Allow but fire an alert webhook
ALLOW = "allow" # Soft limit — log only
@dataclass
class TenantQuotaPolicy:
"""
Defines what a tenant is allowed to consume and what happens
when they exceed it. A tenant can have multiple policies
(e.g., a daily soft limit and a monthly hard limit).
"""
tenant_id: str
period: QuotaPeriod
token_limit: int # Total input + output tokens
action_on_exceed: QuotaAction
# Cost-based limits are a second enforcement layer
cost_limit_usd: Optional[Decimal] = None
# Per-request limits prevent a single runaway query
max_tokens_per_request: int = 4096
# Which models this policy applies to (None = all)
model_filter: Optional[list[str]] = None
@dataclass
class Tenant:
tenant_id: str
name: str
created_at: datetime
# System prompt that gets injected for all requests from this tenant
system_prompt_template: str
# The tenant's own API key for authenticating with your platform
platform_api_key: str
# Which LLM provider credentials to use (enables per-tenant API keys)
llm_credential_key: str = "default"
quota_policies: list[TenantQuotaPolicy] = field(default_factory=list)
# Metadata for billing
billing_tier: str = "standard"
cost_center: Optional[str] = None
@dataclass
class UsageRecord:
"""
Immutable record of a single LLM API call.
This is your audit trail and billing source of truth.
"""
record_id: str
tenant_id: str
session_id: str
request_timestamp: datetime
model: str
input_tokens: int
output_tokens: int
total_tokens: int
estimated_cost_usd: Decimal
# The request that triggered this usage
user_id: Optional[str] = None
workflow_id: Optional[str] = None
# Whether this was a cache hit (reduces billable cost)
from_cache: bool = False
# Raw provider response metadata for reconciliation
provider_request_id: Optional[str] = None
Notice a few intentional decisions here. TenantQuotaPolicy is separated from Tenant because quota policies change frequently (upgrades, downgrades, temporary overrides) while the tenant itself is relatively stable. Having multiple policies per tenant — with different periods and actions — lets you express nuanced limits like "warn at 80% of daily budget, hard-stop at 100% of monthly budget." The QuotaAction enum gives you a spectrum from graceful degradation to hard rejection, which matters for enterprise customers who will write SLA language around your behavior at their limits.
Context isolation has two levels: session-level isolation (ensuring one tenant's conversation doesn't influence another's) and prompt-level isolation (ensuring the tenant's system prompt is correctly constructed and cannot be overridden by user input).
Each tenant's active conversations need to live in a namespace that cannot be accessed or contaminated by other tenants. The session store is the first line of defense.
# session_store.py
import json
import hashlib
from datetime import timedelta
from typing import Optional
import redis.asyncio as redis
from models import Tenant
class TenantSessionStore:
"""
Manages conversation context per tenant using Redis.
All keys are namespaced by tenant_id, making cross-tenant
access structurally impossible, not just policy-enforced.
"""
def __init__(self, redis_client: redis.Redis, ttl_hours: int = 24):
self.redis = redis_client
self.ttl = timedelta(hours=ttl_hours)
def _session_key(self, tenant_id: str, session_id: str) -> str:
"""
Construct an opaque Redis key that embeds the tenant_id.
We hash the session_id to prevent key injection attacks
where a malicious user crafts a session_id that navigates
outside their tenant's namespace.
"""
session_hash = hashlib.sha256(session_id.encode()).hexdigest()[:16]
return f"session:{tenant_id}:{session_hash}"
async def get_messages(
self,
tenant_id: str,
session_id: str
) -> list[dict]:
key = self._session_key(tenant_id, session_id)
raw = await self.redis.get(key)
if raw is None:
return []
return json.loads(raw)
async def append_turn(
self,
tenant_id: str,
session_id: str,
user_message: str,
assistant_message: str,
max_turns: int = 20
) -> list[dict]:
"""
Append a conversation turn and enforce a maximum history depth.
Truncating from the front (oldest messages first) is the right
default — it preserves recent context over ancient context.
"""
key = self._session_key(tenant_id, session_id)
messages = await self.get_messages(tenant_id, session_id)
messages.append({"role": "user", "content": user_message})
messages.append({"role": "assistant", "content": assistant_message})
# Keep only the most recent max_turns pairs
if len(messages) > max_turns * 2:
messages = messages[-(max_turns * 2):]
await self.redis.setex(
key,
self.ttl,
json.dumps(messages)
)
return messages
async def clear_session(self, tenant_id: str, session_id: str) -> None:
key = self._session_key(tenant_id, session_id)
await self.redis.delete(key)
async def get_session_token_estimate(
self,
tenant_id: str,
session_id: str
) -> int:
"""
Rough token estimate for the stored history.
Uses the 4-chars-per-token heuristic, which is
conservative enough for quota checking purposes.
"""
messages = await self.get_messages(tenant_id, session_id)
total_chars = sum(len(m["content"]) for m in messages)
return total_chars // 4
The key insight here is the _session_key method. By hashing the session ID and embedding the tenant ID structurally in the Redis key, you make cross-tenant access impossible at the storage layer. A bug in your routing logic can't accidentally serve Tenant A's history to Tenant B, because Tenant B's lookup key for the same session ID would produce a completely different Redis key. Defense in depth is not paranoia; it's engineering.
Prompt injection is the multi-tenant LLM platform's SQL injection. A user who figures out that your system prompt ends with ---User message below--- can craft a message that exits the user turn and injects new instructions. Here's a robust prompt construction layer:
# prompt_builder.py
import re
from dataclasses import dataclass
from typing import Optional
from models import Tenant
# Sentinel tokens that are structurally unusual enough to be
# unlikely in legitimate user content. Don't rely on these alone.
INJECTION_PATTERNS = [
r"(?i)ignore\s+(all\s+)?previous\s+instructions",
r"(?i)you\s+are\s+now\s+a",
r"(?i)new\s+system\s+prompt",
r"(?i)disregard\s+(your\s+)?instructions",
r"(?i)act\s+as\s+if\s+you\s+(are|were)",
]
COMPILED_PATTERNS = [re.compile(p) for p in INJECTION_PATTERNS]
@dataclass
class BuiltPrompt:
system_message: str
messages: list[dict]
estimated_input_tokens: int
def build_tenant_prompt(
tenant: Tenant,
user_message: str,
conversation_history: list[dict],
context_variables: Optional[dict] = None,
sanitize_input: bool = True
) -> BuiltPrompt:
"""
Construct the full prompt for a tenant request.
This is deliberately not "smart" — it applies the tenant's
system prompt rigidly and treats the user message as data,
not instructions. That's the correct security posture.
"""
if sanitize_input:
user_message = _sanitize_user_input(user_message)
# Render the tenant's system prompt template with any
# platform-controlled variables. Tenants cannot inject
# into these variables — they come from your database.
system_content = _render_system_prompt(
tenant.system_prompt_template,
context_variables or {}
)
# Critical: wrap the user message in a structural boundary
# that the system prompt can reference. This creates a clear
# semantic boundary even if the model doesn't honor it perfectly.
wrapped_user_message = (
f"[USER INPUT START]\n{user_message}\n[USER INPUT END]"
)
# Build the messages array: history first, then current turn
messages = list(conversation_history)
messages.append({"role": "user", "content": wrapped_user_message})
# Rough token estimation for quota pre-flight
system_chars = len(system_content)
history_chars = sum(len(m["content"]) for m in conversation_history)
current_chars = len(wrapped_user_message)
estimated_tokens = (system_chars + history_chars + current_chars) // 4
return BuiltPrompt(
system_message=system_content,
messages=messages,
estimated_input_tokens=estimated_tokens
)
def _sanitize_user_input(text: str) -> str:
"""
Flag (but don't necessarily block) known injection patterns.
Blocking them entirely can cause false positives for legitimate
questions about AI, instructions, or prompt engineering itself.
The right response is to log these and review, not silently drop.
"""
for pattern in COMPILED_PATTERNS:
if pattern.search(text):
# In production, log this with the tenant/user info
# Here we annotate rather than strip
text = f"[FLAGGED CONTENT] {text}"
break
return text
def _render_system_prompt(template: str, variables: dict) -> str:
"""
Safe template rendering. We use explicit variable substitution
rather than f-strings or format() to prevent code execution.
"""
result = template
for key, value in variables.items():
# Only substitute known, validated variable names
if re.match(r'^[a-z_][a-z0-9_]*$', key):
result = result.replace(f"{{{{{key}}}}}", str(value))
return result
Warning: No amount of input sanitization fully prevents prompt injection against current LLMs. Defense in depth matters: sanitize input, structure your prompts with clear boundaries, monitor outputs for anomalies, and design your system prompts to be robust to manipulation. Treat prompt injection as a first-class threat model, not a corner case.
The quota system has two phases: pre-flight (before the API call) and post-flight (after you have the actual token counts). Pre-flight prevents overspending; post-flight gives you accurate accounting.
# quota_manager.py
import time
from decimal import Decimal
from typing import Optional, Tuple
import redis.asyncio as redis
from models import (
Tenant, TenantQuotaPolicy, QuotaPeriod,
QuotaAction, UsageRecord
)
class QuotaExceededError(Exception):
def __init__(self, tenant_id: str, policy: TenantQuotaPolicy,
current_usage: int):
self.tenant_id = tenant_id
self.policy = policy
self.current_usage = current_usage
super().__init__(
f"Tenant {tenant_id} has exceeded {policy.period.value} quota. "
f"Used {current_usage}/{policy.token_limit} tokens."
)
class QuotaManager:
"""
Enforces token quotas using Redis atomic operations.
The core primitive is INCRBY + EXPIREAT on a quota window key.
This gives us atomic increment (preventing race conditions) and
automatic cleanup (the key expires at the end of the window).
"""
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
def _quota_key(self, tenant_id: str, period: QuotaPeriod) -> str:
"""Generate a time-windowed Redis key for quota tracking."""
window = self._current_window(period)
return f"quota:{tenant_id}:{period.value}:{window}"
def _current_window(self, period: QuotaPeriod) -> str:
"""Return a string representing the current quota window."""
import datetime
now = datetime.datetime.utcnow()
if period == QuotaPeriod.HOURLY:
return now.strftime("%Y%m%d%H")
elif period == QuotaPeriod.DAILY:
return now.strftime("%Y%m%d")
elif period == QuotaPeriod.MONTHLY:
return now.strftime("%Y%m")
def _window_expiry_seconds(self, period: QuotaPeriod) -> int:
"""How long until the current window expires, in seconds."""
import datetime
now = datetime.datetime.utcnow()
if period == QuotaPeriod.HOURLY:
next_window = (now + datetime.timedelta(hours=1)).replace(
minute=0, second=0, microsecond=0
)
elif period == QuotaPeriod.DAILY:
next_window = (now + datetime.timedelta(days=1)).replace(
hour=0, minute=0, second=0, microsecond=0
)
elif period == QuotaPeriod.MONTHLY:
if now.month == 12:
next_window = now.replace(
year=now.year + 1, month=1, day=1,
hour=0, minute=0, second=0, microsecond=0
)
else:
next_window = now.replace(
month=now.month + 1, day=1,
hour=0, minute=0, second=0, microsecond=0
)
return int((next_window - now).total_seconds()) + 60 # +60s buffer
async def get_current_usage(
self,
tenant_id: str,
period: QuotaPeriod
) -> int:
key = self._quota_key(tenant_id, period)
value = await self.redis.get(key)
return int(value) if value else 0
async def check_and_reserve(
self,
tenant_id: str,
policies: list[TenantQuotaPolicy],
estimated_tokens: int,
model: str
) -> Tuple[bool, Optional[TenantQuotaPolicy], Optional[QuotaAction]]:
"""
Pre-flight quota check. Uses a reservation model:
we tentatively add the estimated tokens, check if we're
over the limit, and roll back if we need to reject.
Returns: (allowed, violated_policy, action_to_take)
"""
applicable_policies = [
p for p in policies
if p.model_filter is None or model in p.model_filter
]
for policy in applicable_policies:
# Check per-request limit first — this is cheap
if estimated_tokens > policy.max_tokens_per_request:
return False, policy, QuotaAction.REJECT
key = self._quota_key(tenant_id, policy.period)
expiry = self._window_expiry_seconds(policy.period)
# Atomic reservation: increment and check
# Using a Lua script ensures atomicity
lua_script = """
local current = redis.call('INCRBY', KEYS[1], ARGV[1])
if current == tonumber(ARGV[1]) then
-- First increment, set expiry
redis.call('EXPIRE', KEYS[1], ARGV[2])
end
return current
"""
new_total = await self.redis.eval(
lua_script, 1, key,
str(estimated_tokens), str(expiry)
)
new_total = int(new_total)
if new_total > policy.token_limit:
if policy.action_on_exceed == QuotaAction.REJECT:
# Roll back the reservation
await self.redis.decrby(key, estimated_tokens)
return False, policy, QuotaAction.REJECT
elif policy.action_on_exceed == QuotaAction.THROTTLE:
return False, policy, QuotaAction.THROTTLE
# NOTIFY and ALLOW fall through — we let the request proceed
# but the caller should fire an alert for NOTIFY
return True, None, None
async def reconcile_usage(
self,
tenant_id: str,
policies: list[TenantQuotaPolicy],
estimated_tokens: int,
actual_tokens: int,
model: str
) -> None:
"""
After a successful API call, adjust the quota accounting
from the estimated token count to the actual count.
This matters because estimates can be off by 20-40%
for responses with variable length (code, structured data).
"""
delta = actual_tokens - estimated_tokens
if delta == 0:
return
applicable_policies = [
p for p in policies
if p.model_filter is None or model in p.model_filter
]
for policy in applicable_policies:
key = self._quota_key(tenant_id, policy.period)
if delta > 0:
await self.redis.incrby(key, delta)
else:
await self.redis.decrby(key, abs(delta))
The Lua script in check_and_reserve is doing real work: it increments the quota counter and sets the expiry in a single atomic operation. Without atomicity here, you have a race condition where two concurrent requests both read a value under the limit, both decide to proceed, and both push the tenant over quota. This is the "thundering herd at quota boundary" problem, and it's the reason multi-tenant quota systems are hard.
Tip: Redis's
EVALfor Lua scripts runs on a single thread and is guaranteed atomic. It's the right tool here. If you're using Redis Cluster, be aware that Lua scripts can only operate on keys that hash to the same slot — structure your quota keys to use hash tags ({tenant_id}) to ensure this.
Token counts from your provider don't translate to dollars without a pricing table, and pricing tables change. Build a cost calculation layer that's separate from your quota logic so you can update prices without touching enforcement code.
# cost_calculator.py
from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_UP
from typing import Optional
import datetime
@dataclass
class ModelPricing:
model_id: str
input_cost_per_million_tokens: Decimal
output_cost_per_million_tokens: Decimal
# Some models have tiered pricing above a token threshold
high_volume_threshold_tokens: Optional[int] = None
high_volume_input_discount: Decimal = Decimal("0")
high_volume_output_discount: Decimal = Decimal("0")
effective_from: datetime.date = datetime.date(2024, 1, 1)
# Pricing as of late 2024 — update this from your provider's API
# or a configuration store in production
MODEL_PRICING: dict[str, ModelPricing] = {
"gpt-4o": ModelPricing(
model_id="gpt-4o",
input_cost_per_million_tokens=Decimal("2.50"),
output_cost_per_million_tokens=Decimal("10.00"),
),
"gpt-4o-mini": ModelPricing(
model_id="gpt-4o-mini",
input_cost_per_million_tokens=Decimal("0.15"),
output_cost_per_million_tokens=Decimal("0.60"),
),
"claude-3-5-sonnet-20241022": ModelPricing(
model_id="claude-3-5-sonnet-20241022",
input_cost_per_million_tokens=Decimal("3.00"),
output_cost_per_million_tokens=Decimal("15.00"),
),
"claude-3-haiku-20240307": ModelPricing(
model_id="claude-3-haiku-20240307",
input_cost_per_million_tokens=Decimal("0.25"),
output_cost_per_million_tokens=Decimal("1.25"),
),
}
def calculate_request_cost(
model: str,
input_tokens: int,
output_tokens: int,
markup_multiplier: Decimal = Decimal("1.0")
) -> Decimal:
"""
Calculate the cost of a single LLM API call.
markup_multiplier lets you add a margin for resale.
A value of 1.20 represents a 20% markup over provider cost.
Uses Decimal throughout to avoid floating-point errors
that compound across thousands of records into billing discrepancies.
"""
pricing = MODEL_PRICING.get(model)
if pricing is None:
raise ValueError(f"No pricing data for model: {model}")
input_cost = (
Decimal(input_tokens) / Decimal("1_000_000")
) * pricing.input_cost_per_million_tokens
output_cost = (
Decimal(output_tokens) / Decimal("1_000_000")
) * pricing.output_cost_per_million_tokens
total_cost = (input_cost + output_cost) * markup_multiplier
# Round to 8 decimal places — enough precision for micro-transactions
# but not so many that you confuse your accounting system
return total_cost.quantize(Decimal("0.00000001"), rounding=ROUND_HALF_UP)
@dataclass
class TenantCostSummary:
tenant_id: str
period_start: datetime.date
period_end: datetime.date
total_input_tokens: int
total_output_tokens: int
total_tokens: int
total_cost_usd: Decimal
cost_by_model: dict[str, Decimal]
cost_by_workflow: dict[str, Decimal]
request_count: int
average_cost_per_request: Decimal
Using Decimal everywhere in cost calculations is not pedantic — it's necessary. Over thousands of API calls, floating-point rounding errors accumulate into real dollar discrepancies. If you're using these numbers for billing, a penny discrepancy per call becomes $10 per thousand calls and $10,000 per million calls. Your finance team will notice, and they won't be pleased.
Now we assemble the components into a request pipeline. This is the function your API endpoint calls for every LLM request.
# llm_pipeline.py
import uuid
import asyncio
from datetime import datetime
from decimal import Decimal
from typing import Optional, AsyncGenerator
import openai
from models import Tenant, UsageRecord, QuotaAction
from session_store import TenantSessionStore
from prompt_builder import build_tenant_prompt
from quota_manager import QuotaManager, QuotaExceededError
from cost_calculator import calculate_request_cost
class LLMPipeline:
def __init__(
self,
session_store: TenantSessionStore,
quota_manager: QuotaManager,
usage_repository, # Your database DAO
alert_service, # Webhook/notification service
openai_client: openai.AsyncOpenAI,
markup_multiplier: Decimal = Decimal("1.15")
):
self.sessions = session_store
self.quota = quota_manager
self.usage_repo = usage_repository
self.alerts = alert_service
self.openai = openai_client
self.markup = markup_multiplier
async def complete(
self,
tenant: Tenant,
session_id: str,
user_message: str,
model: str = "gpt-4o-mini",
user_id: Optional[str] = None,
workflow_id: Optional[str] = None,
context_variables: Optional[dict] = None,
max_completion_tokens: Optional[int] = None
) -> dict:
"""
Execute a complete LLM request with quota checking,
context injection, and cost attribution.
"""
# 1. Load conversation history for this tenant/session
history = await self.sessions.get_messages(tenant.tenant_id, session_id)
# 2. Build the prompt with tenant context
built_prompt = build_tenant_prompt(
tenant=tenant,
user_message=user_message,
conversation_history=history,
context_variables=context_variables
)
# 3. Estimate total tokens for quota pre-flight
# We add max_completion_tokens as a buffer for output tokens
completion_buffer = max_completion_tokens or 1024
estimated_total = built_prompt.estimated_input_tokens + completion_buffer
# 4. Check quota before making the API call
allowed, violated_policy, action = await self.quota.check_and_reserve(
tenant_id=tenant.tenant_id,
policies=tenant.quota_policies,
estimated_tokens=estimated_total,
model=model
)
if not allowed:
if action == QuotaAction.REJECT:
raise QuotaExceededError(
tenant.tenant_id, violated_policy,
await self.quota.get_current_usage(
tenant.tenant_id, violated_policy.period
)
)
elif action == QuotaAction.THROTTLE:
# In production, push to a priority queue instead of sleeping
await asyncio.sleep(2.0)
if action == QuotaAction.NOTIFY:
await self.alerts.send_quota_alert(
tenant_id=tenant.tenant_id,
policy=violated_policy,
current_usage=await self.quota.get_current_usage(
tenant.tenant_id, violated_policy.period
)
)
# 5. Make the API call
request_start = datetime.utcnow()
try:
response = await self.openai.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": built_prompt.system_message},
*built_prompt.messages
],
max_tokens=max_completion_tokens,
# Tag each request with tenant metadata for provider-side
# cost analysis (OpenAI supports this via metadata)
metadata={
"tenant_id": tenant.tenant_id,
"session_id": session_id,
"workflow_id": workflow_id or "direct"
}
)
except openai.RateLimitError as e:
# Roll back the quota reservation if the provider rejects us
# This is important: we reserved tokens but consumed none
await self.quota.reconcile_usage(
tenant.tenant_id, tenant.quota_policies,
estimated_total, 0, model
)
raise
# 6. Extract actual token usage from the response
actual_input_tokens = response.usage.prompt_tokens
actual_output_tokens = response.usage.completion_tokens
actual_total_tokens = response.usage.total_tokens
assistant_message = response.choices[0].message.content
# 7. Reconcile quota: adjust from estimate to actual
await self.quota.reconcile_usage(
tenant_id=tenant.tenant_id,
policies=tenant.quota_policies,
estimated_tokens=estimated_total,
actual_tokens=actual_total_tokens,
model=model
)
# 8. Calculate cost and write the usage record
cost = calculate_request_cost(
model=model,
input_tokens=actual_input_tokens,
output_tokens=actual_output_tokens,
markup_multiplier=self.markup
)
usage_record = UsageRecord(
record_id=str(uuid.uuid4()),
tenant_id=tenant.tenant_id,
session_id=session_id,
request_timestamp=request_start,
model=model,
input_tokens=actual_input_tokens,
output_tokens=actual_output_tokens,
total_tokens=actual_total_tokens,
estimated_cost_usd=cost,
user_id=user_id,
workflow_id=workflow_id,
provider_request_id=response.id
)
# Write to DB asynchronously — don't block the response
asyncio.create_task(
self.usage_repo.insert_usage_record(usage_record)
)
# 9. Update session history
await self.sessions.append_turn(
tenant_id=tenant.tenant_id,
session_id=session_id,
user_message=user_message,
assistant_message=assistant_message
)
return {
"content": assistant_message,
"usage": {
"input_tokens": actual_input_tokens,
"output_tokens": actual_output_tokens,
"total_tokens": actual_total_tokens,
"estimated_cost_usd": float(cost)
},
"session_id": session_id,
"record_id": usage_record.record_id
}
Notice the error handling in step 6. When the provider returns a rate limit error (which is their rate limit, not yours), you need to roll back the quota reservation. The tenant shouldn't be penalized for a failure on your provider's side. This distinction — your quota limits vs. provider rate limits — matters operationally, and you need to handle both failure modes distinctly.
Quota enforcement keeps you from overspending. Usage reporting tells you who spent what, so you can bill accurately and make product decisions.
# usage_repository.py
from decimal import Decimal
from datetime import date, datetime
from typing import Optional
import asyncpg
from models import UsageRecord, TenantCostSummary
from cost_calculator import calculate_request_cost
class UsageRepository:
def __init__(self, db_pool: asyncpg.Pool):
self.db = db_pool
async def insert_usage_record(self, record: UsageRecord) -> None:
async with self.db.acquire() as conn:
await conn.execute("""
INSERT INTO llm_usage_records (
record_id, tenant_id, session_id, request_timestamp,
model, input_tokens, output_tokens, total_tokens,
estimated_cost_usd, user_id, workflow_id,
from_cache, provider_request_id
) VALUES (
$1, $2, $3, $4, $5, $6, $7, $8,
$9, $10, $11, $12, $13
)
""",
record.record_id, record.tenant_id, record.session_id,
record.request_timestamp, record.model,
record.input_tokens, record.output_tokens, record.total_tokens,
record.estimated_cost_usd, record.user_id, record.workflow_id,
record.from_cache, record.provider_request_id
)
async def get_tenant_cost_summary(
self,
tenant_id: str,
period_start: date,
period_end: date
) -> TenantCostSummary:
"""
Generate a billing-grade cost summary for a tenant.
This query is the foundation of your invoice generation.
"""
async with self.db.acquire() as conn:
# Main aggregation
totals = await conn.fetchrow("""
SELECT
COUNT(*) as request_count,
SUM(input_tokens) as total_input_tokens,
SUM(output_tokens) as total_output_tokens,
SUM(total_tokens) as total_tokens,
SUM(estimated_cost_usd) as total_cost_usd
FROM llm_usage_records
WHERE tenant_id = $1
AND request_timestamp >= $2
AND request_timestamp < $3
AND from_cache = false
""", tenant_id,
datetime.combine(period_start, datetime.min.time()),
datetime.combine(period_end, datetime.min.time()))
# Cost breakdown by model
model_rows = await conn.fetch("""
SELECT
model,
SUM(estimated_cost_usd) as cost_usd
FROM llm_usage_records
WHERE tenant_id = $1
AND request_timestamp >= $2
AND request_timestamp < $3
AND from_cache = false
GROUP BY model
ORDER BY cost_usd DESC
""", tenant_id,
datetime.combine(period_start, datetime.min.time()),
datetime.combine(period_end, datetime.min.time()))
# Cost breakdown by workflow
workflow_rows = await conn.fetch("""
SELECT
COALESCE(workflow_id, 'direct') as workflow_id,
SUM(estimated_cost_usd) as cost_usd
FROM llm_usage_records
WHERE tenant_id = $1
AND request_timestamp >= $2
AND request_timestamp < $3
AND from_cache = false
GROUP BY workflow_id
ORDER BY cost_usd DESC
""", tenant_id,
datetime.combine(period_start, datetime.min.time()),
datetime.combine(period_end, datetime.min.time()))
total_cost = totals["total_cost_usd"] or Decimal("0")
request_count = totals["request_count"] or 0
return TenantCostSummary(
tenant_id=tenant_id,
period_start=period_start,
period_end=period_end,
total_input_tokens=totals["total_input_tokens"] or 0,
total_output_tokens=totals["total_output_tokens"] or 0,
total_tokens=totals["total_tokens"] or 0,
total_cost_usd=total_cost,
cost_by_model={
r["model"]: r["cost_usd"] for r in model_rows
},
cost_by_workflow={
r["workflow_id"]: r["cost_usd"] for r in workflow_rows
},
request_count=request_count,
average_cost_per_request=(
total_cost / request_count
if request_count > 0 else Decimal("0")
)
)
The PostgreSQL schema that backs this needs two indexes to perform well at scale: one on (tenant_id, request_timestamp) for tenant-scoped time range queries, and one on (request_timestamp) for global analytics. Partition the table by month once you're above a few million rows — monthly usage reports against a multi-year unpartitioned table will slow to a crawl.
-- Schema for reference
CREATE TABLE llm_usage_records (
record_id UUID PRIMARY KEY,
tenant_id VARCHAR(64) NOT NULL,
session_id VARCHAR(128) NOT NULL,
request_timestamp TIMESTAMPTZ NOT NULL,
model VARCHAR(128) NOT NULL,
input_tokens INTEGER NOT NULL,
output_tokens INTEGER NOT NULL,
total_tokens INTEGER NOT NULL,
estimated_cost_usd NUMERIC(16, 8) NOT NULL,
user_id VARCHAR(128),
workflow_id VARCHAR(128),
from_cache BOOLEAN NOT NULL DEFAULT FALSE,
provider_request_id VARCHAR(256)
) PARTITION BY RANGE (request_timestamp);
CREATE INDEX idx_usage_tenant_time
ON llm_usage_records (tenant_id, request_timestamp);
CREATE INDEX idx_usage_time
ON llm_usage_records (request_timestamp);
Some enterprise customers will want to bring their own API keys — either for compliance reasons (their data must never touch a shared API key) or because they've negotiated enterprise pricing directly with OpenAI or Anthropic. Your credential manager needs to handle this cleanly.
# credential_manager.py
import os
from typing import Optional
import openai
from cryptography.fernet import Fernet
class CredentialManager:
"""
Manages LLM API credentials per tenant.
Credentials are encrypted at rest using Fernet symmetric encryption.
In production, the encryption key should come from a KMS,
not an environment variable.
"""
def __init__(self, encryption_key: bytes, credential_store: dict):
self.fernet = Fernet(encryption_key)
# In production: encrypted records in your database, not a dict
self._store = credential_store
def store_credential(
self,
credential_key: str,
api_key: str
) -> None:
encrypted = self.fernet.encrypt(api_key.encode())
self._store[credential_key] = encrypted
def get_openai_client(
self,
credential_key: str = "default"
) -> openai.AsyncOpenAI:
if credential_key == "default":
return openai.AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])
encrypted = self._store.get(credential_key)
if not encrypted:
raise ValueError(f"No credentials found for key: {credential_key}")
api_key = self.fernet.decrypt(encrypted).decode()
return openai.AsyncOpenAI(api_key=api_key)
Warning: Bring-your-own-key (BYOK) customers introduce a support complication: when their API key is invalid, revoked, or rate-limited, requests fail in ways that look identical to other failures from your platform's perspective. Build explicit credential validation on ingestion and a clear error path that distinguishes credential failures from capacity failures.
One cost optimization that requires careful thought in a multi-tenant context is semantic caching: if two different users (even from different tenants) ask semantically identical questions, you return a cached response instead of making a new API call.
The multi-tenant wrinkle is that cache keys must include tenant context. A cached response to "What's our refund policy?" from Tenant A (a software company) is completely wrong for Tenant B (a clothing retailer). The cache key must incorporate the tenant's system prompt — or at least a hash of it.
# semantic_cache.py
import hashlib
import json
from typing import Optional
import redis.asyncio as redis
class TenantAwareSemanticCache:
"""
A simple exact-match cache with tenant isolation.
True semantic caching requires embedding similarity search
(e.g., using pgvector or a vector database), but starts here.
"""
def __init__(self, redis_client: redis.Redis, ttl_seconds: int = 3600):
self.redis = redis_client
self.ttl = ttl_seconds
def _cache_key(
self,
tenant_id: str,
system_prompt: str,
user_message: str,
model: str
) -> str:
"""
Cache key must incorporate both the tenant's system prompt
and the user message. The system prompt is the context that
makes the same question have different correct answers across tenants.
"""
content = json.dumps({
"tenant_id": tenant_id,
"system_prompt_hash": hashlib.sha256(
system_prompt.encode()
).hexdigest(),
"user_message": user_message.strip().lower(),
"model": model
}, sort_keys=True)
return f"cache:{hashlib.sha256(content.encode()).hexdigest()}"
async def get(
self,
tenant_id: str,
system_prompt: str,
user_message: str,
model: str
) -> Optional[str]:
key = self._cache_key(tenant_id, system_prompt, user_message, model)
return await self.redis.get(key)
async def set(
self,
tenant_id: str,
system_prompt: str,
user_message: str,
model: str,
response: str
) -> None:
key = self._cache_key(tenant_id, system_prompt, user_message, model)
await self.redis.setex(key, self.ttl, response)
For cache hits, you still need to log a usage record — but with from_cache=True and zero tokens — so your reporting accurately reflects when cost savings occurred. This also lets you calculate your cache hit rate by tenant, which is valuable data for optimizing your caching strategy.
Build a minimal but complete multi-tenant LLM API server using FastAPI that demonstrates all the concepts from this lesson. The exercise has four stages.
Stage 1: Scaffold the system Create the SQLite + Redis data layer for two tenants: "Acme Corp" (a customer service chatbot for a software company) and "Brightline Health" (a medical information assistant). Give them different system prompts, different quota limits (Acme: 100,000 tokens/day, Brightline: 50,000 tokens/day), and different quota actions (Acme: THROTTLE at limit, Brightline: REJECT at limit — you can imagine why a medical assistant might need stricter controls).
Stage 2: Build the request endpoint
Create a POST /v1/chat/{tenant_id} endpoint that:
platform_api_key in the Authorization headerLLMPipeline.complete() flowStage 3: Test quota enforcement Write a script that fires 200 concurrent requests for Acme Corp's tenant. Observe: (a) Do any requests succeed after the quota is exhausted? (b) Is the token count in Redis accurate after the burst? (c) Are all 200 usage records present in the database?
Stage 4: Generate a cost report
Write the SQL query (or use the UsageRepository) to generate a monthly cost summary for both tenants, broken down by workflow. Add a 15% markup to the provider cost and calculate what you'd invoice each customer.
Evaluation criteria: The quota counter in Redis should never allow more than 10% overage above the limit during a burst (due to estimation lag). Every API call, whether successful or not, should have a usage record. The cost summary should reconcile exactly with the sum of individual estimated_cost_usd fields.
The Race Condition at Quota Reset
When a quota window resets (midnight for a daily quota), you'll often see a burst of requests that were queued waiting for the reset. If your reset logic is clock-based and slightly inaccurate across instances, you can get a window where multiple workers all believe the quota has reset and flood through. The fix: use the Redis key expiry as the source of truth, not application-layer clock logic. If the key exists, you're in the window. If it doesn't, the window has reset.
Prompt Leakage Through System Prompt Templates
The most common context isolation bug: a developer puts tenant-specific data (like a customer list or internal pricing) directly into the system prompt template, which gets cached in Redis as part of the session. When the system prompt changes (new pricing, updated policy), the cached sessions still carry the old system prompt. Fix: don't cache the system prompt in the session store. Rebuild it fresh from the tenant record on every request. The session store should contain only the conversation turns.
Estimation Errors on Structured Output
If a tenant's workflow requests structured JSON output (e.g., "respond only in valid JSON"), the output token count can be significantly higher than free-form prose for equivalent information, because JSON structure tokens (brackets, quotes, commas) all count. The 4-chars-per-token heuristic is calibrated for prose. For tenants using structured output, use a larger completion_buffer in your pre-flight estimation, or instrument your system to measure average output tokens for each workflow type and use those empirical values.
The "Ghost Tenant" Cost Problem
You cancel a tenant's account, but their sessions remain active in Redis (TTL hasn't expired) and their usage records reference a now-deleted tenant record. On re-signup, a new tenant might get a session ID that collides with a ghost session. Fix: on tenant deactivation, actively scan and delete their session keys using SCAN with the session:{tenant_id}:* pattern. Don't rely on TTL expiry for security-sensitive cleanup.
Provider Rate Limits vs. Your Rate Limits
Your quota system controls how many tokens a tenant can spend on your platform. Your LLM provider has separate rate limits on your API key (tokens per minute, requests per minute). Under burst load from a single tenant, you can hit the provider's RPM limit before you hit the tenant's token quota. These errors look like provider failures to the tenant, but they're actually architectural: you need to either spread requests across time (rate-limiting at the request queue level) or use multiple provider API keys with load balancing.
Database Write Backpressure
The asyncio.create_task() call for writing usage records is a fire-and-forget pattern that can hide failures. If your database is under load, those background tasks can pile up in memory and fail silently. In production, use a proper async job queue (Celery, ARQ, or a message broker) for the usage record writes, with dead-letter handling for failures. The usage record is your billing record — losing it means losing revenue.
Tip: Add a Prometheus counter that tracks "usage records written" vs. "usage records attempted" and alert when the gap exceeds 0.01%. Silent write failures are the insidious billing bug that only surfaces at month-end invoice reconciliation.
You've built the core of a production multi-tenant LLM platform. The architecture we've built provides structural isolation (Redis key namespacing, hashed session IDs), semantic isolation (tenant-scoped prompt construction), pre-flight quota enforcement using atomic Redis operations, post-flight reconciliation to handle estimation error, cost attribution at the model, workflow, and tenant levels, and a per-tenant credential system for BYOK requirements.
The patterns here compose well. The LLMPipeline class is the integration point — everything else is independently replaceable. You can swap Redis for DynamoDB, PostgreSQL for BigQuery, or OpenAI for Anthropic without touching your quota logic.
Where to go from here:
Streaming responses and quota enforcement: The architecture above works for synchronous completions. Streaming (stream=True) requires a different quota model where you count tokens as they arrive and can cut off mid-stream when a limit is hit. This is architecturally significant because you can't do a pre-flight estimate that's reliable enough for streaming without a per-token counter.
Semantic caching at scale: Replace the exact-match cache with embedding-based similarity search using pgvector. Cache hit rates of 30-40% are achievable for support chatbot workloads, which directly reduces per-tenant costs.
Context window management: The max_turns parameter in the session store is a crude context window manager. A more sophisticated approach uses token-accurate history truncation that preserves the most semantically important turns (not just the most recent), using techniques like conversation summarization.
Tenant isolation for RAG pipelines: If tenants each have their own document corpora, the isolation problem extends to your vector database. Namespaced collections per tenant is the standard approach, but it has scaling implications when you have hundreds of tenants with small corpora.
Audit logging for compliance: Enterprise customers in regulated industries (healthcare, finance) need an immutable audit log of every request and response. This is different from your usage records — it needs to capture the full content, be tamper-evident, and potentially be exportable on demand.
Learning Path: Building with LLMs