大模型网关与API管理完全教程

面向AI工程师与架构师，系统讲解LLM网关的设计原理、核心功能与企业级部署实践。

1. 为什么需要LLM网关

当企业从"试用一个大模型"演进到"在生产环境中使用多个大模型"时，一系列工程挑战会迅速浮现：

挑战	具体表现
多供应商管理	OpenAI、Anthropic、Google、国产模型各有不同的API格式和SDK
成本不可控	不同模型定价差异巨大，缺乏统一视图
安全风险	API Key分散在各个服务中，泄露面大
可用性保障	单一供应商故障导致整个AI功能不可用
合规审计	无法追踪谁在什么时候用了什么模型处理了什么数据

LLM网关（LLM Gateway）正是为了解决这些问题而出现的基础设施层。它位于你的应用与各LLM供应商之间，提供统一的接入点和管理能力。

类比理解：如果LLM供应商是各个银行，那么LLM网关就是你的"统一支付网关"——你只需要对接一个接口，背后可以灵活切换和组合多个银行通道。

2. LLM网关架构设计

2.1 整体架构

一个完整的LLM网关通常包含以下核心组件：

┌─────────────────────────────────────────────────────┐
│                    客户端应用层                        │
│   Web App / Mobile / 后端服务 / 内部工具              │
└──────────────────────┬──────────────────────────────┘
                       │ 统一API（OpenAI兼容格式）
                       ▼
┌─────────────────────────────────────────────────────┐
│                  LLM 网关核心层                       │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌────────┐ │
│  │ 认证鉴权  │ │ 路由引擎  │ │ 速率限制  │ │ 日志   │ │
│  └──────────┘ └──────────┘ └──────────┘ └────────┘ │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌────────┐ │
│  │ 缓存层   │ │ 成本追踪  │ │ 负载均衡  │ │ 缓存   │ │
│  └──────────┘ └──────────┘ └──────────┘ └────────┘ │
└──────────────────────┬──────────────────────────────┘
                       │
        ┌──────────────┼──────────────┐
        ▼              ▼              ▼
┌──────────────┐┌──────────────┐┌──────────────┐
│   OpenAI     ││  Anthropic   ││  国产模型     │
│   API        ││  Claude API  ││  (文心/通义)  │
└──────────────┘└──────────────┘└──────────────┘

2.2 核心设计原则

统一接口：对所有下游模型暴露OpenAI兼容的API格式，降低客户端迁移成本
可观测性：每一次请求都可追踪、可审计、可回放
弹性设计：任何一个供应商故障不应影响整体服务可用性
安全纵深：认证、授权、加密、脱敏多层防护

2.3 请求生命周期

客户端发起请求
    │
    ▼
① 认证（验证API Key / JWT Token）
    │
    ▼
② 鉴权（检查权限：可用模型、配额）
    │
    ▼
③ 速率限制检查（令牌桶 / 滑动窗口）
    │
    ▼
④ 缓存查询（语义缓存 / 精确匹配）
    │  → 命中：直接返回缓存结果
    │
    ▼
⑤ 路由决策（根据策略选择模型和供应商）
    │
    ▼
⑥ 请求转发（格式适配 + 发送到目标模型）
    │
    ▼
⑦ 响应处理（格式标准化 + 敏感内容过滤）
    │
    ▼
⑧ 日志记录 + 成本计算
    │
    ▼
⑨ 返回响应给客户端

3. 多模型路由策略

路由策略是LLM网关最核心的能力——它决定了每个请求应该被发送到哪个模型。

3.1 基于规则的路由

最简单直接的方式，通过预定义规则将请求分配到特定模型：

# 基于请求内容的路由规则示例
ROUTING_RULES = [
    {
        "name": "代码生成任务",
        "condition": lambda req: req.get("metadata", {}).get("task_type") == "code_generation",
        "target_model": "claude-sonnet-4-20250514",
        "fallback": "gpt-4o"
    },
    {
        "name": "简单问答",
        "condition": lambda req: len(req.get("messages", [{}])[-1].get("content", "")) < 200,
        "target_model": "gpt-4o-mini",
        "fallback": "claude-sonnet-4-20250514"
    },
    {
        "name": "长文档处理",
        "condition": lambda req: req.get("max_tokens", 0) > 4000,
        "target_model": "gpt-4o",
        "fallback": "claude-sonnet-4-20250514"
    },
    {
        "name": "默认路由",
        "condition": lambda req: True,
        "target_model": "gpt-4o-mini",
        "fallback": "claude-sonnet-4-20250514"
    }
]

def route_request(request: dict) -> str:
    for rule in ROUTING_RULES:
        if rule["condition"](request):
            return rule["target_model"]
    return "gpt-4o-mini"  # 最终兜底

3.2 基于成本优化的路由

当任务对模型能力要求不高时，优先使用低成本模型：

# 成本优先路由：按价格从低到高尝试，直到质量达标
MODEL_TIERS = [
    {"model": "gpt-4o-mini",       "cost_per_1k": 0.00015, "quality_score": 0.7},
    {"model": "claude-sonnet-4-20250514", "cost_per_1k": 0.003,  "quality_score": 0.88},
    {"model": "gpt-4o",            "cost_per_1k": 0.005,  "quality_score": 0.92},
    {"model": "o1",                "cost_per_1k": 0.015,  "quality_score": 0.96},
]

def cost_optimized_route(request: dict, min_quality: float = 0.8) -> str:
    """选择满足最低质量要求的最便宜模型"""
    eligible = [m for m in MODEL_TIERS if m["quality_score"] >= min_quality]
    if not eligible:
        return MODEL_TIERS[-1]["model"]  # 使用最高质量模型
    return min(eligible, key=lambda m: m["cost_per_1k"])["model"]

3.3 基于A/B测试的路由

在引入新模型时，通过流量分割进行效果对比：

import hashlib

def ab_test_route(user_id: str, request: dict) -> dict:
    """基于用户ID的确定性哈希分流"""
    experiments = {
        "model_upgrade": {
            "control": {"model": "gpt-4o-mini", "weight": 90},
            "treatment": {"model": "claude-sonnet-4-20250514", "weight": 10},
        }
    }

    exp = experiments["model_upgrade"]
    hash_val = int(hashlib.md5(user_id.encode()).hexdigest()[:8], 16)
    bucket = hash_val % 100

    if bucket < exp["control"]["weight"]:
        return {"model": exp["control"]["model"], "group": "control"}
    else:
        return {"model": exp["treatment"]["model"], "group": "treatment"}

4. 负载均衡与故障转移

4.1 多供应商负载均衡

为同一个逻辑模型配置多个供应商通道，实现流量分散：

import random
import time
from dataclasses import dataclass, field

@dataclass
class ModelEndpoint:
    name: str
    provider: str
    model_id: str
    weight: int = 100
    is_healthy: bool = True
    current_connections: int = 0
    avg_latency_ms: float = 0
    error_count: int = 0
    last_error_time: float = 0

class LoadBalancer:
    def __init__(self, endpoints: list[ModelEndpoint]):
        self.endpoints = endpoints
        self.error_threshold = 5
        self.recovery_time = 60  # 秒

    def get_healthy_endpoints(self) -> list[ModelEndpoint]:
        """过滤健康端点，自动恢复不健康的端点"""
        now = time.time()
        for ep in self.endpoints:
            if not ep.is_healthy:
                if now - ep.last_error_time > self.recovery_time:
                    ep.is_healthy = True
                    ep.error_count = 0
        return [ep for ep in self.endpoints if ep.is_healthy]

    def select_endpoint(self, strategy: str = "weighted") -> ModelEndpoint:
        healthy = self.get_healthy_endpoints()
        if not healthy:
            raise Exception("所有端点均不可用")

        if strategy == "weighted":
            total_weight = sum(ep.weight for ep in healthy)
            r = random.uniform(0, total_weight)
            cumulative = 0
            for ep in healthy:
                cumulative += ep.weight
                if r <= cumulative:
                    return ep
        elif strategy == "least_latency":
            return min(healthy, key=lambda ep: ep.avg_latency_ms)
        elif strategy == "least_connections":
            return min(healthy, key=lambda ep: ep.current_connections)

        return healthy[0]

    def mark_error(self, endpoint: ModelEndpoint):
        endpoint.error_count += 1
        endpoint.last_error_time = time.time()
        if endpoint.error_count >= self.error_threshold:
            endpoint.is_healthy = False

# 使用示例
endpoints = [
    ModelEndpoint(name="openai-primary", provider="openai", model_id="gpt-4o", weight=60),
    ModelEndpoint(name="anthropic-primary", provider="anthropic", model_id="claude-sonnet-4-20250514", weight=40),
]
lb = LoadBalancer(endpoints)
selected = lb.select_endpoint(strategy="weighted")

4.2 故障转移链

设计多级降级策略，确保服务持续可用：

import asyncio

async def call_with_fallback(request: dict, fallback_chain: list[dict]) -> dict:
    """
    按优先级依次尝试，直到成功
    fallback_chain 示例:
    [
        {"provider": "openai", "model": "gpt-4o", "timeout": 30},
        {"provider": "anthropic", "model": "claude-sonnet-4-20250514", "timeout": 30},
        {"provider": "azure", "model": "gpt-4o", "timeout": 60},  # 备用区域
    ]
    """
    errors = []

    for target in fallback_chain:
        try:
            response = await asyncio.wait_for(
                call_model(target["provider"], target["model"], request),
                timeout=target.get("timeout", 30)
            )
            return {
                "response": response,
                "used_provider": target["provider"],
                "used_model": target["model"],
                "fallback_level": len(errors)
            }
        except asyncio.TimeoutError:
            errors.append({"provider": target["provider"], "error": "timeout"})
        except Exception as e:
            errors.append({"provider": target["provider"], "error": str(e)})

    raise Exception(f"所有供应商均失败: {errors}")

5. API Key管理与安全

5.1 分层密钥体系

企业级LLM网关应实现三层密钥体系：

层级	说明	示例
网关密钥	网关自身的管理密钥	`gw_admin_xxxx`
业务密钥	分配给各业务团队的密钥	`team_crm_xxxx`
供应商密钥	网关持有，客户端不可见	`sk-openai-xxxx`

import os
import hashlib
import hmac
from datetime import datetime, timedelta

class APIKeyManager:
    def __init__(self):
        self.keys_db = {}  # 生产环境应使用数据库

    def generate_key(self, team: str, permissions: dict, expires_in_days: int = 90) -> str:
        """生成业务API Key"""
        import secrets
        raw_key = secrets.token_urlsafe(32)
        key_prefix = f"llm_{team[:4]}_"
        full_key = f"{key_prefix}{raw_key}"

        self.keys_db[full_key] = {
            "team": team,
            "permissions": permissions,  # {"models": ["gpt-4o*"], "max_rpm": 100}
            "created_at": datetime.utcnow().isoformat(),
            "expires_at": (datetime.utcnow() + timedelta(days=expires_in_days)).isoformat(),
            "is_active": True,
            "usage_count": 0
        }
        return full_key

    def validate_key(self, key: str) -> dict | None:
        """验证并返回密钥信息"""
        key_info = self.keys_db.get(key)
        if not key_info or not key_info["is_active"]:
            return None
        if datetime.fromisoformat(key_info["expires_at"]) < datetime.utcnow():
            return None
        key_info["usage_count"] += 1
        return key_info

    def revoke_key(self, key: str) -> bool:
        """吊销密钥"""
        if key in self.keys_db:
            self.keys_db[key]["is_active"] = False
            return True
        return False

# 使用示例
manager = APIKeyManager()
key = manager.generate_key(
    team="crm",
    permissions={"models": ["gpt-4o-mini", "gpt-4o"], "max_rpm": 50},
    expires_in_days=30
)

5.2 密钥轮换策略

class KeyRotationPolicy:
    def __init__(self, manager: APIKeyManager):
        self.manager = manager
        self.rotation_interval_days = 30
        self.grace_period_hours = 24

    def rotate_key(self, old_key: str) -> str:
        """轮换密钥：生成新密钥，旧密钥进入宽限期"""
        old_info = self.manager.keys_db.get(old_key)
        if not old_info:
            raise ValueError("旧密钥不存在")

        # 生成新密钥
        new_key = self.manager.generate_key(
            team=old_info["team"],
            permissions=old_info["permissions"],
            expires_in_days=self.rotation_interval_days
        )

        # 旧密钥不立即失效，保留宽限期
        self.manager.keys_db[old_key]["grace_until"] = (
            datetime.utcnow() + timedelta(hours=self.grace_period_hours)
        ).isoformat()

        return new_key

6. 速率限制与配额管理

6.1 令牌桶算法实现

import time
import threading

class TokenBucket:
    """令牌桶限流器：支持突发流量，平滑限速"""

    def __init__(self, rate: float, capacity: int):
        self.rate = rate          # 每秒产生的令牌数
        self.capacity = capacity  # 桶容量
        self.tokens = capacity
        self.last_refill = time.monotonic()
        self.lock = threading.Lock()

    def consume(self, tokens: int = 1) -> bool:
        with self.lock:
            now = time.monotonic()
            elapsed = now - self.last_refill
            self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
            self.last_refill = now

            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

class MultiLevelRateLimiter:
    """多级限流：用户级 + 团队级 + 全局级"""

    def __init__(self):
        self.user_buckets: dict[str, TokenBucket] = {}
        self.team_buckets: dict[str, TokenBucket] = {}
        self.global_bucket = TokenBucket(rate=1000, capacity=5000)

    def check(self, user_id: str, team: str, tokens: int = 1) -> dict:
        # 用户级限流
        if user_id not in self.user_buckets:
            self.user_buckets[user_id] = TokenBucket(rate=10, capacity=50)

        # 团队级限流
        if team not in self.team_buckets:
            self.team_buckets[team] = TokenBucket(rate=100, capacity=500)

        results = {
            "user_limit": self.user_buckets[user_id].consume(tokens),
            "team_limit": self.team_buckets[team].consume(tokens),
            "global_limit": self.global_bucket.consume(tokens),
        }
        results["allowed"] = all(results.values())
        return results

6.2 配额管理

import redis
import json

class QuotaManager:
    """基于Redis的配额管理，支持日/月配额"""

    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    def check_quota(self, team: str, quota_type: str = "daily") -> dict:
        key = f"quota:{team}:{quota_type}:{self._get_period_key(quota_type)}"
        current = int(self.redis.get(key) or 0)

        limits = {
            "daily": {"requests": 10000, "tokens": 5_000_000, "cost_usd": 100},
            "monthly": {"requests": 300000, "tokens": 150_000_000, "cost_usd": 3000},
        }
        limit = limits.get(quota_type, limits["daily"])

        return {
            "current_requests": current,
            "limit_requests": limit["requests"],
            "remaining": limit["requests"] - current,
            "usage_percent": round(current / limit["requests"] * 100, 2),
            "is_exceeded": current >= limit["requests"]
        }

    def increment(self, team: str, request_count: int = 1, token_count: int = 0):
        pipe = self.redis.pipeline()
        daily_key = f"quota:{team}:daily:{self._get_period_key('daily')}"
        monthly_key = f"quota:{team}:monthly:{self._get_period_key('monthly')}"

        pipe.incrby(daily_key, request_count)
        pipe.expire(daily_key, 86400)
        pipe.incrby(monthly_key, request_count)
        pipe.expire(monthly_key, 2592000)
        pipe.execute()

    def _get_period_key(self, quota_type: str) -> str:
        now = time.strftime("%Y%m%d")
        if quota_type == "monthly":
            return now[:6]
        return now

7. 请求与响应日志

7.1 结构化日志设计

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

@dataclass
class LLMRequestLog:
    request_id: str
    timestamp: str
    team: str
    user_id: str
    model_requested: str
    model_used: str
    provider: str
    input_tokens: int
    output_tokens: int
    total_tokens: int
    latency_ms: float
    status: str              # success / error / timeout / rate_limited
    http_status: int
    error_message: str | None
    cost_usd: float
    cache_hit: bool
    fallback_level: int
    prompt_preview: str       # 前200字符，用于调试
    response_preview: str     # 前200字符
    metadata: dict

class RequestLogger:
    def __init__(self, storage_backend="file", log_path="/var/log/llm-gateway/"):
        self.storage = storage_backend
        self.log_path = log_path

    def log_request(self, request_context: dict, response_data: dict, timing: dict):
        entry = LLMRequestLog(
            request_id=str(uuid.uuid4()),
            timestamp=datetime.utcnow().isoformat() + "Z",
            team=request_context.get("team", "unknown"),
            user_id=request_context.get("user_id", "unknown"),
            model_requested=request_context.get("model_requested", ""),
            model_used=response_data.get("model", ""),
            provider=response_data.get("provider", ""),
            input_tokens=response_data.get("usage", {}).get("prompt_tokens", 0),
            output_tokens=response_data.get("usage", {}).get("completion_tokens", 0),
            total_tokens=response_data.get("usage", {}).get("total_tokens", 0),
            latency_ms=timing.get("total_ms", 0),
            status=response_data.get("status", "unknown"),
            http_status=response_data.get("http_status", 0),
            error_message=response_data.get("error"),
            cost_usd=self._calculate_cost(response_data),
            cache_hit=response_data.get("cache_hit", False),
            fallback_level=response_data.get("fallback_level", 0),
            prompt_preview=str(request_context.get("messages", ""))[:200],
            response_preview=str(response_data.get("content", ""))[:200],
            metadata=request_context.get("metadata", {})
        )
        self._write_log(entry)
        return entry

    def _calculate_cost(self, response_data: dict) -> float:
        # 基于token数和模型定价计算成本
        pricing = {
            "gpt-4o": {"input": 0.005, "output": 0.015},
            "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
            "claude-sonnet-4-20250514": {"input": 0.003, "output": 0.015},
        }
        model = response_data.get("model", "")
        prices = pricing.get(model, {"input": 0.01, "output": 0.03})
        usage = response_data.get("usage", {})
        input_cost = usage.get("prompt_tokens", 0) / 1000 * prices["input"]
        output_cost = usage.get("completion_tokens", 0) / 1000 * prices["output"]
        return round(input_cost + output_cost, 6)

    def _write_log(self, entry: LLMRequestLog):
        log_line = json.dumps(asdict(entry), ensure_ascii=False)
        # 生产环境写入文件/数据库/消息队列
        print(log_line)  # 简化示例

8. 成本追踪与预算控制

8.1 实时成本计算

class CostTracker:
    """实时成本追踪与预算告警"""

    # 每1000个token的价格（美元）
    PRICING = {
        "gpt-4o":              {"input": 0.0025, "output": 0.01},
        "gpt-4o-mini":         {"input": 0.00015, "output": 0.0006},
        "claude-sonnet-4-20250514": {"input": 0.003, "output": 0.015},
        "claude-haiku":        {"input": 0.00025, "output": 0.00125},
    }

    def __init__(self, budget_config: dict):
        self.budgets = budget_config  # {"team_crm": {"daily_usd": 50, "monthly_usd": 1000}}
        self.usage_log = []

    def record_usage(self, team: str, model: str, input_tokens: int, output_tokens: int):
        cost = self._calc_cost(model, input_tokens, output_tokens)
        self.usage_log.append({
            "team": team, "model": model,
            "input_tokens": input_tokens, "output_tokens": output_tokens,
            "cost_usd": cost, "timestamp": time.time()
        })

        # 检查预算
        alert = self._check_budget(team, cost)
        if alert:
            self._send_alert(alert)

        return cost

    def get_team_report(self, team: str, period: str = "daily") -> dict:
        now = time.time()
        window = 86400 if period == "daily" else 2592000
        entries = [
            e for e in self.usage_log
            if e["team"] == team and now - e["timestamp"] < window
        ]

        total_cost = sum(e["cost_usd"] for e in entries)
        by_model = {}
        for e in entries:
            by_model.setdefault(e["model"], {"cost": 0, "requests": 0})
            by_model[e["model"]]["cost"] += e["cost_usd"]
            by_model[e["model"]]["requests"] += 1

        budget = self.budgets.get(team, {}).get(f"{period}_usd", float("inf"))
        return {
            "team": team, "period": period,
            "total_cost_usd": round(total_cost, 4),
            "budget_usd": budget,
            "budget_used_percent": round(total_cost / budget * 100, 2) if budget != float("inf") else 0,
            "request_count": len(entries),
            "by_model": by_model
        }

    def _calc_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        prices = self.PRICING.get(model, {"input": 0.01, "output": 0.03})
        return (input_tokens / 1000 * prices["input"]) + (output_tokens / 1000 * prices["output"])

    def _check_budget(self, team: str, new_cost: float) -> dict | None:
        report = self.get_team_report(team, "daily")
        if report["budget_used_percent"] >= 90:
            return {"level": "critical", "team": team, "usage": report["budget_used_percent"]}
        if report["budget_used_percent"] >= 70:
            return {"level": "warning", "team": team, "usage": report["budget_used_percent"]}
        return None

    def _send_alert(self, alert: dict):
        print(f"[BUDGET ALERT] {alert['level'].upper()}: {alert['team']} usage at {alert['usage']}%")

9. 缓存策略

9.1 精确匹配缓存

import hashlib
import json
import redis

class LLMCache:
    """LLM响应缓存：支持精确匹配和语义缓存"""

    def __init__(self, redis_client: redis.Redis, ttl: int = 3600):
        self.redis = redis_client
        self.ttl = ttl

    def _build_cache_key(self, messages: list, model: str, **params) -> str:
        """构建缓存键：消息内容 + 模型 + 关键参数"""
        cache_input = {
            "messages": messages,
            "model": model,
            "temperature": params.get("temperature", 0.7),
            "max_tokens": params.get("max_tokens"),
        }
        content_hash = hashlib.sha256(
            json.dumps(cache_input, sort_keys=True).encode()
        ).hexdigest()
        return f"llm_cache:{content_hash}"

    def get(self, messages: list, model: str, **params) -> dict | None:
        key = self._build_cache_key(messages, model, **params)
        cached = self.redis.get(key)
        if cached:
            return json.loads(cached)
        return None

    def set(self, messages: list, model: str, response: dict, **params):
        key = self._build_cache_key(messages, model, **params)
        self.redis.setex(key, self.ttl, json.dumps(response, ensure_ascii=False))

    def get_stats(self) -> dict:
        keys = list(self.redis.scan_iter("llm_cache:*"))
        return {"cached_entries": len(keys), "ttl_seconds": self.ttl}

9.2 语义缓存

对于语义相近但表述不同的查询，精确匹配无法命中。语义缓存通过向量相似度解决这个问题：

import numpy as np

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

    def __init__(self, embedding_model, similarity_threshold: float = 0.95):
        self.embedding_model = embedding_model
        self.threshold = similarity_threshold
        self.cache_entries = []  # [(embedding, response, metadata)]

    def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

    def query(self, text: str) -> dict | None:
        query_embedding = self.embedding_model.encode(text)

        best_match = None
        best_score = 0

        for entry_embedding, response, metadata in self.cache_entries:
            score = self._cosine_similarity(query_embedding, entry_embedding)
            if score > best_score:
                best_score = score
                best_match = response

        if best_score >= self.threshold:
            return {"response": best_match, "similarity": best_score, "cache_hit": True}
        return None

    def store(self, text: str, response: dict):
        embedding = self.embedding_model.encode(text)
        self.cache_entries.append((embedding, response, {"stored_at": time.time()}))

适用场景：FAQ客服系统、知识库问答。不适用于需要实时性的查询（如"今天天气如何"）。

10. LiteLLM与OpenRouter对比

维度	LiteLLM	OpenRouter	自建网关
部署方式	自托管Python库	SaaS云服务	完全自控
支持模型数	100+ 模型	300+ 模型	按需对接
API兼容性	OpenAI格式代理	OpenAI格式代理	可自定义
成本管理	基础预算功能	内置用量仪表板	完全自定义
缓存	Redis缓存	内置缓存	完全自定义
负载均衡	多密钥轮换	自动路由	灵活策略
合规审计	需自行实现	有限日志	完全可控
费用	开源免费	按用量加价	开发维护成本
适合场景	中小团队快速接入	个人开发者/原型	企业级生产环境

LiteLLM快速上手

# 安装: pip install litellm

import litellm
from litellm import completion

# 统一调用不同模型
response = completion(
    model="gpt-4o",
    messages=[{"role": "user", "content": "你好"}],
)
print(response.choices[0].message.content)

# 切换到其他模型，代码无需修改
response = completion(
    model="claude-sonnet-4-20250514",
    messages=[{"role": "user", "content": "你好"}],
)

# 使用代理模式（启动LiteLLM Proxy Server）
# litellm --model gpt-4o --port 4000
# 然后通过 http://localhost:4000/v1/chat/completions 访问

OpenRouter使用示例

import openai

client = openai.OpenAI(
    base_url="https://openrouter.ai/api/v1",
    api_key="sk-or-xxxx"  # OpenRouter API Key
)

response = client.chat.completions.create(
    model="anthropic/claude-sonnet-4-20250514",  # OpenRouter使用 provider/model 格式
    messages=[{"role": "user", "content": "你好"}]
)

选型建议

个人/小团队原型验证 → OpenRouter（零运维，快速接入多模型）
中型团队内部使用 → LiteLLM Proxy（自托管，成本可控）
大型企业生产环境 → 自建网关 + LiteLLM作为组件（完全可控，深度定制）

11. 企业级LLM网关部署实战

11.1 技术栈选择

┌─────────────────────────────────────────┐
│             API Gateway (Kong/APISIX)    │
│        认证 / 限流 / 日志 / 监控          │
└───────────────────┬─────────────────────┘
                    │
┌───────────────────▼─────────────────────┐
│           LLM 路由服务 (Go/Python)       │
│     路由决策 / 格式转换 / 缓存 / 成本      │
└───────────────────┬─────────────────────┘
                    │
         ┌──────────┼──────────┐
         ▼          ▼          ▼
    ┌────────┐ ┌────────┐ ┌────────┐
    │OpenAI  │ │Anthropic│ │ 国产   │
    │Endpoint│ │Endpoint │ │ 模型   │
    └────────┘ └────────┘ └────────┘

基础设施: Redis (缓存/限流) + PostgreSQL (日志/配置) + Prometheus + Grafana

11.2 Docker Compose部署示例

# docker-compose.yml
version: "3.8"

services:
  llm-gateway:
    build: ./gateway
    ports:
      - "8080:8080"
    environment:
      - REDIS_URL=redis://redis:6379
      - DATABASE_URL=postgresql://postgres:secret@postgres:5432/llm_gateway
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
    depends_on:
      - redis
      - postgres
    deploy:
      replicas: 3
      resources:
        limits:
          memory: 512M

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data
    command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru

  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: llm_gateway
      POSTGRES_PASSWORD: secret
    volumes:
      - pg_data:/var/lib/postgresql/data

  prometheus:
    image: prom/prometheus
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - "9090:9090"

  grafana:
    image: grafana/grafana
    ports:
      - "3000:3000"

volumes:
  redis_data:
  pg_data:

11.3 监控指标

部署完成后，应重点监控以下指标：

指标	告警阈值	说明
请求延迟 P99	> 10s	可能存在供应商故障
错误率	> 5%	需要排查原因
缓存命中率	< 30%	缓存策略可能需要调整
每分钟成本	> 预算的1.5倍	成本异常
各供应商可用性	< 99%	需要触发故障转移

12. 最佳实践总结

设计阶段

统一接口格式：选择OpenAI兼容格式作为标准，降低迁移成本
分层架构：认证、路由、缓存、日志各层解耦，独立扩展
幂等设计：相同请求在缓存有效期内返回一致结果

安全阶段

最小权限原则：每个团队/应用只分配必要的模型访问权限
密钥定期轮换：不超过90天，设置自动提醒
敏感数据脱敏：日志中不记录完整Prompt和Response内容

运维阶段

多供应商冗余：至少配置2个供应商，关键业务配置3个
渐进式发布：新模型先接5%流量，观察质量后再扩大
成本预算告警：设置70%警告、90%告警、100%自动熔断

优化阶段

智能缓存：结合精确缓存和语义缓存，平衡命中率和实时性
模型分级：简单任务用轻量模型，复杂任务用高端模型
持续评估：定期评估各模型的性价比，及时调整路由策略

总结：LLM网关不是一个简单的API代理，它是企业AI基础设施的核心组件。从统一接入、智能路由到成本管控、安全合规，一个设计良好的LLM网关可以让团队在享受多模型能力的同时，保持对成本、安全和质量的全面掌控。选择自建还是使用开源方案，取决于团队规模、定制需求和安全合规要求。无论哪种路径，本文介绍的核心模块和最佳实践都值得参考。

大模型网关与API管理完全教程

教程简介

大模型网关与API管理完全教程

目录

1. 为什么需要LLM网关

2. LLM网关架构设计

2.1 整体架构

2.2 核心设计原则

2.3 请求生命周期

3. 多模型路由策略

3.1 基于规则的路由

3.2 基于成本优化的路由

3.3 基于A/B测试的路由

4. 负载均衡与故障转移

4.1 多供应商负载均衡

4.2 故障转移链

5. API Key管理与安全

5.1 分层密钥体系

5.2 密钥轮换策略

6. 速率限制与配额管理

6.1 令牌桶算法实现

6.2 配额管理

7. 请求与响应日志

7.1 结构化日志设计

8. 成本追踪与预算控制

8.1 实时成本计算

9. 缓存策略

9.1 精确匹配缓存

9.2 语义缓存

10. LiteLLM与OpenRouter对比

LiteLLM快速上手

OpenRouter使用示例

选型建议

11. 企业级LLM网关部署实战

11.1 技术栈选择

11.2 Docker Compose部署示例

11.3 监控指标

12. 最佳实践总结

设计阶段

安全阶段

运维阶段

优化阶段

相关教程

Python入门完全教程

Java基础编程教程

Go语言实战教程

Python编程入门完全教程

目录

快速导航