Building AI SaaS with Chinese Models

Every SaaS product is adding AI features in 2026. But building a multi-tenant AI application is fundamentally different from traditional SaaS: you are reselling API capacity, managing per-customer usage, handling variable costs, and dealing with model failures. This guide shows you how to build production-grade AI SaaS on top of Chinese models through TokenEase.

Architecture Overview

A production AI SaaS has five layers:

1. API Gateway: Rate limiting, authentication, request routing
2. Tenant Isolation: Per-customer API keys, usage quotas, data boundaries
3. AI Proxy: Model selection, fallback handling, response caching
4. Usage Tracking: Real-time token counting, cost attribution, analytics
5. Billing: Usage-based invoicing, credit systems, overage handling

Layer 1: API Gateway with Tenant Authentication

Every request must identify the tenant and enforce their limits before reaching the AI layer.

FastAPI Gateway Implementation

from fastapi import FastAPI, HTTPException, Depends, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import redis
import time

app = FastAPI()
security = HTTPBearer()
redis_client = redis.Redis(host='localhost', port=6379, db=0)

# Tenant database (use PostgreSQL in production)
TENANTS = {
    "tkn_tenant_001": {
        "name": "Acme Corp",
        "plan": "pro",
        "rate_limit": 100,  # requests per minute
        "monthly_quota": 1_000_000,  # tokens
        "tokens_used_this_month": 0
    }
}

def get_tenant(credentials: HTTPAuthorizationCredentials = Depends(security)):
    api_key = credentials.credentials
    if not api_key.startswith("tkn_tenant_"):
        raise HTTPException(status_code=401, detail="Invalid API key format")
    
    tenant = TENANTS.get(api_key)
    if not tenant:
        raise HTTPException(status_code=401, detail="Unknown API key")
    
    return tenant

def check_rate_limit(tenant_id: str, limit: int):
    key = f"rate_limit:{tenant_id}:{int(time.time()) // 60}"
    current = redis_client.incr(key)
    if current == 1:
        redis_client.expire(key, 60)
    
    if current > limit:
        raise HTTPException(status_code=429, detail="Rate limit exceeded")

@app.post("/v1/chat/completions")
async def chat_completion(
    request: Request,
    tenant: dict = Depends(get_tenant)
):
    # Enforce rate limit
    check_rate_limit(tenant["name"], tenant["rate_limit"])
    
    # Check quota
    if tenant["tokens_used_this_month"] >= tenant["monthly_quota"]:
        raise HTTPException(status_code=403, detail="Monthly quota exceeded")
    
    # Forward to AI layer
    body = await request.json()
    response = await process_ai_request(body, tenant)
    
    # Track usage
    tokens_used = response.get("usage", {}).get("total_tokens", 0)
    tenant["tokens_used_this_month"] += tokens_used
    
    return response

Layer 2: Tenant Isolation

Each tenant gets isolated resources, settings, and data boundaries.

Per-Tenant Configuration

class TenantConfig:
    def __init__(self, tenant_id):
        self.tenant_id = tenant_id
        self.config = self._load_config()
    
    def _load_config(self):
        # Load from database
        return {
            "allowed_models": ["deepseek", "qwen", "doubao"],
            "default_model": "qwen",
            "max_tokens_per_request": 4000,
            "temperature_range": [0.0, 1.0],
            "custom_system_prompt": None,
            "data_retention_days": 30
        }
    
    def is_model_allowed(self, model: str) -> bool:
        return model in self.config["allowed_models"]
    
    def validate_request(self, request_body: dict):
        model = request_body.get("model", self.config["default_model"])
        if not self.is_model_allowed(model):
            raise ValueError(f"Model '{model}' not allowed for this tenant")
        
        max_tokens = request_body.get("max_tokens", 0)
        if max_tokens > self.config["max_tokens_per_request"]:
            raise ValueError(f"max_tokens exceeds limit of {self.config['max_tokens_per_request']}")

# Usage in request handler
config = TenantConfig(tenant["name"])
config.validate_request(body)

Layer 3: AI Proxy with Smart Routing

The proxy layer sits between your gateway and TokenEase. It handles model selection, retries, and fallbacks.

Smart Model Router

import openai
import asyncio
from typing import Optional

class AIProxy:
    def __init__(self, tokenease_key: str):
        self.client = openai.AsyncOpenAI(
            base_url="https://tokenease.io/v1",
            api_key=tokenease_key
        )
        self.fallback_chain = ["deepseek", "qwen", "doubao"]
    
    async def complete_with_fallback(
        self,
        messages,
        preferred_model: str,
        max_retries: int = 2
    ):
        models_to_try = [preferred_model] + [
            m for m in self.fallback_chain if m != preferred_model
        ]
        
        last_error = None
        for model in models_to_try:
            for attempt in range(max_retries):
                try:
                    response = await self.client.chat.completions.create(
                        model=model,
                        messages=messages,
                        timeout=30
                    )
                    return {
                        "success": True,
                        "model_used": model,
                        "response": response,
                        "fallback": model != preferred_model
                    }
                except Exception as e:
                    last_error = e
                    if attempt < max_retries - 1:
                        await asyncio.sleep(2 ** attempt)
        
        return {
            "success": False,
            "error": str(last_error),
            "model_used": None
        }
    
    async def route_by_complexity(self, messages):
        # Simple heuristic-based routing
        prompt = messages[-1]["content"] if messages else ""
        prompt_length = len(prompt)
        
        if prompt_length < 100:
            model = "doubao"      # Cheap for simple queries
        elif "code" in prompt.lower() or "debug" in prompt.lower():
            model = "deepseek"    # Best for coding
        elif prompt_length > 2000:
            model = "kimi"        # Long context
        else:
            model = "qwen"        # General purpose
        
        return await self.complete_with_fallback(messages, model)

# Usage
proxy = AIProxy("sk-your-tokenease-key")
result = await proxy.route_by_complexity(messages)
if result["success"]:
    print(f"Response from {result['model_used']}")
    if result["fallback"]:
        print("Note: Fallback model was used")

Layer 4: Usage Tracking and Analytics

Real-time usage tracking is essential for billing, quota enforcement, and business intelligence.

Token Usage Tracking

import json
from datetime import datetime, timedelta
from collections import defaultdict

class UsageTracker:
    def __init__(self):
        self.daily_usage = defaultdict(lambda: defaultdict(int))
    
    def record_usage(
        self,
        tenant_id: str,
        model: str,
        input_tokens: int,
        output_tokens: int,
        latency_ms: float
    ):
        date_key = datetime.now().strftime("%Y-%m-%d")
        
        self.daily_usage[tenant_id][f"{date_key}:{model}:input"] += input_tokens
        self.daily_usage[tenant_id][f"{date_key}:{model}:output"] += output_tokens
        self.daily_usage[tenant_id][f"{date_key}:requests"] += 1
        self.daily_usage[tenant_id][f"{date_key}:latency_total"] += latency_ms
    
    def get_tenant_summary(self, tenant_id: str, days: int = 30):
        summary = {
            "total_input_tokens": 0,
            "total_output_tokens": 0,
            "total_requests": 0,
            "avg_latency_ms": 0,
            "cost_by_model": defaultdict(float)
        }
        
        pricing = {
            "deepseek": {"input": 0.50, "output": 2.00},
            "qwen": {"input": 0.40, "output": 1.60},
            "kimi": {"input": 0.80, "output": 3.20},
            "doubao": {"input": 0.30, "output": 1.20}
        }
        
        for key, value in self.daily_usage[tenant_id].items():
            if ":input" in key:
                model = key.split(":")[1]
                summary["total_input_tokens"] += value
                summary["cost_by_model"][model] += value * pricing[model]["input"] / 1_000_000
            elif ":output" in key:
                model = key.split(":")[1]
                summary["total_output_tokens"] += value
                summary["cost_by_model"][model] += value * pricing[model]["output"] / 1_000_000
            elif ":requests" in key:
                summary["total_requests"] += value
            elif ":latency_total" in key:
                total_requests = summary["total_requests"] or 1
                summary["avg_latency_ms"] = value / total_requests
        
        summary["total_cost"] = sum(summary["cost_by_model"].values())
        return summary

# Usage
tracker = UsageTracker()
tracker.record_usage("acme_corp", "deepseek", 500, 200, 450)
summary = tracker.get_tenant_summary("acme_corp")
print(f"Total cost: ${summary['total_cost']:.2f}")

Layer 5: Usage-Based Billing

Convert token usage into customer-facing bills. Common pricing models for AI SaaS:

Model	Description	Best For
Pay-as-you-go	Charge per 1K tokens used	Variable usage, enterprise
Credit packs	Pre-purchased token credits	Self-serve, predictability
Tiered plans	Fixed monthly fee + overage	Most SaaS products
Per-seat	Fixed per user, unlimited usage	Team products
Hybrid	Base fee + usage component	Growth stage SaaS

Markup Strategy

Typical SaaS markup on AI API costs:

• Low margin (1.2-1.5x): High volume, competitive market
• Standard (2-3x): Most AI SaaS products
• Premium (5-10x): Vertical-specific with unique value-add

Billing Calculation

class BillingEngine:
    def __init__(self):
        self.markup_multiplier = 2.5  # 2.5x markup on raw API costs
    
    def calculate_invoice(self, tenant_id, usage_summary):
        raw_cost = usage_summary["total_cost"]
        marked_up_cost = raw_cost * self.markup_multiplier
        
        # Apply plan discounts
        plan = TENANTS[tenant_id]["plan"]
        if plan == "enterprise":
            discount = 0.20  # 20% volume discount
        elif plan == "pro":
            discount = 0.10
        else:
            discount = 0.0
        
        final_cost = marked_up_cost * (1 - discount)
        
        return {
            "raw_api_cost": raw_cost,
            "marked_up_cost": marked_up_cost,
            "discount": discount,
            "final_amount": final_cost,
            "currency": "USD"
        }

# Generate monthly invoice
usage = tracker.get_tenant_summary("acme_corp", days=30)
invoice = BillingEngine().calculate_invoice("tkn_tenant_001", usage)
print(f"Invoice amount: ${invoice['final_amount']:.2f}")

Complete Request Flow

Here is how a complete request flows through the system:

1. Client sends request with tenant API key
2. Gateway authenticates tenant, checks rate limits
3. Tenant Config validates model selection and parameters
4. AI Proxy routes to optimal model, handles fallbacks
5. TokenEase processes the request via Chinese AI model
6. AI Proxy returns response to gateway
7. Usage Tracker records tokens, latency, and model used
8. Gateway returns response to client

Production Checklist

Security

• Encrypt tenant API keys at rest (AES-256)
• Use TLS 1.3 for all client communications
• Validate all request parameters against tenant limits
• Implement request signing for webhook callbacks
• Log all access with tenant attribution for audit trails

Reliability

• Implement circuit breakers for AI provider failures
• Set request timeouts (recommend 30s for chat, 60s for completions)
• Use retry with exponential backoff and jitter
• Maintain a warm connection pool to TokenEase
• Implement graceful degradation when quotas are exceeded

Scalability

• Cache frequent responses (respect tenant isolation)
• Use async I/O for concurrent request handling
• Shard usage data by tenant for horizontal scaling
• Pre-aggregate billing data hourly to reduce month-end load

Cost Example: Running AI SaaS at Scale

A typical AI SaaS with 1,000 active tenants:

Metric	Value
Average tokens per tenant per day	50,000
Total daily tokens	50M
Raw API cost (via TokenEase)	$25/day
Monthly raw cost	$750
With 2.5x markup	$1,875 revenue
Gross margin	60%
Monthly profit (before infrastructure)	$1,125

Frequently Asked Questions

How do I prevent one tenant from consuming all capacity?

Implement per-tenant rate limits, token quotas, and concurrent request limits. Use Redis for distributed rate limiting across your server fleet.

Should I pass through raw model errors to tenants?

No. Map provider errors to your own error codes. Expose only generic messages to tenants while logging detailed errors internally.

How do I handle model updates that change behavior?

Pin model versions in production. Test new versions in a staging environment before rolling out. Use A/B testing for major version changes.

Can tenants bring their own API keys?

Some SaaS products offer a BYOK (Bring Your Own Key) option. This eliminates your API costs but adds complexity. Most SaaS businesses prefer the simplicity of handling API access themselves.