大模型接入的认证与计费多模型供应商的统一网关设计一、三个团队各用各的 API Key月底账单混乱财务说分不清谁花了多少钱多模型供应商共存的企业场景下OpenAI Claude 自建 vLLM接入层需要解决两个关键问题。第一个是认证授权——不同供应商的认证方式不同OpenAI 用 API Key、Claude 用 API Key Organization ID、vLLM 可能用 JWT 或无认证。第二个是计费分摊——谁调用了多少 Token、花了多少钱需要精确到团队/项目/用户的维度。统一网关API Gateway 模型封装了上游多供应商的差异对下游暴露一个统一接口。下游调用方只需要关心用哪个模型不需要知道这个模型走 OpenAI 还是 vLLM。认证、鉴权、限流、计费全部在网关层统一处理和业务代码解耦。二、底层机制与原理剖析统一网关的四个核心模块认证模块发放统一的 API Key。每个团队/项目有自己的 API Key在数据库中记录 Key → Team 的映射关系。接收请求时解析 API Key挂载team_id到请求上下文中。后端根据team_id做限流和计费。模型路由根据model参数决定转发到哪个上游。路由表在配置中心维护——新增供应商或模型只需要改路由表不需要改代码。路由逻辑可以按团队做特殊化——Team A 的gpt-4走 OpenAITeam B 的同模型走 Azure。计费模块不管上游供应商是 OpenAI按 token 收费还是自建 vLLM按 GPU 时间计费在网关层统一折算为内部计费单位。从 API 响应中提取usage.total_tokens记录到数据库。支持按团队/模型/时间维度的聚合查询。格式适配器不同供应商的 API 格式不完全相同。OpenAI 用messages数组Anthropic 用messagessystem字段。网关层负责做格式转换——下游只需要用统一格式如 OpenAI 格式上游差异由适配器处理。三、生产级代码实现# unified-llm-gateway.py 统一 LLM API 网关 功能 1. 统一认证API Key → 团队/用户 2. 模型路由model → provider 3. 请求格式适配统一格式 ↔ 各供应商格式 4. 计费记录token 用量统计 import hashlib import hmac import time import json import logging from typing import Optional, Dict, Any, List from dataclasses import dataclass from enum import Enum from urllib.request import Request, urlopen from urllib.error import HTTPError import redis logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) # --------------------------------------------------------------------------- # 数据模型 # --------------------------------------------------------------------------- dataclass class ApiKeyInfo: API Key 对应的元信息 team_id: str user_id: str tier: str # free / pro / enterprise rate_limit_qpm: int # 每分钟请求数限制 daily_token_budget: int # 每日 Token 预算 dataclass class ProviderRoute: 供应商路由配置 provider: str # openai / anthropic / vllm base_url: str # API 基础 URL api_key: str # 该供应商的 API Key model_mapping: dict # 统一 model 名 → 供应商 model 名 default_headers: dict # 额外的请求头 # --------------------------------------------------------------------------- # 网关核心 # --------------------------------------------------------------------------- class LLMGateway: LLM 统一网关 请求流程 1. 认证解析 API Key → 确定 team/user 2. 限流检查 team/user 是否超出配额 3. 路由根据 model 查找上游供应商 4. 适配格式转换 5. 转发发送到上游 6. 计费记录 token 用量 def __init__(self, redis_client: redis.Redis None): self.redis redis_client # API Key 存储生产环境放数据库 self.api_keys: Dict[str, ApiKeyInfo] {} # 供应商路由表从配置中心加载 self.providers: Dict[str, ProviderRoute] { openai: ProviderRoute( provideropenai, base_urlhttps://api.openai.com/v1, api_keysk-xxx, # 实际从 secrets manager 读取 model_mapping{gpt-4: gpt-4-turbo, gpt-3.5: gpt-3.5-turbo}, default_headers{OpenAI-Organization: org-xxx}, ), anthropic: ProviderRoute( provideranthropic, base_urlhttps://api.anthropic.com/v1, api_keysk-ant-xxx, model_mapping{claude-3: claude-3-opus-20240229}, default_headers{anthropic-version: 2023-06-01}, ), vllm: ProviderRoute( providervllm, base_urlhttp://vllm-service.ai-inference.svc:8000/v1, api_key, # 内部服务可能不需要 API Key model_mapping{llama-3: meta-llama/Meta-Llama-3-70B-Instruct}, default_headers{}, ), } # 模型 → 供应商映射 self.model_routes { gpt-4: openai, gpt-3.5-turbo: openai, claude-3-opus: anthropic, claude-3-sonnet: anthropic, llama-3-70b: vllm, } def register_api_key(self, api_key: str, info: ApiKeyInfo): 注册 API Key self.api_keys[api_key] info def handle_chat_completion(self, api_key: str, model: str, messages: list, **kwargs) - Dict[str, Any]: 处理 chat completion 请求核心入口 # 1. 认证 key_info self.api_keys.get(api_key) if not key_info: return {error: Invalid API key, status: 401} # 2. 限流检查 if not self._check_rate_limit(key_info): return {error: Rate limit exceeded, status: 429} # 3. 路由查找 provider_name self.model_routes.get(model) if not provider_name: return {error: fUnknown model: {model}, status: 400} provider self.providers.get(provider_name) if not provider: return {error: fProvider not configured: {provider_name}, status: 500} # 4. 请求格式适配 adapted_request self._adapt_request(provider, model, messages, **kwargs) # 5. 转发到上游供应商 upstream_url f{provider.base_url}/chat/completions response self._forward_request( upstream_url, adapted_request, provider ) if isinstance(response, dict) and error in response: return response # 6. 计费记录 usage response.get(usage, {}) self._record_billing( team_idkey_info.team_id, user_idkey_info.user_id, modelmodel, providerprovider_name, prompt_tokensusage.get(prompt_tokens, 0), completion_tokensusage.get(completion_tokens, 0), ) # 7. 响应格式统一 return self._adapt_response(response, provider) def _adapt_request(self, provider: ProviderRoute, model: str, messages: list, **kwargs) - dict: 请求格式适配统一格式 → 供应商格式 # 映射 model 名 mapped_model provider.model_mapping.get(model, model) if provider.provider anthropic: # Anthropic 格式system 从 messages 中提取 system_msg filtered_messages [] for msg in messages: if msg.get(role) system: system_msg msg.get(content, ) else: filtered_messages.append(msg) return { model: mapped_model, messages: filtered_messages, system: system_msg, max_tokens: kwargs.get(max_tokens, 1024), temperature: kwargs.get(temperature, 0.7), } # OpenAI 兼容格式vLLM 也兼容 return { model: mapped_model, messages: messages, max_tokens: kwargs.get(max_tokens, 1024), temperature: kwargs.get(temperature, 0.7), stream: kwargs.get(stream, False), } def _adapt_response(self, response: dict, provider: ProviderRoute) - dict: 响应格式适配供应商格式 → 统一格式 # 统一包装为 OpenAI 兼容格式 return { id: response.get(id, ), object: chat.completion, created: response.get(created, int(time.time())), model: response.get(model, ), choices: response.get(choices, []), usage: response.get(usage, {}), provider: provider.provider, # 额外字段标记来源 } def _forward_request(self, url: str, body: dict, provider: ProviderRoute) - dict: 转发请求到上游供应商 headers { Content-Type: application/json, Authorization: fBearer {provider.api_key}, **provider.default_headers, } data json.dumps(body).encode(utf-8) try: req Request(url, datadata, headersheaders, methodPOST) with urlopen(req, timeout60) as resp: resp_body resp.read().decode(utf-8) return json.loads(resp_body) except HTTPError as e: error_body e.read().decode(utf-8) if e.fp else logger.error(Upstream error: %s %s, e.code, error_body) return { error: fUpstream error ({e.code}), detail: error_body, status: e.code, } def _check_rate_limit(self, key_info: ApiKeyInfo) - bool: 检查请求频率限制 if not self.redis: return True # 每分钟请求数限制 rl_key fratelimit:{key_info.team_id}:qpm current self.redis.incr(rl_key) if current 1: self.redis.expire(rl_key, 60) # 1 分钟窗口 return current key_info.rate_limit_qpm def _record_billing(self, team_id: str, user_id: str, model: str, provider: str, prompt_tokens: int, completion_tokens: int): 记录计费信息 # Token 成本映射示例——实际从配置读取 cost_per_1k { (gpt-4, openai): 0.03, (gpt-3.5-turbo, openai): 0.002, (claude-3-opus, anthropic): 0.015, (llama-3-70b, vllm): 0, # 自建模型按 GPU 时间计费 } total_tokens prompt_tokens completion_tokens unit_cost cost_per_1k.get((model, provider), 0.001) estimated_cost total_tokens * unit_cost / 1000 # 写入计费日志实际应写入数据库 logger.info( Billing: team%s user%s model%s provider%s tokens%d (prompt%d, completion%d) cost$%.4f, team_id, user_id, model, provider, total_tokens, prompt_tokens, completion_tokens, estimated_cost, ) # 在 Redis 中记录每日累计用于日报表 if self.redis: date_key time.strftime(%Y%m%d) bill_key fbilling:{date_key}:{team_id} self.redis.hincrby(bill_key, total_tokens, total_tokens) self.redis.hincrbyfloat(bill_key, total_cost, estimated_cost) self.redis.expire(bill_key, 86400 * 30) # 保留 30 天 # --------------------------------------------------------------------------- # 使用示例 # --------------------------------------------------------------------------- if __name__ __main__: gateway LLMGateway() # 注册 API Key gateway.register_api_key(sk-team-a-xxx, ApiKeyInfo( team_idteam-a, user_idservice-account, tierenterprise, rate_limit_qpm1000, daily_token_budget10_000_000, )) # 发起请求 response gateway.handle_chat_completion( api_keysk-team-a-xxx, modelgpt-4, messages[ {role: system, content: 你是一个有用的助手}, {role: user, content: 什么是 Kubernetes}, ], max_tokens500, ) print(json.dumps(response, indent2, ensure_asciiFalse))四、边界分析与架构权衡网关单点问题网关成为所有 LLM 请求的唯一入口如果网关挂了所有 LLM 调用不可用解决方案多副本部署K8s Deployment replicas3 负载均衡。网关自身应该是无状态的状态放 Redis/DB认证转发 vs 认证终结网关可以把下游 API Key 换成上游 API Key 后转发认证中转——下游不需要知道上游的 API Key也可以把下游的 API Key 直接透传给上游认证透传——如果上下游 Key 一致推荐中转模式——上游 API Key 只在网关持有下游泄漏自己的 Key 不会泄漏上游 Key流式响应的处理ChatGPT 的 SSEServer-Sent Events流式响应在网关层需要特殊处理——不能等上游完全返回再转给下游网关需要支持 SSE 代理——上游 SSE 事件逐个转发同时累计 token 用于计费五、总结LLM 统一网关的核心价值是封装上游多供应商差异认证方式、API 格式、计费模型对下游暴露统一接口。四个模块各司其职认证鉴权API Key → team/user、模型路由model → provider URL、格式适配统一格式 ↔ 各供应商格式、计费记录token 用量 → 成本分摊。关键设计决策认证走中转模式上游 Key 只在网关持有、计费层记录到数据库支持按团队/模型的成本分析、网关自身无状态状态放 Redis。