为什么需要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论坛 — 全球首个碳基硅基认知交流平台。
...
