一、引言：为什么多租户AI平台需要资源隔离
在企业级AI平台的建设过程中，多租户（Multi-Tenancy）架构是实现资源复用、降低成本、提高运营效率的关键设计。随着AI技术在不同业务线的广泛渗透，企业内部往往存在多个团队或业务部门需要共享AI基础设施的场景。然而，不同团队对GPU算力、模型服务、数据存储的需求差异巨大，如何在保障安全隔离的前提下实现资源的高效利用，是每个AI平台架构师必须面对的核心挑战。
传统的单租户部署模式存在明显的局限性：每个团队独立部署一套完整的AI基础设施，不仅造成资源浪费，还增加了运维复杂度和成本。以一个典型的大型企业为例，如果每个业务线都独立部署AI能力，可能需要数十套独立的GPU集群、模型管理服务和数据存储系统，这显然是不可接受的。
多租户AI平台的核心价值在于通过统一的基础设施抽象，为多个租户提供隔离但共享的计算资源。然而，这种设计也带来了新的挑战：如何在共享资源的同时确保租户之间的严格隔离？如何在保证公平性的同时满足不同租户的差异化需求？如何实现精细化的资源配额管理？
本文将从计算资源隔离、GPU配额管理、模型服务隔离、数据隔离、配额管理等多个维度，详细介绍多租户AI平台的设计思路和实现方案。
二、多租户隔离的整体架构
2.1 隔离维度的全景视图
多租户AI平台的隔离设计需要从多个维度进行综合考虑，形成一个立体的防护体系：
┌─────────────────────────────────────────────────────────────────────────┐
│ 多租户AI平台隔离架构 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │
│ │ 租户 A (Team) │ │ 租户 B (Team) │ │ 租户 C (Team) │ │
│ └────────┬────────┘ └────────┬────────┘ └────────┬────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ 逻辑隔离层 (Logical Isolation) │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌──────────┐ │ │
│ │ │ 模型服务隔离 │ │ 数据隔离 │ │ 配额管理 │ │ 访问控制 │ │ │
│ │ └─────────────┘ └─────────────┘ └─────────────┘ └──────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ 物理隔离层 (Physical Isolation) │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ GPU资源池 │ │ 存储资源池 │ │ 网络资源池 │ │ │
│ │ │ (K8s NS) │ │ (S3/OSS) │ │ (VPC/NSG) │ │ │
│ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘从架构图中可以看出，多租户隔离主要分为以下几个维度：
计算资源隔离：这是最核心的隔离维度，涉及GPU/ CPU资源的分配和调度。每个租户应该能够独立申请和使用计算资源，而不会影响其他租户的性能。Kubernetes的命名空间（Namespace）机制为计算资源隔离提供了基础设施层面的支持。
模型服务隔离：不同租户可能使用不同的模型，或者对同一模型有不同的性能要求。服务隔离确保租户的推理请求不会相互干扰，同时支持模型的独立部署和版本管理。
数据隔离：这是安全性要求最高的维度。租户的数据必须严格隔离，防止未授权访问。同时，数据隔离还需要考虑性能问题，避免过度隔离导致的数据访问效率低下。
配额管理：配额是实现资源公平分配和成本控制的关键机制。通过Token配额、计算时长配额、存储配额等维度，可以精细化控制每个租户的资源使用量。
2.2 Kubernetes命名空间隔离方案
Kubernetes的Namespace是实现多租户隔离的基础设施层。每个租户对应一个独立的Namespace，通过RBAC（Role-Based Access Control）机制控制权限，通过ResourceQuota和LimitRange控制资源使用：
┌──────────────────────────────────────────────────────────────────────────┐
│ K8s 多租户命名空间拓扑 │
├──────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌───────────────────────┐ │
│ │ kube-system (系统) │ │
│ └───────────────────────┘ │
│ │
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │
│ │ namespace: │ │ namespace: │ │ namespace: │ │
│ │ tenant-a │ │ tenant-b │ │ tenant-c │ │
│ │ ├─ ResourceQuota│ │ ├─ ResourceQuota│ │ ├─ ResourceQuota│ │
│ │ ├─ LimitRange │ │ ├─ LimitRange │ │ ├─ LimitRange │ │
│ │ ├─ RBAC │ │ ├─ RBAC │ │ ├─ RBAC │ │
│ │ └─ NetworkPolicy│ │ └─ NetworkPolicy│ │ └─ NetworkPolicy│ │
│ └─────────────────┘ └─────────────────┘ └─────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ 共享资源层 (Shared Resources) │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ GPU节点池 │ │ 共享存储 │ │ 镜像仓库 │ │ │
│ │ │ (nvidia.com/gpu)│ │ (MinIO/S3) │ │ (Harbor) │ │ │
│ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
└──────────────────────────────────────────────────────────────────────────┘三、GPU资源配额管理
3.1 资源配额（ResourceQuota）配置
ResourceQuota是Kubernetes原生的资源配额管理机制，可以限制Namespace内所有Pod的CPU、内存、GPU等资源总量：
# k8s-resource-quota.yaml
# Kubernetes资源配额配置 - 为每个租户Namespace设置资源上限
apiVersion: v1
kind: ResourceQuota
metadata:
name: tenant-a-quota # 租户A的资源配额
namespace: tenant-a # 关联到租户A的命名空间
spec:
hard:
# requests.cpu: 限制所有Pod请求的CPU总量（单位：核）
requests.cpu: "32"
# limits.cpu: 限制所有Pod上限的CPU总量
limits.cpu: "64"
# requests.memory: 限制所有Pod请求的内存总量
requests.memory: "128Gi"
# limits.memory: 限制所有Pod上限的内存总量
limits.memory: "256Gi"
# requests.nvidia.com/gpu: GPU请求配额（需要安装nvidia-device-plugin）
requests.nvidia.com/gpu: "4"
# limits.nvidia.com/gpu: GPU上限配额
limits.nvidia.com/gpu: "4"
# pods: 限制租户可创建的Pod数量
pods: "50"
# services: 限制租户可创建的服务数量
services: "20"
# persistentvolumeclaims: 限制租户可创建的PVC数量
persistentvolumeclaims: "10"
# configmaps: 限制租户可创建的ConfigMap数量
configmaps: "30"
# secrets: 限制租户可创建的Secret数量
secrets: "30"
# requests.storage: 限制租户可请求的存储总量
requests.storage: "500Gi"
---
apiVersion: v1
kind: ResourceQuota
metadata:
name: tenant-b-quota # 租户B的资源配额（相对较小）
namespace: tenant-b
spec:
hard:
requests.cpu: "16"
limits.cpu: "32"
requests.memory: "64Gi"
limits.memory: "128Gi"
requests.nvidia.com/gpu: "2"
limits.nvidia.com/gpu: "2"
pods: "30"
services: "10"
persistentvolumeclaims: "5"
requests.storage: "200Gi"3.2 限制范围（LimitRange）配置
LimitRange用于设置Namespace内单个Pod/容器的默认资源限制和最大最小值：
# k8s-limit-range.yaml
# LimitRange配置 - 设置容器级别的资源限制
apiVersion: v1
kind: LimitRange
metadata:
name: tenant-a-limits # 租户A的LimitRange
namespace: tenant-a
spec:
limits:
# 容器级别的资源限制
- type: Container # 限制类型：容器
max:
cpu: "8" # 单个容器最大CPU（核）
memory: "32Gi" # 单个容器最大内存
nvidia.com/gpu: "2" # 单个容器最大GPU数
min:
cpu: "100m" # 单个容器最小CPU（100m = 0.1核）
memory: "128Mi" # 单个容器最小内存
default:
cpu: "1" # 未指定limits时的默认值
memory: "2Gi"
defaultRequest:
cpu: "500m" # 未指定requests时的默认值
memory: "1Gi"
# CPU和内存的比例因子
maxLimitRequestRatio:
cpu: "4" # limits.cpu / requests.cpu 最大比例
memory: "2" # limits.memory / requests.memory 最大比例
# Pod级别的资源限制
- type: Pod
max:
cpu: "16"
memory: "64Gi"
nvidia.com/gpu: "4"
min:
cpu: "200m"
memory: "256Mi"
# PVC（持久卷声明）级别的存储限制
- type: PersistentVolumeClaim
max:
storage: "100Gi" # 单个PVC最大存储
min:
storage: "1Gi" # 单个PVC最小存储3.3 GPU污点与容忍策略
在共享GPU集群中，需要通过Taint（污点）和Toleration（容忍）机制来控制Pod调度：
# k8s-gpu-taint.yaml
# GPU节点污点配置
apiVersion: v1
kind: Node
metadata:
name: gpu-node-1
labels:
node-type: gpu # 标记为GPU节点
gpu-model: "A100-80G" # GPU型号标签
spec:
taints:
# 污点键：gpu，值为A100-80G，效果为NoSchedule
# 表示除非Pod显式声明容忍此污点，否则不会调度到此节点
- key: "nvidia.com/gpu"
value: "A100-80G"
effect: "NoSchedule"
---
# GPU推理服务的Toleration配置
apiVersion: apps/v1
kind: Deployment
metadata:
name: tenant-a-inference
namespace: tenant-a
spec:
template:
spec:
tolerations:
# 容忍GPU污点，允许调度到GPU节点
- key: "nvidia.com/gpu"
operator: "Exists" # 操作符：Exists表示key存在即可
effect: "NoSchedule"
containers:
- name: inference
image: your-registry.com/inference:v1
resources:
limits:
nvidia.com/gpu: "1" # 请求1个GPU
memory: "16Gi"
requests:
nvidia.com/gpu: "1"
memory: "8Gi"四、模型服务隔离策略
4.1 独立推理服务 vs 共享推理服务
模型服务隔离是多租户AI平台的核心设计决策之一，主要有两种模式：
独立推理服务模式：每个租户拥有独立的推理服务实例，完全隔离但资源利用率较低。
共享推理服务模式：多个租户共享同一推理服务实例，通过请求路由和资源隔离实现逻辑分离，资源利用率高但复杂度增加。
┌─────────────────────────────────────────────────────────────────────────┐
│ 模型服务隔离架构对比 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ 【模式一：独立推理服务】 【模式二：共享推理服务】 │
│ │
│ ┌─────────────────────────────┐ ┌─────────────────────────────┐ │
│ │ 共享GPU集群 │ │ 共享GPU集群 │ │
│ │ ┌───────────────────────┐ │ │ ┌─────────────────────────┐ │ │
│ │ │ 租户A推理服务 │ │ │ │ 共享vLLM推理引擎 │ │ │
│ │ │ ├─ Deployment A1 │ │ │ │ ├─ 模型1 (租户A的模型) │ │ │
│ │ │ └─ Deployment A2 │ │ │ │ ├─ 模型2 (租户B的模型) │ │ │
│ │ └───────────────────────┘ │ │ │ └─ 模型3 (租户C的模型) │ │ │
│ │ ┌───────────────────────┐ │ │ └─────────────────────────┘ │ │
│ │ │ 租户B推理服务 │ │ │ │ │ │
│ │ │ └─ Deployment B1 │ │ │ ▼ │ │
│ │ └───────────────────────┘ │ │ ┌─────────────────────────┐ │ │
│ │ ┌───────────────────────┐ │ │ │ 请求路由层 (Gateway) │ │ │
│ │ │ 租户C推理服务 │ │ │ │ ├─ 租户A → 模型1 │ │ │
│ │ │ └─ Deployment C1 │ │ │ │ ├─ 租户B → 模型2 │ │ │
│ │ └───────────────────────┘ │ │ │ └─ 租户C → 模型3 │ │ │
│ └─────────────────────────────┘ │ └─────────────────────────┘ │ │
│ └─────────────────────────────┘ │
│ │
│ 优点：完全隔离、安全性高 优点：资源利用率高、成本低 │
│ 缺点：资源浪费、管理复杂 缺点：需要额外的隔离机制 │
│ │
└─────────────────────────────────────────────────────────────────────────┘4.2 模型服务的K8s部署配置
以下是独立推理服务模式的完整部署配置：
# k8s-inference-deployment.yaml
# 模型推理服务部署配置 - 每个租户独立的推理服务
apiVersion: v1
kind: Service
metadata:
name: tenant-a-inference-svc # 租户A推理服务
namespace: tenant-a
labels:
app: inference
tenant: tenant-a
spec:
type: ClusterIP
ports:
- port: 8000
targetPort: 8000
protocol: TCP
name: http
selector:
app: inference
tenant: tenant-a
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: tenant-a-inference # 租户A推理部署
namespace: tenant-a
labels:
app: inference
tenant: tenant-a
version: v1
spec:
replicas: 2 # 高可用：2个副本
selector:
matchLabels:
app: inference
tenant: tenant-a
template:
metadata:
labels:
app: inference
tenant: tenant-a
version: v1
spec:
# 亲和性配置：优先调度到GPU节点
affinity:
nodeAffinity:
requiredDuringSchedulingIgnoredDuringExecution:
nodeSelectorTerms:
- matchExpressions:
- key: "nvidia.com/gpu"
operator: Exists
preferredDuringSchedulingIgnoredDuringExecution:
- weight: 100
preference:
matchExpressions:
- key: "gpu-model"
operator: In
values: ["A100-80G"]
# Pod反亲和性：副本分布在不同节点
podAntiAffinity:
requiredDuringSchedulingIgnoredDuringExecution:
- labelSelector:
matchExpressions:
- key: "app"
operator: In
values: ["inference"]
- key: "tenant"
operator: In
values: ["tenant-a"]
topologyKey: "kubernetes.io/hostname"
# 容忍GPU污点
tolerations:
- key: "nvidia.com/gpu"
operator: Exists
effect: "NoSchedule"
containers:
- name: inference
image: your-registry.com/vllm-inference:0.4.0
imagePullPolicy: IfNotPresent
ports:
- containerPort: 8000
name: http
env:
# 模型路径
- name: MODEL_PATH
value: "/models/tenant-a-qwen2-7b"
# vLLM配置
- name: GPU_MEMORY_UTILIZATION
value: "0.9"
- name: MAX_MODEL_LEN
value: "8192"
- name: TENSOR_PARALLEL_SIZE
value: "1"
resources:
limits:
nvidia.com/gpu: "1"
memory: "16Gi"
cpu: "4"
requests:
nvidia.com/gpu: "1"
memory: "8Gi"
cpu: "2"
# 健康检查
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 30
periodSeconds: 10
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 60
periodSeconds: 15
volumeMounts:
- name: model-storage
mountPath: /models
# 资源策略：更新时保持可用
resources:
limits:
nvidia.com/gpu: "1"
volumes:
- name: model-storage
persistentVolumeClaim:
claimName: tenant-a-models-pvc
# 优雅终止：等待正在处理的请求完成
terminationGracePeriodSeconds: 60五、数据隔离方案
5.1 对象存储路径策略
数据隔离是保障多租户安全的关键环节。推荐采用路径级别的数据隔离策略：
┌─────────────────────────────────────────────────────────────────────────┐
│ 对象存储多租户路径结构 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ s3://ai-platform-bucket/ │
│ ├── datasets/ │
│ │ ├── tenant-a/ │
│ │ │ ├── raw/ │
│ │ │ │ ├── 2025/ │
│ │ │ │ │ ├── 01/ │
│ │ │ │ │ └── 02/ │
│ │ │ │ └── 2026/ │
│ │ │ └── processed/ │
│ │ │ └── v1.0/ │
│ │ ├── tenant-b/ │
│ │ │ ├── raw/ │
│ │ │ └── processed/ │
│ │ └── tenant-c/ │
│ │ └── ... │
│ ├── models/ │
│ │ ├── tenant-a/ │
│ │ │ ├── qwen2-7b-instruct/ │
│ │ │ │ ├── v1.0/ │
│ │ │ │ └── v1.1/ │
│ │ │ └── llava-7b/ │
│ │ └── tenant-b/ │
│ ├── checkpoints/ │
│ │ ├── tenant-a/ │
│ │ │ └── training-runs/ │
│ │ └── tenant-b/ │
│ └── outputs/ │
│ ├── tenant-a/ │
│ │ ├── inference-results/ │
│ │ └── evaluation-reports/ │
│ └── tenant-b/ │
│ │
└─────────────────────────────────────────────────────────────────────────┘5.2 Django多租户数据模型
以下是使用Django实现的完整多租户数据模型设计：
# models.py
# 多租户AI平台数据模型设计
import uuid
from django.db import models
from django.contrib.auth.models import User
class Tenant(models.Model):
"""
租户模型 - 代表一个独立的组织/团队
每个租户拥有独立的资源配置和隔离的命名空间
"""
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
name = models.CharField(max_length=100, unique=True, help_text="租户名称")
slug = models.SlugField(max_length=50, unique=True, help_text="租户标识符")
# K8s命名空间关联
kubernetes_namespace = models.CharField(
max_length=100,
unique=True,
help_text="关联的Kubernetes命名空间"
)
# 存储路径前缀
storage_prefix = models.CharField(
max_length=255,
help_text="对象存储路径前缀，如: datasets/tenant-a"
)
# 状态管理
is_active = models.BooleanField(default=True, help_text="租户是否激活")
# 资源配置
max_gpu_count = models.PositiveIntegerField(default=2, help_text="最大GPU数量")
max_storage_gb = models.PositiveIntegerField(default=500, help_text="最大存储空间(GB)")
max_model_count = models.PositiveIntegerField(default=10, help_text="最大模型数量")
# 配额配置
token_quota_monthly = models.BigIntegerField(
default=1000000000,
help_text="每月Token配额"
)
compute_quota_hours_monthly = models.PositiveIntegerField(
default=720,
help_text="每月计算时长配额(小时)"
)
# 计费信息
billing_email = models.EmailField(help_text="账单接收邮箱")
subscription_tier = models.CharField(
max_length=20,
choices=[
('free', '免费版'),
('pro', '专业版'),
('enterprise', '企业版'),
],
default='free'
)
# 时间戳
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
db_table = 'tenants'
ordering = ['-created_at']
def __str__(self):
return f"{self.name} ({self.slug})"
class TenantUser(models.Model):
"""
租户用户关联模型 - 实现用户与租户的多对多关系
一个用户可以属于多个租户，一个租户可以有多个用户
"""
class Role(models.TextChoices):
OWNER = 'owner', '所有者'
ADMIN = 'admin', '管理员'
DEVELOPER = 'developer', '开发者'
VIEWER = 'viewer', '查看者'
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
user = models.ForeignKey(
User,
on_delete=models.CASCADE,
related_name='tenant_memberships'
)
tenant = models.ForeignKey(
Tenant,
on_delete=models.CASCADE,
related_name='user_memberships'
)
role = models.CharField(
max_length=20,
choices=Role.choices,
default=Role.DEVELOPER
)
# K8s RBAC绑定
kubernetes_service_account = models.CharField(
max_length=100,
blank=True,
null=True,
help_text="K8s ServiceAccount名称"
)
is_active = models.BooleanField(default=True)
joined_at = models.DateTimeField(auto_now_add=True)
class Meta:
db_table = 'tenant_users'
unique_together = ['user', 'tenant']
def __str__(self):
return f"{self.user.username} @ {self.tenant.name}"
class Dataset(models.Model):
"""
数据集模型 - 支持版本化的数据集管理
"""
class Status(models.TextChoices):
UPLOADING = 'uploading', '上传中'
PROCESSING = 'processing', '处理中'
READY = 'ready', '就绪'
ARCHIVED = 'archived', '已归档'
class DataType(models.TextChoices):
TEXT = 'text', '文本'
IMAGE = 'image', '图像'
AUDIO = 'audio', '音频'
VIDEO = 'video', '视频'
MULTIMODAL = 'multimodal', '多模态'
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
tenant = models.ForeignKey(
Tenant,
on_delete=models.CASCADE,
related_name='datasets'
)
name = models.CharField(max_length=200, help_text="数据集名称")
description = models.TextField(blank=True, help_text="数据集描述")
# 版本管理
version = models.CharField(max_length=50, default='v1.0.0')
parent_version = models.ForeignKey(
'self',
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name='child_versions',
help_text="父版本（用于追踪版本历史）"
)
# 数据类型和统计
data_type = models.CharField(
max_length=20,
choices=DataType.choices,
default=DataType.TEXT
)
total_size_bytes = models.BigIntegerField(default=0, help_text="总大小(字节)")
file_count = models.PositiveIntegerField(default=0, help_text="文件数量")
record_count = models.BigIntegerField(default=0, help_text="记录数量")
# 存储位置
storage_path = models.CharField(
max_length=500,
help_text="对象存储路径"
)
# 元数据
metadata = models.JSONField(default=dict, help_text="自定义元数据")
tags = models.JSONField(default=list, help_text="标签列表")
status = models.CharField(
max_length=20,
choices=Status.choices,
default=Status.UPLOADING
)
created_by = models.ForeignKey(
User,
on_delete=models.SET_NULL,
null=True,
related_name='created_datasets'
)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
db_table = 'datasets'
unique_together = ['tenant', 'name', 'version']
ordering = ['-created_at']
def __str__(self):
return f"{self.tenant.slug}/{self.name}:{self.version}"
def get_full_path(self):
"""获取完整存储路径"""
return f"{self.tenant.storage_prefix}/datasets/{self.name}/{self.version}"
class ModelVersion(models.Model):
"""
模型版本模型 - 管理模型的多个版本
"""
class Status(models.TextChoices):
REGISTERED = 'registered', '已注册'
VALIDATING = 'validating', '验证中'
DEPLOYED = 'deployed', '已部署'
DEPRECATED = 'deprecated', '已废弃'
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
tenant = models.ForeignKey(
Tenant,
on_delete=models.CASCADE,
related_name='model_versions'
)
# 模型基础信息
name = models.CharField(max_length=200, help_text="模型名称")
base_model = models.CharField(max_length=200, help_text="基础模型")
version = models.CharField(max_length=50, help_text="版本号")
# 模型规格
model_size_params = models.PositiveIntegerField(
help_text="参数量（如：7000000000表示7B）"
)
quantization = models.CharField(
max_length=20,
choices=[
('fp16', 'FP16'),
('fp32', 'FP32'),
('int8', 'INT8'),
('int4', 'INT4'),
('awq', 'AWQ'),
('gptq', 'GPTQ'),
],
default='fp16'
)
# 存储信息
model_path = models.CharField(max_length=500, help_text="模型文件路径")
model_size_bytes = models.BigIntegerField(help_text="模型大小(字节)")
# 配置
config = models.JSONField(default=dict, help_text="模型配置")
requirements = models.JSONField(default=list, help_text="依赖包列表")
status = models.CharField(
max_length=20,
choices=Status.choices,
default=Status.REGISTERED
)
# 评估指标
metrics = models.JSONField(default=dict, help_text="评估指标")
created_by = models.ForeignKey(
User,
on_delete=models.SET_NULL,
null=True
)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
db_table = 'model_versions'
unique_together = ['tenant', 'name', 'version']
ordering = ['-created_at']
def __str__(self):
return f"{self.tenant.slug}/{self.name}:{self.version}"
class QuotaUsage(models.Model):
"""
配额使用记录模型 - 追踪租户的资源使用情况
"""
class QuotaType(models.TextChoices):
GPU_HOURS = 'gpu_hours', 'GPU时长'
TOKEN = 'token', 'Token'
STORAGE = 'storage', '存储'
API_CALLS = 'api_calls', 'API调用'
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
tenant = models.ForeignKey(
Tenant,
on_delete=models.CASCADE,
related_name='quota_usages'
)
quota_type = models.CharField(
max_length=20,
choices=QuotaType.choices
)
# 使用量统计
usage_value = models.BigIntegerField(default=0, help_text="使用量")
usage_period_start = models.DateTimeField(help_text="统计周期开始")
usage_period_end = models.DateTimeField(help_text="统计周期结束")
# 关联资源（可选）
resource_id = models.UUIDField(null=True, blank=True, help_text="关联资源ID")
resource_type = models.CharField(max_length=50, null=True, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
db_table = 'quota_usages'
indexes = [
models.Index(fields=['tenant', 'quota_type', 'usage_period_start']),
]六、配额检查中间件
6.1 配额检查与扣减逻辑
配额管理是多租户AI平台的核心功能之一，需要在每个API请求中验证和扣减配额：
# middleware.py
# 配额检查中间件 - 在请求处理前验证租户配额
import time
import logging
from functools import wraps
from typing import Optional, Callable, Dict, Any
from datetime import datetime, timedelta
from django.http import JsonResponse
from django.core.cache import cache
logger = logging.getLogger(__name__)
class QuotaExceededException(Exception):
"""配额超限异常"""
def __init__(self, quota_type: str, requested: int, available: int):
self.quota_type = quota_type
self.requested = requested
self.available = available
super().__init__(
f"配额不足: {quota_type}, 请求: {requested}, 可用: {available}"
)
class QuotaManager:
"""
配额管理器 - 负责配额的检查、预留和扣减
采用乐观锁机制确保并发安全
"""
# 配额缓存时间（秒）
CACHE_TTL = 60
def __init__(self, tenant_id: str):
self.tenant_id = tenant_id
self._cache_key_prefix = f"quota:{tenant_id}"
def _get_cache_key(self, quota_type: str) -> str:
"""获取配额缓存键"""
return f"{self._cache_key_prefix}:{quota_type}"
def _get_quota_limit(self, quota_type: str) -> Optional[int]:
"""
从数据库获取租户的配额上限
实际实现应从Tenant模型读取
"""
quota_limits = {
'gpu_hours': 720, # 每月720小时
'token': 1000000000, # 每月10亿Token
'storage': 500, # 500GB
'api_calls': 100000, # 每月10万次API调用
}
return quota_limits.get(quota_type)
def _get_current_usage(self, quota_type: str) -> int:
"""
获取当前使用量
优先从缓存读取，缓存未命中时从数据库查询
"""
cache_key = self._get_cache_key(quota_type)
cached_usage = cache.get(cache_key)
if cached_usage is not None:
return cached_usage
# 模拟从数据库查询实际使用量
# 实际实现应查询QuotaUsage表
current_period_start = datetime.now().replace(
day=1, hour=0, minute=0, second=0, microsecond=0
)
# TODO: 实际查询数据库
# usage = QuotaUsage.objects.filter(
# tenant_id=self.tenant_id,
# quota_type=quota_type,
# usage_period_start__gte=current_period_start
# ).aggregate(total=models.Sum('usage_value'))['total'] or 0
usage = 0
cache.set(cache_key, usage, self.CACHE_TTL)
return usage
def check_quota(self, quota_type: str, requested: int = 1) -> Dict[str, Any]:
"""
检查配额是否足够
Args:
quota_type: 配额类型
requested: 请求的数量
Returns:
包含检查结果的字典
"""
limit = self._get_quota_limit(quota_type)
if limit is None:
return {'allowed': True, 'reason': '无配额限制'}
current_usage = self._get_current_usage(quota_type)
available = limit - current_usage
if available < requested:
return {
'allowed': False,
'reason': f'配额不足: {quota_type}',
'requested': requested,
'available': available,
'limit': limit,
'usage': current_usage
}
return {
'allowed': True,
'available': available,
'limit': limit,
'usage': current_usage
}
def reserve_quota(self, quota_type: str, requested: int) -> bool:
"""
预留配额（原子操作）
使用Redis WATCH机制实现乐观锁
Returns:
是否预留成功
"""
limit = self._get_quota_limit(quota_type)
if limit is None:
return True
cache_key = self._get_cache_key(quota_type)
# 获取当前值并检查
current_usage = self._get_current_usage(quota_type)
if current_usage + requested > limit:
logger.warning(
f"配额预留失败: tenant={self.tenant_id}, "
f"type={quota_type}, requested={requested}, "
f"current={current_usage}, limit={limit}"
)
return False
# 预留操作（增量更新）
# 使用Redis INCRBY保证原子性
try:
new_usage = cache.incrby(cache_key, requested)
# 设置过期时间（本月剩余时间）
days_in_month = 31
remaining_seconds = (days_in_month - datetime.now().day + 1) * 86400
cache.expire(cache_key, remaining_seconds)
logger.info(
f"配额预留成功: tenant={self.tenant_id}, "
f"type={quota_type}, reserved={requested}, "
f"new_total={new_usage}"
)
return True
except Exception as e:
logger.error(f"配额预留异常: {e}")
return False
def release_quota(self, quota_type: str, amount: int) -> None:
"""释放预留的配额（用于失败回滚）"""
cache_key = self._get_cache_key(quota_type)
try:
cache.decrby(cache_key, amount)
logger.info(
f"配额释放: tenant={self.tenant_id}, "
f"type={quota_type}, released={amount}"
)
except Exception as e:
logger.error(f"配额释放异常: {e}")
def record_usage(self, quota_type: str, amount: int, resource_id: str = None) -> None:
"""
记录实际使用量到数据库
异步任务实现，实际扣减配额
"""
# TODO: 实现异步任务，将使用量记录到数据库
# QuotaUsageService.record(
# tenant_id=self.tenant_id,
# quota_type=quota_type,
# amount=amount,
# resource_id=resource_id
# )
logger.info(
f"配额使用记录: tenant={self.tenant_id}, "
f"type={quota_type}, amount={amount}"
)
def check_quota(quota_type: str, requested: int = 1):
"""
配额检查装饰器
用法:
@check_quota('token', requested=1000)
def infer(request, model_id, prompt):
...
"""
def decorator(view_func: Callable) -> Callable:
@wraps(view_func)
async def async_wrapper(request, *args, **kwargs):
# 从请求中获取租户ID
tenant_id = getattr(request, 'tenant_id', None)
if not tenant_id:
return JsonResponse({'error': '未指定租户'}, status=400)
manager = QuotaManager(tenant_id)
result = manager.check_quota(quota_type, requested)
if not result['allowed']:
return JsonResponse({
'error': '配额不足',
'quota_type': quota_type,
'requested': requested,
'available': result['available'],
'limit': result['limit'],
'usage': result['usage']
}, status=429)
# 预留配额
if not manager.reserve_quota(quota_type, requested):
return JsonResponse({
'error': '配额预留失败，请重试',
'quota_type': quota_type
}, status=503)
try:
response = await view_func(request, *args, **kwargs)
# 成功后记录实际使用
manager.record_usage(quota_type, requested)
return response
except Exception as e:
# 失败时释放预留
manager.release_quota(quota_type, requested)
raise
@wraps(view_func)
def sync_wrapper(request, *args, **kwargs):
tenant_id = getattr(request, 'tenant_id', None)
if not tenant_id:
return JsonResponse({'error': '未指定租户'}, status=400)
manager = QuotaManager(tenant_id)
result = manager.check_quota(quota_type, requested)
if not result['allowed']:
return JsonResponse({
'error': '配额不足',
'details': result
}, status=429)
if not manager.reserve_quota(quota_type, requested):
return JsonResponse({
'error': '配额预留失败'
}, status=503)
try:
response = view_func(request, *args, **kwargs)
manager.record_usage(quota_type, requested)
return response
except Exception as e:
manager.release_quota(quota_type, requested)
raise
# 根据视图函数类型返回对应包装器
import asyncio
if asyncio.iscoroutinefunction(view_func):
return async_wrapper
return sync_wrapper
return decorator七、计费与用量统计
7.1 用量统计服务
完整的计费和用量统计需要聚合多个数据源：
# billing.py
# 计费与用量统计服务
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import List, Dict, Optional
from decimal import Decimal
import logging
logger = logging.getLogger(__name__)
@dataclass
class UsageRecord:
"""使用记录"""
tenant_id: str
quota_type: str
amount: int
resource_id: str
resource_name: str
timestamp: datetime
metadata: Dict
@dataclass
class BillingItem:
"""账单项目"""
description: str
quantity: Decimal
unit_price: Decimal
total: Decimal
class BillingService:
"""
计费服务 - 计算租户的账单
支持按量付费和订阅制两种模式
"""
# 单价配置（示例）
UNIT_PRICES = {
'gpu_a100_hour': Decimal('15.00'), # A100 GPU/小时
'gpu_v100_hour': Decimal('10.00'), # V100 GPU/小时
'gpu_a10_hour': Decimal('5.00'), # A10 GPU/小时
'token_inference': Decimal('0.0001'), # 推理Token
'token_training': Decimal('0.0002'), # 训练Token
'storage_gb_month': Decimal('0.10'), # 存储GB/月
'api_call': Decimal('0.001'), # API调用
}
def __init__(self, tenant_id: str):
self.tenant_id = tenant_id
def calculate_gpu_cost(self, usage_hours: float, gpu_type: str = 'A100') -> Decimal:
"""计算GPU使用成本"""
unit_key = f'gpu_{gpu_type.lower()}_hour'
unit_price = self.UNIT_PRICES.get(unit_key, Decimal('10.00'))
return Decimal(str(usage_hours)) * unit_price
def calculate_token_cost(self, token_count: int, usage_type: str = 'inference') -> Decimal:
"""计算Token使用成本"""
unit_key = f'token_{usage_type}'
unit_price = self.UNIT_PRICES.get(unit_key, Decimal('0.0001'))
return Decimal(str(token_count)) * unit_price
def calculate_storage_cost(self, storage_gb: float, months: int = 1) -> Decimal:
"""计算存储成本"""
unit_price = self.UNIT_PRICES['storage_gb_month']
return Decimal(str(storage_gb)) * unit_price * Decimal(str(months))
def generate_monthly_bill(
self,
start_date: datetime,
end_date: datetime
) -> Dict:
"""
生成月度账单
Returns:
包含账单详情的字典
"""
billing_items: List[BillingItem] = []
# TODO: 从数据库查询实际使用量
# 示例：假设有以下使用记录
usage_records = self._query_usage_records(start_date, end_date)
for record in usage_records:
if record.quota_type == 'gpu_hours':
cost = self.calculate_gpu_cost(
record.amount / 3600, # 转换为小时
gpu_type=record.metadata.get('gpu_type', 'A100')
)
billing_items.append(BillingItem(
description=f"GPU计算时长: {record.metadata.get('gpu_type', 'A100')}",
quantity=Decimal(str(record.amount / 3600)),
unit_price=self.UNIT_PRICES.get(
f"gpu_{record.metadata.get('gpu_type', 'A100').lower()}_hour",
Decimal('10.00')
),
total=cost
))
elif record.quota_type == 'token':
cost = self.calculate_token_cost(
record.amount,
usage_type=record.metadata.get('usage_type', 'inference')
)
billing_items.append(BillingItem(
description=f"Token使用: {record.metadata.get('usage_type', 'inference')}",
quantity=Decimal(str(record.amount)),
unit_price=self.UNIT_PRICES.get(
f"token_{record.metadata.get('usage_type', 'inference')}",
Decimal('0.0001')
),
total=cost
))
elif record.quota_type == 'storage':
cost = self.calculate_storage_cost(record.amount / (1024**3))
billing_items.append(BillingItem(
description="对象存储",
quantity=Decimal(str(record.amount / (1024**3))),
unit_price=self.UNIT_PRICES['storage_gb_month'],
total=cost
))
# 计算总价
subtotal = sum(item.total for item in billing_items)
tax = subtotal * Decimal('0.06') # 6%税率
total = subtotal + tax
return {
'tenant_id': self.tenant_id,
'billing_period': {
'start': start_date.isoformat(),
'end': end_date.isoformat()
},
'items': [
{
'description': item.description,
'quantity': float(item.quantity),
'unit_price': float(item.unit_price),
'total': float(item.total)
}
for item in billing_items
],
'subtotal': float(subtotal),
'tax': float(tax),
'total': float(total),
'currency': 'CNY'
}
def _query_usage_records(
self,
start: datetime,
end: datetime
) -> List[UsageRecord]:
"""
查询使用记录
实际实现应从数据库查询
"""
# 模拟数据
return [
UsageRecord(
tenant_id=self.tenant_id,
quota_type='gpu_hours',
amount=7200, # 2小时
resource_id='training-run-001',
resource_name='qwen-finetune-job',
timestamp=datetime.now() - timedelta(days=5),
metadata={'gpu_type': 'A100', 'instance_type': 'train'}
),
UsageRecord(
tenant_id=self.tenant_id,
quota_type='token',
amount=1000000,
resource_id='inference-001',
resource_name='API调用',
timestamp=datetime.now() - timedelta(days=1),
metadata={'usage_type': 'inference', 'model': 'qwen2-7b'}
),
]
class UsageAggregationService:
"""
用量聚合服务 - 聚合多租户的实时使用量
"""
def __init__(self):
# Redis连接用于实时聚合
self.redis_client = None # 实际应初始化Redis连接
async def get_realtime_usage(self, tenant_id: str) -> Dict:
"""获取租户的实时使用量"""
# 从Redis获取实时计数器
# 实际实现应查询Redis
return {
'tenant_id': tenant_id,
'gpu_hours': 125.5,
'gpu_utilization': 0.78,
'token_usage': 5000000,
'storage_used_gb': 125.5,
'api_calls_today': 1500,
'timestamp': datetime.now().isoformat()
}
async def get_usage_trend(
self,
tenant_id: str,
days: int = 30
) -> Dict:
"""获取使用趋势"""
# 从时序数据库查询历史数据
# 实际实现应查询InfluxDB/Prometheus
return {
'tenant_id': tenant_id,
'period_days': days,
'daily_usage': [
{
'date': (datetime.now() - timedelta(days=i)).strftime('%Y-%m-%d'),
'gpu_hours': 10.5 + i * 0.5,
'token_count': 500000 + i * 10000,
'api_calls': 1000 + i * 50
}
for i in range(days)
][::-1],
'total': {
'gpu_hours': sum(10.5 + i * 0.5 for i in range(days)),
'token_count': sum(500000 + i * 10000 for i in range(days)),
'api_calls': sum(1000 + i * 50 for i in range(days))
}
}八、总结与展望
8.1 设计要点回顾
本文详细介绍了多租户AI平台的资源隔离与配额管理设计，主要涵盖以下几个核心维度：
计算资源隔离：通过Kubernetes命名空间实现基础隔离，结合ResourceQuota和LimitRange实现精细化的资源配额管理。GPU污点与容忍机制确保了推理服务能够正确调度到GPU节点。
模型服务隔离：分析了独立推理服务和共享推理服务两种模式的优劣，企业可根据实际需求选择。独立模式安全性高但成本较高，共享模式资源利用率高但需要额外的隔离机制。
数据隔离：采用对象存储路径前缀策略实现数据隔离，配合访问控制确保租户数据的机密性。这种轻量级的隔离方案在保证安全的同时具有较高的灵活性。
配额管理：通过配额检查中间件实现了API级别的配额验证和扣减，采用缓存+数据库的双层架构平衡性能和一致性。配额预留机制确保了高并发场景下的配额安全。
计费统计：完整的计费服务支持按量付费和订阅制两种模式，用量聚合服务提供了实时和历史的使用统计。
8.2 未来优化方向
多租户AI平台的建设是一个持续迭代的过程，以下是一些值得考虑的优化方向：
更精细的资源隔离：可以考虑引入GPU虚拟化技术（如NVIDIA vGPU、MIG），实现单卡多租户的细粒度隔离，进一步提高资源利用率。
智能配额调度：引入基于历史使用模式的智能配额推荐，根据租户的增长趋势动态调整配额，避免资源浪费。
跨云多集群管理：支持跨多个云服务商和本地集群的统一配额管理，实现真正的混合云多租户架构。
实时成本可视化：增强用量统计的实时性，提供更精细的成本分析和优化建议，帮助租户降低AI使用成本。
本文从工程实践角度出发，详细介绍了多租户AI平台的设计思路和实现方案，希望能为正在进行AI平台建设的团队提供一些参考。
评论区