Pain Points of Enterprise AI Services

With the proliferation of AI applications in enterprises, a typical scenario is: the marketing department integrates GPT-4 for copy generation, the R&D team uses DeepSeek for code completion, and the customer service system uses a self-trained intent recognition model. These AI services operate in silos, lacking unified management—API keys are scattered, call volumes cannot be monitored, costs are difficult to allocate, and model switching requires code changes. The goal of AI middleware is to solve these fragmentation issues and build a unified AI capability access layer.

Core Capabilities of AI Gateway

The core of AI middleware is the AI gateway, which should have the following capabilities: Unified API (all models are accessed through a unified OpenAI-compatible interface, and upstream applications do not need to care about the backend model), Intelligent Routing (automatically selects the optimal model based on request type—simple Q&A goes to cheap models, complex reasoning goes to advanced models), Rate Limiting and Quotas (rate limiting and token quota management per tenant/application), Protocol Conversion (converts OpenAI-format requests to native formats of different providers), Caching and Degradation (caches high-frequency requests, degrades to backup models when the model is unavailable).

Core Implementation of AI Gateway

import time, hashlib, json
from collections import defaultdict
from openai import OpenAI

class AIGateway:
    def __init__(self):
        self.models = {}           # Model registry
        self.rate_limits = {}      # Tenant rate limit config
        self.usage = defaultdict(list)  # Usage statistics
        self.cache = {}

    def register_model(self, name, provider, model_id, api_key, base_url, cost_per_1k):
        """Register a model"""
        self.models[name] = {
            "provider": provider, "model_id": model_id,
            "client": OpenAI(api_key=api_key, base_url=base_url),
            "cost_per_1k": cost_per_1k
        }

    def set_rate_limit(self, tenant_id, rpm, tpm):
        """Set tenant rate limit (requests per minute, tokens per minute)"""
        self.rate_limits[tenant_id] = {"rpm": rpm, "tpm": tpm, "requests": [], "tokens": 0}

    def _check_limit(self, tenant_id):
        if tenant_id not in self.rate_limits:
            return True
        lim = self.rate_limits[tenant_id]
        now = time.time()
        lim["requests"] = [t for t in lim["requests"] if now - t < 60]
        if len(lim["requests"]) >= lim["rpm"]:
            return False
        return True

    def route(self, messages, tenant_id, max_cost=None):
        """Intelligent routing—select model based on task complexity"""
        msg_text = json.dumps(messages, ensure_ascii=False)
        if len(msg_text) < 200:
            preferred = ["deepseek-chat", "gpt-3.5-turbo"]  # Simple tasks use cheap models
        else:
            preferred = ["gpt-4", "deepseek-chat"]  # Complex tasks use advanced models

        for model in preferred:
            if model in self.models and self._check_limit(tenant_id):
                return model
        return list(self.models.keys())[0]  # Fallback

    def chat(self, tenant_id, model, messages, **kwargs):
        """Execute LLM call with rate limiting and caching"""
        if not self._check_limit(tenant_id):
            return {"error": "rate limited", "retry_after": 60}

        # Cache check
        cache_key = hashlib.md5(
            json.dumps({"model":model,"msgs":messages},sort_keys=True).encode()
        ).hexdigest()
        if cache_key in self.cache:
            return self.cache[cache_key]

        m = self.models[model]
        resp = m["client"].chat.completions.create(
            model=m["model_id"], messages=messages, **kwargs
        )
        result = resp.choices[0].message.content

        # Update statistics
        self.rate_limits.setdefault(tenant_id,
            {"rpm":9999,"tpm":999999,"requests":[],"tokens":0}
        )["requests"].append(time.time())

        self.cache[cache_key] = result
        return result

    def cost_report(self, tenant_id):
        """Generate cost report"""
        usage = self.usage.get(tenant_id, [])
        return {"total_calls": len(usage), "estimated_cost": sum(u.get("c
ost", 0) for u in usage)}

gw = AIGateway()
gw.register_model("deepseek-chat", "deepseek", "deepseek-chat",
                  "sk-xxx", "https://api.deepseek.com", 0.001)
gw.set_rate_limit("tenant-1", rpm=100, tpm=100000)
print(gw.chat("tenant-1", "deepseek-chat", [{"role":"user","content":"Hello"}]))


Production-Grade Deployment Considerations
AI gateways in production environments require additional attention: high availability (multi-instance deployment + health checks + automatic failover), request queuing (queue requests during traffic spikes rather than rejecting them outright, with configurable queue length and timeout), streaming responses (support SSE passthrough for typewriter effect, handle stream interruptions and reconnections), logging and auditing (record complete information for all requests for cost accounting and troubleshooting, with careful redaction of user data), multi-region deployment (route to nearest region to reduce latency, with cross-region failover).
Comparison of Open-Source AI Gateway Solutions
If you don't want to build an AI gateway from scratch, the following open-source solutions are worth attention: LiteLLM — the most popular AI gateway proxy, supporting unified OpenAI-format interfaces for 100+ model providers, with built-in load balancing, rate limiting, and spend tracking. Portkey — an enhanced AI gateway that additionally provides prompt management, A/B testing, caching, and canary releases. Helicone — an AI gateway focused on observability, offering request logs, cost analysis, and usage dashboards. Recommendations for choosing among the three: if you only need unified API and basic rate limiting → LiteLLM; if you need full LLMOps capabilities → Portkey; if your core need is observability and cost analysis → Helicone. We ultimately chose a combination of LiteLLM plus a self-developed routing module — LiteLLM handles protocol unification and basic capabilities, while the self-developed module handles complex multi-model routing and degradation strategies.
Want to orchestrate this skill chain yourself?
Open in Skill Chain →