LLM Gateway 架构设计：统一模型路由、限流与成本管控

from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel
from typing import Optional
import time
import hashlib

app = FastAPI()

class ChatRequest(BaseModel):
    model: str
    messages: list[dict]
    temperature: float = 0.7
    max_tokens: Optional[int] = None
    metadata: Optional[dict] = None  # team_id, project_id

class GatewayConfig:
    def __init__(self):
        self.route_rules: dict = {}
        self.rate_limits: dict = {}
        self.fallback_chains: dict = {}
        self.cost_config: dict = {}

class LLMGateway:
    def __init__(self, config: GatewayConfig):
        self.config = config
        self.router = ModelRouter(config.route_rules)
        self.rate_limiter = TokenRateLimiter(config.rate_limits)
        self.cache = SemanticCache()
        self.cost_tracker = CostTracker(config.cost_config)
        self.fallback_manager = FallbackManager(config.fallback_chains)

    async def process_request(self, request: ChatRequest, team_id: str) -> dict:
        # 1. 限流检查
        if not await self.rate_limiter.allow(team_id, request):
            raise HTTPException(429, "Token quota exceeded")

        # 2. 语义缓存查询
        cached = await self.cache.get(request.messages)
        if cached:
            return cached

        # 3. 路由决策
        target_model = await self.router.route(request)

        # 4. 带降级的模型调用
        response = await self.fallback_manager.call_with_fallback(
            target_model, request
        )

        # 5. 异步任务：缓存写入 + 成本记录
        await self.cache.set(request.messages, response)
        await self.cost_tracker.record(team_id, target_model, response)

        return response

from enum import Enum
import tiktoken

class TaskComplexity(Enum):
    SIMPLE = "simple"      # 格式转换、简单问答
    MODERATE = "moderate"  # 摘要、翻译、一般推理
    COMPLEX = "complex"    # 代码生成、数学推理、长文分析

class ContentBasedRouter:
    """基于内容复杂度的智能路由"""

    COMPLEXITY_SIGNALS = {
        "code_keywords": ["implement", "debug", "refactor", "algorithm"],
        "reasoning_keywords": ["analyze", "compare", "evaluate", "prove"],
        "simple_keywords": ["format", "convert", "translate", "summarize"],
    }

    MODEL_MAP = {
        TaskComplexity.SIMPLE: "gpt-4o-mini",
        TaskComplexity.MODERATE: "claude-3-5-haiku",
        TaskComplexity.COMPLEX: "claude-sonnet-4",
    }

    def estimate_complexity(self, messages: list[dict]) -> TaskComplexity:
        last_message = messages[-1]["content"].lower()
        token_count = len(tiktoken.encoding_for_model("gpt-4").encode(last_message))

        # 长 Prompt 通常是复杂任务
        if token_count > 2000:
            return TaskComplexity.COMPLEX

        # 关键词匹配
        for keyword in self.COMPLEXITY_SIGNALS["code_keywords"]:
            if keyword in last_message:
                return TaskComplexity.COMPLEX

        for keyword in self.COMPLEXITY_SIGNALS["reasoning_keywords"]:
            if keyword in last_message:
                return TaskComplexity.MODERATE

        return TaskComplexity.SIMPLE

    async def route(self, request: "ChatRequest") -> str:
        complexity = self.estimate_complexity(request.messages)
        base_model = self.MODEL_MAP[complexity]

        # 如果用户显式指定了模型，尊重用户选择
        if request.model and request.model != "auto":
            return request.model

        return base_model

// TypeScript: 预算感知路由器
interface BudgetConfig {
  teamId: string;
  monthlyBudget: number;       // 月预算（美元）
  currentSpend: number;        // 当月已消耗
  alertThreshold: number;      // 告警阈值（0.8 = 80%）
  hardLimitThreshold: number;  // 硬限制阈值（0.95 = 95%）
}

interface ModelPricing {
  model: string;
  inputCostPer1K: number;   // 每 1K input token 成本
  outputCostPer1K: number;  // 每 1K output token 成本
}

class BudgetAwareRouter {
  private pricing: Map<string, ModelPricing> = new Map([
    ['gpt-4o', { model: 'gpt-4o', inputCostPer1K: 0.0025, outputCostPer1K: 0.01 }],
    ['gpt-4o-mini', { model: 'gpt-4o-mini', inputCostPer1K: 0.00015, outputCostPer1K: 0.0006 }],
    ['claude-sonnet-4', { model: 'claude-sonnet-4', inputCostPer1K: 0.003, outputCostPer1K: 0.015 }],
    ['claude-3-5-haiku', { model: 'claude-3-5-haiku', inputCostPer1K: 0.0008, outputCostPer1K: 0.004 }],
  ]);

  async route(request: ChatRequest, budget: BudgetConfig): Promise<string> {
    const spendRatio = budget.currentSpend / budget.monthlyBudget;

    // 超过硬限制，强制使用最便宜的模型
    if (spendRatio >= budget.hardLimitThreshold) {
      console.warn(`[Budget] Team ${budget.teamId} hit hard limit, forcing cheapest model`);
      return 'gpt-4o-mini';
    }

    // 超过告警阈值，降级到中等模型
    if (spendRatio >= budget.alertThreshold) {
      return this.selectMidTierModel(request);
    }

    // 预算充足，正常路由
    return request.model || 'gpt-4o';
  }

  private selectMidTierModel(request: ChatRequest): string {
    // 根据任务类型选择性价比最优的中等模型
    const isCodeTask = request.messages.some(m =>
      m.content.includes('```') || m.content.includes('code')
    );
    return isCodeTask ? 'claude-3-5-haiku' : 'gpt-4o-mini';
  }
}

import random
from dataclasses import dataclass

@dataclass
class CanaryRule:
    model_a: str           # 基准模型
    model_b: str           # 候选模型
    traffic_percent: float # 候选模型流量比例 (0-1)
    metric_key: str        # 评估指标

class CanaryRouter:
    """模型灰度发布路由器"""

    def __init__(self):
        self.active_canaries: dict[str, CanaryRule] = {}
        self.metrics_collector = MetricsCollector()

    async def route(self, request: "ChatRequest", canary_id: str) -> str:
        rule = self.active_canaries.get(canary_id)
        if not rule:
            return request.model

        # 按流量比例分配
        if random.random() < rule.traffic_percent:
            selected = rule.model_b
            variant = "canary"
        else:
            selected = rule.model_a
            variant = "baseline"

        # 记录分组信息，用于后续分析
        self.metrics_collector.tag_request(
            request_id=request.metadata.get("request_id"),
            variant=variant,
            model=selected
        )
        return selected

import redis.asyncio as redis
import time
import tiktoken

class TokenRateLimiter:
    """基于滑动窗口的 Token 限流器"""

    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.encoder = tiktoken.encoding_for_model("gpt-4")

    def count_tokens(self, messages: list[dict]) -> int:
        total = 0
        for msg in messages:
            total += len(self.encoder.encode(msg.get("content", "")))
            total += 4  # role + formatting overhead
        return total

    async def allow(self, team_id: str, request: "ChatRequest") -> bool:
        """检查是否允许请求通过"""
        token_count = self.count_tokens(request.messages)
        now = time.time()
        window_key = f"ratelimit:token:{team_id}"
        window_size = 60  # 1 分钟滑动窗口

        pipe = self.redis.pipeline()
        # 清理过期条目
        pipe.zremrangebyscore(window_key, 0, now - window_size)
        # 获取当前窗口内的 Token 总量
        pipe.zrangebyscore(window_key, now - window_size, now)
        results = await pipe.execute()

        current_tokens = sum(int(score) for score in results[1]) if results[1] else 0
        limit = await self.get_team_limit(team_id)

        if current_tokens + token_count > limit:
            return False

        # 记录本次请求的 Token 数
        await self.redis.zadd(window_key, {f"{now}:{token_count}": now})
        await self.redis.expire(window_key, window_size + 10)
        return True

    async def get_team_limit(self, team_id: str) -> int:
        """获取团队的 TPM 限制"""
        config = await self.redis.hget("team_limits", team_id)
        return int(config) if config else 100_000  # 默认 100K TPM

    async def record_actual_usage(self, team_id: str, actual_tokens: int):
        """流式响应完成后，校正实际 Token 消耗"""
        now = time.time()
        window_key = f"ratelimit:token:{team_id}"
        await self.redis.zadd(window_key, {f"{now}:output:{actual_tokens}": now})

// TypeScript: 层级化配额管理
interface QuotaConfig {
  organization: {
    monthlyTokens: number;    // 组织级月配额
    maxConcurrent: number;    // 最大并发请求
  };
  team: {
    dailyTokens: number;      // 团队级日配额
    rpm: number;              // 每分钟请求数
    tpm: number;              // 每分钟 Token 数
  };
  user: {
    hourlyTokens: number;     // 用户级小时配额
    rpm: number;
  };
}

class HierarchicalQuotaManager {
  private redis: Redis;

  async checkQuota(
    orgId: string,
    teamId: string,
    userId: string,
    estimatedTokens: number
  ): Promise<{ allowed: boolean; reason?: string }> {
    // 从最细粒度开始检查
    const checks = [
      this.checkUserQuota(userId, estimatedTokens),
      this.checkTeamQuota(teamId, estimatedTokens),
      this.checkOrgQuota(orgId, estimatedTokens),
    ];

    const results = await Promise.all(checks);

    for (const result of results) {
      if (!result.allowed) {
        return result;
      }
    }

    return { allowed: true };
  }

  private async checkTeamQuota(
    teamId: string,
    tokens: number
  ): Promise<{ allowed: boolean; reason?: string }> {
    const key = `quota:team:${teamId}:${this.getTodayKey()}`;
    const current = await this.redis.get(key);
    const used = parseInt(current || '0');
    const limit = await this.getTeamDailyLimit(teamId);

    if (used + tokens > limit) {
      return {
        allowed: false,
        reason: `Team daily quota exceeded: ${used}/${limit} tokens`
      };
    }
    return { allowed: true };
  }

  private getTodayKey(): string {
    return new Date().toISOString().split('T')[0];
  }
}

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
import asyncio

@dataclass
class UsageRecord:
    timestamp: datetime
    team_id: str
    project_id: str
    model: str
    input_tokens: int
    output_tokens: int
    cost_usd: float
    latency_ms: int
    cache_hit: bool = False
    request_id: str = ""

class CostTracker:
    """实时成本追踪与预算管控"""

    PRICING = {
        "gpt-4o": {"input": 2.50, "output": 10.00},         # per 1M tokens
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "claude-sonnet-4": {"input": 3.00, "output": 15.00},
        "claude-3-5-haiku": {"input": 0.80, "output": 4.00},
        "gemini-2.5-pro": {"input": 1.25, "output": 10.00},
    }

    def __init__(self):
        self.records: list[UsageRecord] = []
        self.budget_alerts: dict[str, float] = {}

    def calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        pricing = self.PRICING.get(model)
        if not pricing:
            return 0.0
        input_cost = (input_tokens / 1_000_000) * pricing["input"]
        output_cost = (output_tokens / 1_000_000) * pricing["output"]
        return round(input_cost + output_cost, 6)

    async def record(
        self,
        team_id: str,
        project_id: str,
        model: str,
        input_tokens: int,
        output_tokens: int,
        latency_ms: int,
        cache_hit: bool = False,
    ):
        cost = 0.0 if cache_hit else self.calculate_cost(model, input_tokens, output_tokens)

        record = UsageRecord(
            timestamp=datetime.utcnow(),
            team_id=team_id,
            project_id=project_id,
            model=model,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            cost_usd=cost,
            latency_ms=latency_ms,
            cache_hit=cache_hit,
        )
        self.records.append(record)

        # 检查预算告警
        await self._check_budget_alert(team_id, cost)

    async def _check_budget_alert(self, team_id: str, new_cost: float):
        monthly_spend = await self.get_monthly_spend(team_id)
        budget = self.budget_alerts.get(team_id, float("inf"))

        if monthly_spend / budget > 0.8:
            await self._send_alert(team_id, monthly_spend, budget)

    async def get_monthly_spend(self, team_id: str) -> float:
        # 实际生产中从时序数据库查询
        return sum(
            r.cost_usd for r in self.records
            if r.team_id == team_id
            and r.timestamp.month == datetime.utcnow().month
        )

import numpy as np
from typing import Optional

class SemanticCache:
    """基于向量相似度的语义缓存"""

    def __init__(self, similarity_threshold: float = 0.93):
        self.threshold = similarity_threshold
        self.embeddings: list[np.ndarray] = []
        self.responses: list[dict] = []
        self.ttl_seconds = 3600  # 1 小时过期

    async def get_embedding(self, text: str) -> np.ndarray:
        # 使用轻量级 embedding 模型
        # 生产中推荐 text-embedding-3-small（成本低、速度快）
        response = await embedding_client.create(
            model="text-embedding-3-small",
            input=text
        )
        return np.array(response.data[0].embedding)

    async def get(self, messages: list[dict]) -> Optional[dict]:
        query_text = messages[-1]["content"]
        query_embedding = await self.get_embedding(query_text)

        if not self.embeddings:
            return None

        # 计算余弦相似度
        similarities = np.dot(self.embeddings, query_embedding) / (
            np.linalg.norm(self.embeddings, axis=1) * np.linalg.norm(query_embedding)
        )
        max_idx = np.argmax(similarities)
        max_sim = similarities[max_idx]

        if max_sim >= self.threshold:
            return self.responses[max_idx]
        return None

    async def set(self, messages: list[dict], response: dict):
        query_text = messages[-1]["content"]
        embedding = await self.get_embedding(query_text)
        self.embeddings.append(embedding)
        self.responses.append(response)

import asyncio
from dataclasses import dataclass

@dataclass
class FallbackChain:
    primary: str
    secondary: str
    tertiary: str
    timeout_ms: int = 30000

class FallbackManager:
    """三级降级管理器"""

    DEFAULT_CHAINS = {
        "gpt-4o": FallbackChain("gpt-4o", "claude-sonnet-4", "gpt-4o-mini"),
        "claude-sonnet-4": FallbackChain("claude-sonnet-4", "gpt-4o", "claude-3-5-haiku"),
    }

    async def call_with_fallback(self, model: str, request: "ChatRequest") -> dict:
        chain = self.DEFAULT_CHAINS.get(model, FallbackChain(model, "gpt-4o-mini", "gpt-4o-mini"))

        for i, target in enumerate([chain.primary, chain.secondary, chain.tertiary]):
            try:
                response = await asyncio.wait_for(
                    self._call_model(target, request),
                    timeout=chain.timeout_ms / 1000
                )
                if i > 0:
                    # 记录降级事件
                    await self._record_fallback(model, target, i)
                return response
            except (asyncio.TimeoutError, Exception) as e:
                if i == 2:  # 所有模型都失败
                    raise HTTPException(503, f"All models failed: {str(e)}")
                continue

    async def _call_model(self, model: str, request: "ChatRequest") -> dict:
        # 实际调用模型供应商 API
        ...

    async def _record_fallback(self, original: str, actual: str, level: int):
        # 记录降级指标，触发告警
        ...

from opentelemetry import trace
from opentelemetry.trace import StatusCode

tracer = trace.get_tracer("llm-gateway")

class ObservableGateway:
    """集成 OpenTelemetry 的可观测 Gateway"""

    async def process_request(self, request: "ChatRequest", team_id: str) -> dict:
        with tracer.start_as_current_span("gateway.process") as span:
            span.set_attribute("llm.team_id", team_id)
            span.set_attribute("llm.requested_model", request.model)
            span.set_attribute("llm.message_count", len(request.messages))

            # 路由决策
            with tracer.start_as_current_span("gateway.route") as route_span:
                target_model = await self.router.route(request)
                route_span.set_attribute("llm.routed_model", target_model)

            # 模型调用
            with tracer.start_as_current_span("gateway.inference") as infer_span:
                response = await self.call_model(target_model, request)
                infer_span.set_attribute("llm.input_tokens", response["usage"]["input"])
                infer_span.set_attribute("llm.output_tokens", response["usage"]["output"])
                infer_span.set_attribute("llm.cost_usd", response["cost"])

            span.set_status(StatusCode.OK)
            return response

Phase 1: 代理模式（透传所有请求，收集基线数据）
Phase 2: 启用缓存（验证缓存命中率和正确性）
Phase 3: 启用路由（灰度 10% 流量到智能路由）
Phase 4: 启用限流（逐步收紧配额）
Phase 5: 全量上线（关闭直连通道）