AI API网关与大模型路由完全教程

引言

随着大语言模型（LLM）生态的爆发式增长，企业面临一个现实挑战：如何同时管理OpenAI、Anthropic、Google、开源模型等多个LLM服务，并在它们之间实现智能路由、成本控制和高可用保障？答案就是——AI API网关。

本教程将从架构设计到实战部署，全面讲解AI API网关与大模型路由的核心技术，帮助你构建一个企业级的LLM统一接入层。

1. 什么是AI API网关

1.1 核心概念

AI API网关是位于客户端与多个LLM后端之间的中间层服务，它承担以下职责：

• 统一路由：将请求路由到合适的模型（如根据任务类型、成本、延迟）
• 负载均衡：在多个同模型实例间分配请求
• 故障转移：当主模型不可用时自动切换到备用模型
• API Key管理：统一管理多个LLM提供商的密钥
• 速率限制：控制每个用户/团队的请求频率
• 成本追踪：监控和预警各模型的Token消耗与费用
• 日志审计：记录所有请求/响应，便于合规和调试
• 缓存：缓存重复请求的响应，降低成本和延迟

1.2 架构总览

┌─────────────┐     ┌──────────────────────┐     ┌─────────────────┐
│   Client    │────▶│   AI API Gateway     │────▶│  OpenAI API     │
│  (App/Bot)  │     │                      │     │  Claude API     │
└─────────────┘     │  - 路由策略          │     │  Gemini API     │
                    │  - 负载均衡          │     │  本地模型(Ollama)│
                    │  - 故障转移          │     │  ...            │
                    │  - 认证/鉴权         │     └─────────────────┘
                    │  - 日志/监控          │
                    │  - 缓存              │
                    │  - 成本控制          │
                    └──────────────────────┘

2. 主流AI API网关对比

2.1 LiteLLM

LiteLLM 是目前最流行的开源LLM代理网关，支持100+模型提供商，提供统一的OpenAI兼容接口。

核心特性：

• 支持100+模型（OpenAI、Anthropic、Google、Cohere、HuggingFace、Ollama等）
• OpenAI兼容API格式
• 内置负载均衡和故障转移
• 细粒度的预算和速率限制
• 完善的日志和回调系统
• 支持团队/虚拟Key管理

安装与基础配置：

pip install litellm[proxy]

基础代理配置文件 litellm_config.yaml：

model_list:
  - model_name: gpt-4o
    litellm_params:
      model: openai/gpt-4o
      api_key: os.environ/OPENAI_API_KEY
      api_base: https://api.openai.com/v1

  - model_name: gpt-4o
    litellm_params:
      model: openai/gpt-4o
      api_key: os.environ/OPENAI_API_KEY_2
      api_base: https://api.openai.com/v1

  - model_name: claude-sonnet
    litellm_params:
      model: anthropic/claude-3-5-sonnet-20241022
      api_key: os.environ/ANTHROPIC_API_KEY

  - model_name: local-llama
    litellm_params:
      model: ollama/llama3.1
      api_base: http://localhost:11434

router_settings:
  routing_strategy: "simple-shuffle"  # 负载均衡策略
  num_retries: 3                      # 重试次数
  timeout: 60                         # 超时时间（秒）
  allowed_fails: 2                    # 允许失败次数
  cooldown_time: 30                   # 冷却时间（秒）

general_settings:
  master_key: os.environ/LITELLM_MASTER_KEY
  database_url: os.environ/DATABASE_URL  # PostgreSQL用于持久化

启动LiteLLM Proxy：

litellm --config litellm_config.yaml --port 4000

客户端调用：

import openai

# 使用与OpenAI完全兼容的接口
client = openai.OpenAI(
    api_key="your-litellm-key",
    base_url="http://localhost:4000/v1"
)

response = client.chat.completions.create(
    model="gpt-4o",  # 或 "claude-sonnet"、"local-llama"
    messages=[
        {"role": "user", "content": "用Python实现快速排序"}
    ]
)
print(response.choices[0].message.content)

2.2 OpenRouter

OpenRouter 是一个商业化的LLM路由服务，无需自建基础设施，提供统一API访问数百个模型。

核心特性：

• 访问200+模型，无需分别注册
• 自动模型路由（根据任务选择最佳模型）
• 按Token计费，价格透明
• 支持模型回退链
• 隐私模式（不存储请求数据）

使用方式：

import openai

client = openai.OpenAI(
    base_url="https://openrouter.ai/api/v1",
    api_key="sk-or-your-openrouter-key",
)

# 直接指定模型，格式为 provider/model
response = client.chat.completions.create(
    model="anthropic/claude-3.5-sonnet",
    messages=[
        {"role": "user", "content": "解释什么是Transformer架构"}
    ],
    extra_body={
        "route": "fallback",  # 启用回退路由
        "models": [
            "anthropic/claude-3.5-sonnet",
            "openai/gpt-4o",
            "google/gemini-pro-1.5"
        ]
    }
)

2.3 Portkey

Portkey 是一个面向企业的AI网关平台，提供可观测性、可靠性和安全性。

核心特性：

• 多模型自动回退
• 语义缓存（基于嵌入的相似度缓存）
• 完整的请求追踪和可观测性
• 细粒度的Guardrails（输入/输出安全护栏）
• A/B测试和金丝雀发布

安装与使用：

pip install portkey-ai

from portkey_ai import Portkey

# 初始化客户端
portkey = Portkey(
    api_key="your-portkey-api-key",
    virtual_key="openai-xxx"  # 虚拟Key，隐藏真实API Key
)

# 基础调用
response = portkey.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "什么是RAG？"}]
)

# 配置自动回退
fallback_config = {
    "strategy": {"mode": "fallback"},
    "targets": [
        {"virtual_key": "openai-key", "model": "gpt-4o"},
        {"virtual_key": "anthropic-key", "model": "claude-3-5-sonnet-20241022"},
        {"virtual_key": "google-key", "model": "gemini-pro"},
    ]
}

response = portkey.with_config(fallback_config).chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}]
)

2.4 综合对比表

3. 多模型统一路由策略

3.1 基于任务类型的路由

不同任务适合不同的模型。我们可以根据请求中的任务标签进行智能路由：

from litellm import Router
import litellm

# 定义模型路由配置
model_list = [
    {
        "model_name": "coding-task",
        "litellm_params": {
            "model": "anthropic/claude-3-5-sonnet-20241022",
            "api_key": "your-key"
        }
    },
    {
        "model_name": "chat-task",
        "litellm_params": {
            "model": "openai/gpt-4o-mini",
            "api_key": "your-key"
        }
    },
    {
        "model_name": "creative-task",
        "litellm_params": {
            "model": "openai/gpt-4o",
            "api_key": "your-key"
        }
    }
]

router = Router(model_list=model_list)

# 根据任务类型路由
TASK_MODEL_MAP = {
    "code_generation": "coding-task",
    "code_review": "coding-task",
    "casual_chat": "chat-task",
    "summarization": "chat-task",
    "creative_writing": "creative-task",
    "translation": "creative-task",
}

async def route_by_task(task_type: str, messages: list):
    model_name = TASK_MODEL_MAP.get(task_type, "chat-task")
    response = await router.acompletion(
        model=model_name,
        messages=messages
    )
    return response

3.2 基于成本的路由

当预算有限时，可以设计基于成本的路由策略：

from litellm import Router

model_list = [
    {
        "model_name": "cost-optimized",
        "litellm_params": {
            "model": "openai/gpt-4o-mini",    # 低成本
            "api_key": "your-key"
        }
    },
    {
        "model_name": "cost-optimized",
        "litellm_params": {
            "model": "anthropic/claude-3-5-sonnet-20241022",  # 高性能
            "api_key": "your-key"
        }
    }
]

router = Router(
    model_list=model_list,
    routing_strategy="cost-based-routing"  # 基于成本的路由
)

# 默认使用低成本模型
response = await router.acompletion(
    model="cost-optimized",
    messages=[{"role": "user", "content": "简单问候"}]
)

3.3 基于延迟的路由

对实时性要求高的场景，可以基于历史延迟数据选择最快的模型：

router = Router(
    model_list=model_list,
    routing_strategy="latency-based-routing",  # 基于延迟的路由
    latency_tracking_cache_ttl: 300  # 缓存延迟数据5分钟
)

3.4 自定义路由函数

对于更复杂的路由需求，可以编写自定义路由逻辑：

import time
from litellm import Router

router = Router(model_list=model_list)

async def smart_router(
    messages: list,
    user_tier: str = "free",
    priority: str = "balanced"
):
    """智能路由：综合考虑用户等级、优先级和时间因素"""
    
    hour = time.localtime().tm_hour
    
    # 免费用户在高峰期使用轻量模型
    if user_tier == "free" and 9 <= hour <= 18:
        model = "chat-task"  # gpt-4o-mini
    # 付费用户始终使用高级模型
    elif user_tier == "premium":
        if priority == "quality":
            model = "creative-task"  # gpt-4o
        elif priority == "speed":
            model = "coding-task"    # claude
        else:
            model = "creative-task"
    else:
        model = "chat-task"
    
    response = await router.acompletion(
        model=model,
        messages=messages,
        user=user_tier
    )
    return response

4. 负载均衡与故障转移

4.1 LiteLLM负载均衡策略

LiteLLM内置多种负载均衡策略：

# litellm_config.yaml
router_settings:
  # 策略选项：
  # - simple-shuffle: 随机分配
  # - least-busy: 最少繁忙实例
  # - latency-based-routing: 基于延迟
  # - cost-based-routing: 基于成本
  # - usage-based-routing: 基于使用量
  routing_strategy: "least-busy"
  
  # 健康检查
  allowed_fails: 2           # 连续失败次数后标记为不健康
  cooldown_time: 30          # 不健康模型的冷却时间（秒）
  retry_after: 5             # 重试等待时间
  
  # 超时设置
  timeout: 60
  max_parallel_requests: 100 # 最大并发请求

4.2 手动实现故障转移

import asyncio
from typing import Optional

class LLMFailover:
    """自定义LLM故障转移管理器"""
    
    def __init__(self):
        self.providers = [
            {"name": "openai", "model": "gpt-4o", "priority": 1, "healthy": True},
            {"name": "anthropic", "model": "claude-3-5-sonnet", "priority": 2, "healthy": True},
            {"name": "google", "model": "gemini-pro", "priority": 3, "healthy": True},
        ]
        self.failure_counts = {}
    
    async def call_with_failover(self, messages: list, **kwargs) -> Optional[dict]:
        """按优先级调用，失败时自动切换到下一个"""
        sorted_providers = sorted(
            [p for p in self.providers if p["healthy"]],
            key=lambda x: x["priority"]
        )
        
        for provider in sorted_providers:
            try:
                response = await self._call_provider(provider, messages, **kwargs)
                # 成功，重置失败计数
                self.failure_counts[provider["name"]] = 0
                return response
            except Exception as e:
                print(f"[{provider['name']}] 调用失败: {e}")
                count = self.failure_counts.get(provider["name"], 0) + 1
                self.failure_counts[provider["name"]] = count
                
                if count >= 3:
                    provider["healthy"] = False
                    print(f"[{provider['name']}] 标记为不健康，进入冷却")
                    asyncio.create_task(self._cooldown(provider))
        
        raise Exception("所有Provider均不可用")
    
    async def _call_provider(self, provider: dict, messages: list, **kwargs):
        """调用具体的Provider"""
        import openai
        client = openai.AsyncOpenAI(api_key=f"your-{provider['name']}-key")
        response = await client.chat.completions.create(
            model=provider["model"],
            messages=messages,
            **kwargs
        )
        return response
    
    async def _cooldown(self, provider: dict):
        """冷却后恢复健康状态"""
        await asyncio.sleep(60)
        provider["healthy"] = True
        self.failure_counts[provider["name"]] = 0
        print(f"[{provider['name']}] 恢复健康状态")

# 使用示例
failover = LLMFailover()
result = await failover.call_with_failover(
    messages=[{"role": "user", "content": "Hello!"}]
)

5. API Key管理与安全

5.1 虚拟Key管理

不要直接暴露真实API Key给客户端，使用虚拟Key进行映射：

import hashlib
import secrets
import sqlite3
from datetime import datetime

class VirtualKeyManager:
    """虚拟Key管理器"""
    
    def __init__(self, db_path: str = "keys.db"):
        self.db = sqlite3.connect(db_path)
        self._init_db()
    
    def _init_db(self):
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS virtual_keys (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                virtual_key TEXT UNIQUE NOT NULL,
                real_key_name TEXT NOT NULL,
                team TEXT DEFAULT 'default',
                budget_limit REAL DEFAULT 100.0,
                budget_used REAL DEFAULT 0.0,
                rate_limit_per_min INTEGER DEFAULT 60,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                is_active BOOLEAN DEFAULT 1
            )
        """)
        self.db.commit()
    
    def create_key(self, team: str, real_key_name: str, 
                   budget_limit: float = 100.0,
                   rate_limit: int = 60) -> str:
        """创建虚拟Key"""
        virtual_key = f"sk-{secrets.token_hex(32)}"
        self.db.execute(
            """INSERT INTO virtual_keys 
               (virtual_key, real_key_name, team, budget_limit, rate_limit_per_min)
               VALUES (?, ?, ?, ?, ?)""",
            (virtual_key, real_key_name, team, budget_limit, rate_limit)
        )
        self.db.commit()
        return virtual_key
    
    def validate_key(self, virtual_key: str) -> dict | None:
        """验证虚拟Key并返回配置"""
        cursor = self.db.execute(
            """SELECT real_key_name, team, budget_limit, budget_used, 
                      rate_limit_per_min, is_active
               FROM virtual_keys WHERE virtual_key = ?""",
            (virtual_key,)
        )
        row = cursor.fetchone()
        if not row or not row[5]:
            return None
        return {
            "real_key_name": row[0],
            "team": row[1],
            "budget_limit": row[2],
            "budget_used": row[3],
            "rate_limit": row[4]
        }
    
    def update_usage(self, virtual_key: str, cost: float):
        """更新使用量"""
        self.db.execute(
            "UPDATE virtual_keys SET budget_used = budget_used + ? WHERE virtual_key = ?",
            (cost, virtual_key)
        )
        self.db.commit()

# 使用示例
manager = VirtualKeyManager()

# 为团队创建Key
key = manager.create_key(
    team="engineering",
    real_key_name="OPENAI_API_KEY",
    budget_limit=500.0,
    rate_limit=120
)
print(f"Virtual Key: {key}")

# 验证Key
config = manager.validate_key(key)
if config:
    print(f"Team: {config['team']}, Budget: ${config['budget_used']}/{config['budget_limit']}")

5.2 密钥轮转

import os
from datetime import datetime, timedelta

class KeyRotator:
    """API Key自动轮转管理"""
    
    def __init__(self):
        self.keys = {}  # provider -> list of keys
        self.current_index = {}
    
    def add_key(self, provider: str, key: str, expires_at: datetime = None):
        if provider not in self.keys:
            self.keys[provider] = []
            self.current_index[provider] = 0
        self.keys[provider].append({
            "key": key,
            "expires_at": expires_at,
            "created_at": datetime.now()
        })
    
    def get_current_key(self, provider: str) -> str:
        """获取当前有效的Key，自动轮转过期Key"""
        if provider not in self.keys:
            raise ValueError(f"No keys for provider: {provider}")
        
        keys = self.keys[provider]
        idx = self.current_index[provider]
        
        # 检查当前Key是否过期
        current = keys[idx]
        if current["expires_at"] and datetime.now() > current["expires_at"]:
            idx = (idx + 1) % len(keys)
            self.current_index[provider] = idx
            print(f"[KeyRotator] Rotated to key index {idx} for {provider}")
        
        return keys[idx]["key"]

# 使用
rotator = KeyRotator()
rotator.add_key("openai", "sk-key-1", expires_at=datetime.now() + timedelta(days=30))
rotator.add_key("openai", "sk-key-2", expires_at=datetime.now() + timedelta(days=60))

6. 速率限制与配额控制

6.1 基于Redis的速率限制

import time
import redis

class RateLimiter:
    """基于Redis的滑动窗口速率限制器"""
    
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
    
    def is_allowed(
        self,
        key: str,
        max_requests: int,
        window_seconds: int
    ) -> tuple[bool, dict]:
        """检查请求是否在速率限制内"""
        now = time.time()
        window_start = now - window_seconds
        pipe = self.redis.pipeline()
        
        # 清除窗口外的记录
        pipe.zremrangebyscore(key, 0, window_start)
        # 添加当前请求
        pipe.zadd(key, {str(now): now})
        # 统计窗口内的请求数
        pipe.zcard(key)
        # 设置Key过期
        pipe.expire(key, window_seconds)
        
        results = pipe.execute()
        request_count = results[2]
        
        allowed = request_count <= max_requests
        remaining = max(0, max_requests - request_count)
        
        return allowed, {
            "limit": max_requests,
            "remaining": remaining,
            "reset_at": int(now + window_seconds)
        }

# 使用示例
limiter = RateLimiter()

# 检查用户速率（每分钟60次）
allowed, info = limiter.is_allowed(
    key="rate:user:123",
    max_requests=60,
    window_seconds=60
)

if not allowed:
    print(f"速率限制！重置时间: {info['reset_at']}")
else:
    print(f"允许请求，剩余: {info['remaining']}")

6.2 Token配额控制

class TokenQuotaManager:
    """Token配额管理器"""
    
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
    
    def check_quota(
        self,
        user_id: str,
        estimated_tokens: int,
        daily_limit: int,
        monthly_limit: int
    ) -> tuple[bool, str]:
        """检查用户Token配额"""
        today = time.strftime("%Y-%m-%d")
        month = time.strftime("%Y-%m")
        
        daily_key = f"quota:{user_id}:daily:{today}"
        monthly_key = f"quota:{user_id}:monthly:{month}"
        
        daily_used = int(self.redis.get(daily_key) or 0)
        monthly_used = int(self.redis.get(monthly_key) or 0)
        
        if daily_used + estimated_tokens > daily_limit:
            return False, f"日配额不足: 已用{daily_used}, 限额{daily_limit}"
        
        if monthly_used + estimated_tokens > monthly_limit:
            return False, f"月配额不足: 已用{monthly_used}, 限额{monthly_limit}"
        
        return True, "OK"
    
    def record_usage(self, user_id: str, tokens_used: int):
        """记录Token使用量"""
        today = time.strftime("%Y-%m-%d")
        month = time.strftime("%Y-%m")
        
        daily_key = f"quota:{user_id}:daily:{today}"
        monthly_key = f"quota:{user_id}:monthly:{month}"
        
        pipe = self.redis.pipeline()
        pipe.incrby(daily_key, tokens_used)
        pipe.expire(daily_key, 86400 * 2)  # 保留2天
        pipe.incrby(monthly_key, tokens_used)
        pipe.expire(monthly_key, 86400 * 35)  # 保留35天
        pipe.execute()

7. 请求/响应日志与审计

7.1 结构化日志系统

import json
import time
import uuid
from datetime import datetime
from dataclasses import dataclass, asdict
from typing import Optional

@dataclass
class RequestLog:
    request_id: str
    timestamp: str
    user_id: str
    team: str
    model: str
    provider: str
    messages_count: int
    total_tokens: int
    prompt_tokens: int
    completion_tokens: int
    cost_usd: float
    latency_ms: float
    status: str  # success, error, timeout
    error_message: Optional[str] = None
    
    def to_json(self) -> str:
        return json.dumps(asdict(self), ensure_ascii=False)

class AuditLogger:
    """审计日志管理器"""
    
    def __init__(self, log_file: str = "llm_audit.jsonl"):
        self.log_file = log_file
    
    def log_request(self, log: RequestLog):
        """记录请求日志"""
        with open(self.log_file, "a") as f:
            f.write(log.to_json() + "\n")
    
    def get_user_costs(
        self, user_id: str, start_date: str, end_date: str
    ) -> dict:
        """统计用户成本"""
        total_cost = 0.0
        total_tokens = 0
        request_count = 0
        
        with open(self.log_file, "r") as f:
            for line in f:
                entry = json.loads(line)
                if (entry["user_id"] == user_id and 
                    start_date <= entry["timestamp"][:10] <= end_date):
                    total_cost += entry["cost_usd"]
                    total_tokens += entry["total_tokens"]
                    request_count += 1
        
        return {
            "total_cost": round(total_cost, 4),
            "total_tokens": total_tokens,
            "request_count": request_count
        }

# 使用示例
logger = AuditLogger()

log = RequestLog(
    request_id=str(uuid.uuid4()),
    timestamp=datetime.now().isoformat(),
    user_id="user_123",
    team="engineering",
    model="gpt-4o",
    provider="openai",
    messages_count=3,
    total_tokens=1500,
    prompt_tokens=1000,
    completion_tokens=500,
    cost_usd=0.015,
    latency_ms=2300,
    status="success"
)
logger.log_request(log)

8. 成本追踪与预算告警

8.1 实时成本监控

import asyncio
from datetime import datetime, timedelta

class CostMonitor:
    """成本监控与告警"""
    
    # 各模型的每1K Token价格（输入/输出）
    MODEL_PRICING = {
        "gpt-4o": {"input": 0.0025, "output": 0.01},
        "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
        "claude-3-5-sonnet": {"input": 0.003, "output": 0.015},
        "claude-3-haiku": {"input": 0.00025, "output": 0.00125},
        "gemini-pro": {"input": 0.000125, "output": 0.000375},
    }
    
    def __init__(self, alert_callback=None):
        self.daily_costs = {}  # team -> cost
        self.monthly_costs = {}
        self.alerts_sent = set()
        self.alert_callback = alert_callback
    
    def calculate_cost(
        self, model: str, prompt_tokens: int, completion_tokens: int
    ) -> float:
        """计算单次请求成本"""
        pricing = self.MODEL_PRICING.get(model)
        if not pricing:
            return 0.0
        return (
            (prompt_tokens / 1000) * pricing["input"] +
            (completion_tokens / 1000) * pricing["output"]
        )
    
    def record_cost(self, team: str, cost: float):
        """记录成本并检查告警"""
        today = datetime.now().strftime("%Y-%m-%d")
        month = datetime.now().strftime("%Y-%m")
        
        daily_key = f"{team}:{today}"
        monthly_key = f"{team}:{month}"
        
        self.daily_costs[daily_key] = self.daily_costs.get(daily_key, 0) + cost
        self.monthly_costs[monthly_key] = self.monthly_costs.get(monthly_key, 0) + cost
        
        # 检查告警阈值
        self._check_alerts(team)
    
    def _check_alerts(self, team: str):
        """检查预算告警"""
        today = datetime.now().strftime("%Y-%m-%d")
        month = datetime.now().strftime("%Y-%m")
        
        daily_cost = self.daily_costs.get(f"{team}:{today}", 0)
        monthly_cost = self.monthly_costs.get(f"{team}:{month}", 0)
        
        # 日预算告警
        if daily_cost > 50 and f"{team}:daily:50" not in self.alerts_sent:
            self._send_alert(team, "daily", daily_cost, 50)
            self.alerts_sent.add(f"{team}:daily:50")
        
        if daily_cost > 100 and f"{team}:daily:100" not in self.alerts_sent:
            self._send_alert(team, "daily", daily_cost, 100)
            self.alerts_sent.add(f"{team}:daily:100")
        
        # 月预算告警
        if monthly_cost > 1000 and f"{team}:monthly:1000" not in self.alerts_sent:
            self._send_alert(team, "monthly", monthly_cost, 1000)
            self.alerts_sent.add(f"{team}:monthly:1000")
    
    def _send_alert(self, team: str, period: str, current: float, threshold: float):
        """发送告警"""
        message = (
            f"⚠️ 预算告警\n"
            f"团队: {team}\n"
            f"周期: {period}\n"
            f"当前: ${current:.2f}\n"
            f"阈值: ${threshold:.2f}"
        )
        print(message)
        if self.alert_callback:
            self.alert_callback(message)
    
    def get_report(self, team: str) -> dict:
        """获取成本报告"""
        today = datetime.now().strftime("%Y-%m-%d")
        month = datetime.now().strftime("%Y-%m")
        
        return {
            "daily_cost": round(self.daily_costs.get(f"{team}:{today}", 0), 4),
            "monthly_cost": round(self.monthly_costs.get(f"{team}:{month}", 0), 4),
            "daily_budget": 100,
            "monthly_budget": 5000
        }

9. 缓存策略

9.1 精确缓存

import hashlib
import json
import redis

class ExactCache:
    """精确匹配缓存"""
    
    def __init__(self, redis_url: str = "redis://localhost:6379", ttl: int = 3600):
        self.redis = redis.from_url(redis_url)
        self.ttl = ttl
    
    def _generate_key(self, model: str, messages: list) -> str:
        """生成缓存Key"""
        content = json.dumps({
            "model": model,
            "messages": messages
        }, sort_keys=True)
        return f"llm_cache:{hashlib.sha256(content.encode()).hexdigest()}"
    
    def get(self, model: str, messages: list) -> dict | None:
        """查询缓存"""
        key = self._generate_key(model, messages)
        cached = self.redis.get(key)
        if cached:
            return json.loads(cached)
        return None
    
    def set(self, model: str, messages: list, response: dict):
        """设置缓存"""
        key = self._generate_key(model, messages)
        self.redis.setex(key, self.ttl, json.dumps(response))
    
    def get_stats(self) -> dict:
        """获取缓存统计"""
        keys = self.redis.keys("llm_cache:*")
        return {
            "total_entries": len(keys),
            "memory_used": sum(self.redis.memory_usage(k) or 0 for k in keys[:100])
        }

9.2 语义缓存

import numpy as np
from typing import Optional

class SemanticCache:
    """基于嵌入向量的语义缓存"""
    
    def __init__(self, similarity_threshold: float = 0.95):
        self.cache = []  # list of (embedding, response)
        self.similarity_threshold = similarity_threshold
    
    async def get_embedding(self, text: str) -> list[float]:
        """获取文本嵌入向量"""
        # 这里可以使用OpenAI、Cohere等嵌入模型
        import openai
        client = openai.AsyncOpenAI()
        response = await client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return response.data[0].embedding
    
    def cosine_similarity(self, a: list, b: list) -> float:
        """计算余弦相似度"""
        a, b = np.array(a), np.array(b)
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
    
    async def get(self, query: str) -> Optional[dict]:
        """语义查询缓存"""
        query_embedding = await self.get_embedding(query)
        
        best_match = None
        best_similarity = 0
        
        for embedding, response in self.cache:
            sim = self.cosine_similarity(query_embedding, embedding)
            if sim > best_similarity:
                best_similarity = sim
                best_match = response
        
        if best_similarity >= self.similarity_threshold:
            print(f"语义缓存命中！相似度: {best_similarity:.4f}")
            return best_match
        
        return None
    
    async def set(self, query: str, response: dict):
        """添加到语义缓存"""
        embedding = await self.get_embedding(query)
        self.cache.append((embedding, response))
        
        # 限制缓存大小
        if len(self.cache) > 10000:
            self.cache = self.cache[-5000:]

10. 企业级多租户部署

10.1 多租户架构设计

from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.security import HTTPBearer
import jwt

app = FastAPI()
security = HTTPBearer()

# 租户配置
TENANT_CONFIG = {
    "tenant_a": {
        "team_id": "team_a",
        "models": ["gpt-4o", "gpt-4o-mini"],
        "daily_budget": 100,
        "rate_limit": 60,
        "priority": "high"
    },
    "tenant_b": {
        "team_id": "team_b",
        "models": ["gpt-4o-mini", "claude-haiku"],
        "daily_budget": 50,
        "rate_limit": 30,
        "priority": "normal"
    }
}

async def get_tenant(request: Request, token = Depends(security)):
    """从JWT Token中解析租户信息"""
    try:
        payload = jwt.decode(
            token.credentials,
            "your-secret-key",
            algorithms=["HS256"]
        )
        tenant_id = payload.get("tenant_id")
        if tenant_id not in TENANT_CONFIG:
            raise HTTPException(status_code=403, detail="Unknown tenant")
        return {"tenant_id": tenant_id, **TENANT_CONFIG[tenant_id]}
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

@app.post("/v1/chat/completions")
async def chat_completions(request: Request, tenant = Depends(get_tenant)):
    """多租户聊天接口"""
    body = await request.json()
    model = body.get("model")
    
    # 检查模型权限
    if model not in tenant["models"]:
        raise HTTPException(
            status_code=403,
            detail=f"Tenant {tenant['tenant_id']} cannot access model {model}"
        )
    
    # 检查预算和速率限制（使用前面的RateLimiter和CostMonitor）
    # ... 
    
    # 转发请求到实际的LLM
    # ...
    
    return {"status": "ok", "tenant": tenant["tenant_id"]}

11. 与Kong/Nginx集成

11.1 Nginx反向代理配置

upstream litellm_backend {
    server 127.0.0.1:4000;
    # 多实例负载均衡
    server 127.0.0.1:4001;
    server 127.0.0.1:4002;
}

server {
    listen 443 ssl;
    server_name llm-api.yourcompany.com;
    
    ssl_certificate /etc/ssl/certs/your-cert.pem;
    ssl_certificate_key /etc/ssl/private/your-key.pem;
    
    # 请求体大小限制（大上下文窗口需要）
    client_max_body_size 50m;
    
    # 速率限制
    limit_req_zone $binary_remote_addr zone=llm_rate:10m rate=60r/m;
    
    location /v1/ {
        limit_req zone=llm_rate burst=20 nodelay;
        
        proxy_pass http://litellm_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        
        # 超时设置（LLM响应可能较慢）
        proxy_read_timeout 120s;
        proxy_connect_timeout 10s;
        proxy_send_timeout 30s;
        
        # 流式响应支持
        proxy_buffering off;
        proxy_cache off;
    }
}

11.2 Kong Gateway插件配置

# Kong服务配置
_format_version: "3.0"
services:
  - name: llm-gateway
    url: http://litellm:4000
    routes:
      - name: llm-route
        paths:
          - /v1
    plugins:
      - name: rate-limiting
        config:
          minute: 60
          policy: redis
          redis_host: redis
      - name: key-auth
        config:
          key_names:
            - api_key
      - name: http-log
        config:
          http_endpoint: http://log-collector:8080/logs

12. 完整实战：构建生产级LLM网关

12.1 项目结构

llm-gateway/
├── config/
│   ├── litellm_config.yaml
│   └── nginx.conf
├── src/
│   ├── __init__.py
│   ├── router.py          # 路由逻辑
│   ├── auth.py            # 认证鉴权
│   ├── rate_limiter.py    # 速率限制
│   ├── cache.py           # 缓存层
│   ├── logger.py          # 日志系统
│   ├── cost_monitor.py    # 成本监控
│   └── main.py            # FastAPI应用
├── tests/
├── docker-compose.yaml
├── Dockerfile
└── requirements.txt

12.2 Docker Compose部署

version: "3.8"

services:
  litellm:
    image: ghcr.io/berriai/litellm:main-latest
    ports:
      - "4000:4000"
    volumes:
      - ./config/litellm_config.yaml:/app/config.yaml
    command: ["--config", "/app/config.yaml", "--port", "4000"]
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
    depends_on:
      - redis
      - postgres

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data

  postgres:
    image: postgres:16-alpine
    ports:
      - "5432:5432"
    environment:
      - POSTGRES_DB=litellm
      - POSTGRES_USER=litellm
      - POSTGRES_PASSWORD=${DB_PASSWORD}
    volumes:
      - postgres_data:/var/lib/postgresql/data

  nginx:
    image: nginx:alpine
    ports:
      - "443:443"
    volumes:
      - ./config/nginx.conf:/etc/nginx/conf.d/default.conf
      - ./certs:/etc/ssl/certs
    depends_on:
      - litellm

volumes:
  redis_data:
  postgres_data:

13. 最佳实践总结

1. 始终使用虚拟Key：不要将真实API Key暴露给客户端
2. 实施多层缓存：精确缓存 + 语义缓存可显著降低成本
3. 设置合理的超时：LLM响应时间长，需要适当放宽超时
4. 监控成本：实时追踪Token消耗，设置预算告警
5. 故障转移必备：至少配置2个备用模型
6. 日志要全：记录每次请求的完整上下文，便于排查问题
7. 流式支持：生产环境务必支持SSE流式响应
8. 定期轮转Key：定期更换API Key，降低泄露风险
9. 隔离租户：多租户场景下严格隔离资源和配额
10. 压测先行：上线前进行充分的压力测试

总结

AI API网关是企业级LLM应用的核心基础设施。通过本教程，你已经掌握了从开源工具选型（LiteLLM、OpenRouter、Portkey）到自建网关的完整技术栈，包括路由策略、负载均衡、安全管控、成本监控和缓存优化等关键能力。

选择合适的方案取决于你的具体需求：

• 快速验证：直接使用OpenRouter
• 灵活自建：部署LiteLLM
• 企业级需求：Portkey + 自定义中间件

无论选择哪种方案，核心原则不变：统一路由、智能降级、全面监控、安全第一。