AI Gateway的统一接入层设计：多模型路由、限流与成本控制方案

AI Gateway的统一接入层设计：多模型路由、限流与成本控制方案

随着组织内部署的AI模型数量和种类快速增长（GPT-4o、Claude、开源模型、自训练模型），API管理碎片化成为一个突出的工程问题。AI Gateway作为统一接入层，解决了多模型路由、认证聚合、速率限制、成本追踪和故障转移五个核心诉求。本文从架构层面设计一个生产可用的AI Gateway方案，基于OpenAI兼容API规范实现多模型透明代理，并给出基于滑动窗口的分布式限流和基于token使用量的成本归因实现。

一、AI Gateway的核心职责

AI Gateway位于客户端应用和后端模型服务之间，作为所有AI请求的单一入口。其核心职责包括：

统一协议适配：将不同模型提供商（OpenAI、Anthropic、开源自部署）的API格式统一为OpenAI兼容的/v1/chat/completions接口，使上层应用无需感知底层模型的切换。

智能路由：根据请求特征（如token长度、任务类型）、模型可用性和成本预算，将请求路由到最合适的模型。例如，简单分类任务路由到GPT-4o-mini，复杂推理任务路由到GPT-4o/Claude 3.5。

速率限制与配额管理：按用户、API Key、租户等维度实现多层级速率限制，防止单个用户或应用耗尽API配额。

成本追踪与归因：精确记录每次请求的token消耗和成本，支持按项目、团队和用户的成本归因分析。

故障转移：当主模型不可用（超时、限流、返回错误）时，自动切换到备用模型。

二、多模型路由的规则引擎

路由决策需要考虑三个维度：任务特征（复杂度、模态、延迟要求）、成本预算（每请求/每用户限额）和模型能力（准确率、支持的模态）。

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from enum import Enum
import re

class TaskComplexity(Enum):
"""任务复杂度分级，决定模型选择策略。"""
SIMPLE = "simple" # 翻译、摘要、基础问答 → 小模型
MODERATE = "moderate" # 代码生成、中等推理 → 中等模型
COMPLEX = "complex" # 多步推理、数学证明 → 大模型

@dataclass
class ModelEndpoint:
"""模型端点的元数据定义。"""
name: str
provider: str # openai / anthropic / self_hosted
cost_per_1k_input_tokens: float # 千token成本（美元）
cost_per_1k_output_tokens: float
max_context_length: int
capabilities: List[str] = field(default_factory=list)
priority: int = 0 # 同级别模型间的优先级
max_rpm: int = 1000 # 该模型的最大请求速率

class AIRouter:
"""
智能路由器：基于任务特征和成本预算选择最优模型。
"""

def __init__(self):
self.models = [
ModelEndpoint(
name="gpt-4o-mini",
provider="openai",
cost_per_1k_input_tokens=0.00015,
cost_per_1k_output_tokens=0.0006,
max_context_length=128000,
capabilities=["text", "code", "function_calling"],
priority=10,
),
ModelEndpoint(
name="gpt-4o",
provider="openai",
cost_per_1k_input_tokens=0.0025,
cost_per_1k_output_tokens=0.01,
max_context_length=128000,
capabilities=["text", "code", "function_calling", "vision"],
priority=5,
),
ModelEndpoint(
name="claude-3-5-sonnet",
provider="anthropic",
cost_per_1k_input_tokens=0.003,
cost_per_1k_output_tokens=0.015,
max_context_length=200000,
capabilities=["text", "code", "vision", "long_context"],
priority=6,
),
ModelEndpoint(
name="llama-3-70b-self",
provider="self_hosted",
cost_per_1k_input_tokens=0.0, # 自部署零边际成本
cost_per_1k_output_tokens=0.0,
max_context_length=8192,
capabilities=["text", "code"],
priority=3,
max_rpm=500, # 自部署容量有限
),
]
self.fallback_chain = [
"gpt-4o",
"claude-3-5-sonnet",
"llama-3-70b-self",
]

def estimate_complexity(self, messages: List[dict]) -> TaskComplexity:
"""
基于输入消息估计任务复杂度。
简单的启发式规则——实际系统应使用更复杂的分类器。
"""
# 合并所有消息的文本
full_text = " ".join(
msg.get("content", "") for msg in messages
if isinstance(msg.get("content"), str)
)

complexity_signals = {
"step by step": 2,
"explain your reasoning": 2,
"prove": 3,
"derive": 3,
"write code for": 2,
"analyze": 1,
"multimodal": 2,
}

score = 0
for signal, weight in complexity_signals.items():
if signal.lower() in full_text.lower():
score += weight

if score >= 3:
return TaskComplexity.COMPLEX
elif score >= 1:
return TaskComplexity.MODERATE
else:
return TaskComplexity.SIMPLE

def route(
self,
messages: List[dict],
user_budget_remaining: float = float("inf"),
preferred_model: Optional[str] = None,
) -> Tuple[ModelEndpoint, str]:
"""
智能路由决策。

Returns:
(选中的模型端点, 决策原因)
"""
# 1. 如果用户指定了 preferred_model，优先使用
if preferred_model:
for m in self.models:
if m.name == preferred_model:
return m, f"用户指定模型: {m.name}"

# 2. 估计任务复杂度
complexity = self.estimate_complexity(messages)

# 3. 基于复杂度和成本选择模型
candidates = []
for m in self.models:
# 筛选：成本在预算内
estimated_cost = (
len(str(messages)) / 4 * m.cost_per_1k_input_tokens / 1000
)
if estimated_cost > user_budget_remaining:
continue

# 基于复杂度打分
if complexity == TaskComplexity.SIMPLE:
score = -m.cost_per_1k_input_tokens * 1000 # 越便宜越好
elif complexity == TaskComplexity.MODERATE:
score = m.priority
else: # COMPLEX
# 优先选择有 "long_context" 或高优先级的模型
score = m.priority + (
3 if "long_context" in m.capabilities else 0
)

candidates.append((score, m))

if not candidates:
raise ValueError("没有满足条件的模型可用")

# 选择得分最高的模型
candidates.sort(key=lambda x: x[0], reverse=True)
best_model = candidates[0][1]

return best_model, (
f"复杂度={complexity.value}, "
f"选中模型={best_model.name}, "
f"成本=${best_model.cost_per_1k_input_tokens}/1K tokens"
)

三、滑动窗口限流的实现

AI Gateway的限流算法需要考虑一个特殊因素：模型API通常按RPM（Requests Per Minute）和TPM（Tokens Per Minute）双维度限流。因此，Gateway的限流也需要双维度设计。

滑动窗口算法（Sliding Window Log）比固定窗口和令牌桶更适合API限流场景——它消除了固定窗口的"边界突发"问题，同时保持较高的实现效率。核心思想是维护一个按时间戳排序的请求日志，每次新请求到达时，删除窗口外的旧请求记录，检查窗口内的请求数是否超过限制。

# 基于 Redis Sorted Set 的滑动窗口限流（生产级实现）
import time
import redis
from typing import Optional

class SlidingWindowRateLimiter:
"""
基于 Redis Sorted Set 的滑动窗口限流器。
支持 RPM（请求数）和 TPM（Token 数）双维度限流。
"""

def __init__(
self,
redis_client: redis.Redis,
window_size_seconds: int = 60, # 窗口大小（秒）
):
self.redis = redis_client
self.window_size = window_size_seconds

def is_allowed(
self,
key: str, # 限流键（如 "user:123:gpt-4o"）
max_requests: int, # 窗口内最大请求数
max_tokens: Optional[int] = None, # 窗口内最大 Token 数
estimated_tokens: int = 0, # 本次请求的预估 Token 数
) -> Tuple[bool, dict]:
"""
检查请求是否被允许。

使用 Redis Sorted Set：
– member: 请求的唯一 ID（timestamp + random）
– score: Unix 时间戳
"""
now = time.time()
window_start = now – self.window_size

pipeline = self.redis.pipeline()

# 1. 删除窗口外的旧请求
pipeline.zremrangebyscore(key, 0, window_start)

# 2. 统计当前窗口内的请求数
pipeline.zcard(key)

# 3. 如果设置了 token 限制，统计 token 总数
# token 数据存储在 Hash 中: {request_id: token_count}
token_key = f"{key}:tokens"
if max_tokens:
pipeline.hgetall(token_key)

results = pipeline.execute()
# results[0]: zremrangebyscore 删除数量
# results[1]: zcard 当前请求数

current_requests = results[1]
current_tokens = 0

# 解析 token 统计
if max_tokens and len(results) > 2:
token_data = results[2]
current_tokens = sum(int(v) for v in token_data.values())

# 4. 判断是否允许
request_allowed = current_requests < max_requests
tokens_allowed = (
not max_tokens or
current_tokens + estimated_tokens <= max_tokens
)

allowed = request_allowed and tokens_allowed

if allowed:
# 5. 记录本次请求
request_id = f"{now}:{hash(str(now))}"
pipeline.zadd(key, {request_id: now})
if max_tokens:
pipeline.hset(
token_key, request_id, estimated_tokens
)
pipeline.expire(token_key, self.window_size + 10)
pipeline.expire(key, self.window_size + 10)
pipeline.execute()

return allowed, {
"current_requests": current_requests,
"max_requests": max_requests,
"remaining_requests": max_requests – current_requests,
"current_tokens": current_tokens,
"reset_in_seconds": self.window_size – (now – window_start),
}

四、成本追踪与归因

成本追踪需要精确到每次请求。核心是在Gateway层面记录每次请求的输入/输出token数，并基于模型定价计算实际成本。

token计数依赖不同模型的tokenizer。对于OpenAI模型，可以直接从响应中的usage字段获取；对于自部署的开源模型，需要在Gateway中集成token计数器。

成本归因的核心是在请求中附加"成本中心"标签。通过API Key或请求Header中的X-Project-Id、X-Cost-Center字段将每次请求关联到具体的项目或团队。

五、总结

AI Gateway作为统一接入层，通过多模型路由实现智能模型选择（基于任务复杂度和成本预算），通过滑动窗口限流保障配额公平分配，通过token级别的成本追踪实现精细化的模型使用成本归因。在组织内部署多个AI模型的场景中，Gateway的存在将"选择哪个模型"和"控制成本"的决策从应用开发者转移到基础设施层，实现了关注点分离和集中管理。OpenAI兼容API的广泛采用为Gateway的协议适配层提供了事实标准——只需适配到这一接口，所有应用即可透明使用后端任何模型。