为什么需要AI网关？

当企业使用多个LLM提供商（OpenAI、Anthropic、本地模型等）时，直接对接各家API会面临：密钥管理分散、无法统一限流、缺乏请求日志、故障切换困难。AI网关统一管理所有LLM请求，提供路由、缓存、限流、监控等基础设施。

核心架构

客户端 → AI网关 → LLM提供商A
                 → LLM提供商B
                 → 本地vLLM

实现方案

统一API接口

from fastapi import FastAPI, Request
from pydantic import BaseModel

app = FastAPI()

class ChatRequest(BaseModel):
    model: str
    messages: list
    temperature: float = 0.7
    max_tokens: int = 2048
    stream: bool = False

# 提供商配置
PROVIDERS = {
    "openai": {"base_url": "https://api.openai.com/v1", "api_key": "..."},
    "anthropic": {"base_url": "https://api.anthropic.com", "api_key": "..."},
    "local": {"base_url": "http://localhost:8000/v1", "api_key": "..."},
}

# 模型到提供商的路由
MODEL_ROUTING = {
    "gpt-4": "openai",
    "claude-3": "anthropic",
    "qwen3-32b": "local",
}

@app.post("/v1/chat/completions")
async def chat_completions(request: ChatRequest):
    provider = MODEL_ROUTING.get(request.model, "local")
    config = PROVIDERS[provider]
    
    # 转发请求
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{config['base_url']}/chat/completions",
            json=request.dict(),
            headers={"Authorization": f"Bearer {config['api_key']}"},
            timeout=120
        )
        return response.json()

故障切换

class FailoverRouter:
    def __init__(self, routing_config):
        self.routing = routing_config  # {model: [provider1, provider2, ...]}
        self.health = {p: True for providers in routing_config.values() for p in providers}
    
    async def route(self, model, request):
        providers = self.routing.get(model, ["local"])
        
        for provider in providers:
            if not self.health[provider]:
                continue
            
            try:
                result = await self.call_provider(provider, request)
                return result
            except Exception as e:
                logger.warning(f"Provider {provider} failed: {e}")
                self.health[provider] = False
                continue
        
        raise ServiceUnavailableError("All providers failed")

请求缓存

import hashlib
import redis.asyncio as redis

class ResponseCache:
    def __init__(self, redis_url="redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
    
    def cache_key(self, model, messages, temperature):
        content = json.dumps({"model": model, "messages": messages, "temp": temperature})
        return hashlib.sha256(content.encode()).hexdigest()
    
    async def get(self, model, messages, temperature):
        key = self.cache_key(model, messages, temperature)
        cached = await self.redis.get(key)
        return json.loads(cached) if cached else None
    
    async def set(self, model, messages, temperature, response, ttl=3600):
        key = self.cache_key(model, messages, temperature)
        await self.redis.setex(key, ttl, json.dumps(response))

限流

from datetime import datetime, timedelta

class RateLimiter:
    def __init__(self, redis):
        self.redis = redis
    
    async def check(self, user_id, limit=60, window=60):
        key = f"rate:{user_id}:{datetime.now().strftime('%Y%m%d%H%M')}"
        current = await self.redis.incr(key)
        if current == 1:
            await self.redis.expire(key, window)
        
        if current > limit:
            return False
        return True

日志与监控

class RequestLogger:
    def __init__(self):
        self.logger = structlog.get_logger()
    
    async def log(self, request, response, user_id, duration):
        self.logger.info("llm_request",
            user_id=user_id,
            model=request.model,
            input_tokens=response.get("usage", {}).get("prompt_tokens", 0),
            output_tokens=response.get("usage", {}).get("completion_tokens", 0),
            duration_ms=duration * 1000,
            provider=response.get("provider", "unknown"),
            status="success" if response.get("choices") else "error"
        )

部署配置

Docker Compose

version: '3.8'
services:
  gateway:
    build: .
    ports:
      - "8080:8080"
    environment:
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis
  
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
  
  prometheus:
    image: prom/prometheus
    ports:
      - "9090:9090"
  
  grafana:
    image: grafana/grafana
    ports:
      - "3000:3000"

结语

AI网关是LLM生产基础设施的核心组件。统一API、故障切换、缓存、限流和监控这五大功能，让LLM服务具备企业级的可靠性和可观测性。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。