当前位置：首页 > news >正文

企业级 Agent 产品：多租户隔离与资源配额的架构设计

news 2026/6/12 2:06:58

企业级 Agent 产品：多租户隔离与资源配额的架构设计

一、共享集群的公平性危机：当大客户吃掉所有 GPU

企业级 Agent 产品通常采用多租户架构——多个客户共享同一套计算集群。当某个大客户发起大量并发推理请求时，可能占满所有 GPU 资源，导致其他客户的请求排队等待甚至超时。这种"资源抢占"问题不仅影响用户体验，还可能违反 SLA 合同中的延迟承诺。

更复杂的是，不同客户对资源的需求模式差异巨大：A 客户白天高峰期需要大量 GPU，B 客户则是夜间批量处理。静态分配资源会导致白天 A 浪费 B 的配额，夜间 B 浪费 A 的配额。需要一套动态的、可配置的资源隔离和配额管理系统。

flowchart TB subgraph 无隔离 R1[请求队列] --> GPU1[GPU 池] T1[租户A<br/>100并发] --> R1 T2[租户B<br/>5并发] --> R1 Note1[租户A占满GPU<br/>租户B饥饿等待] -.-> GPU1 end subgraph 多租户隔离 T3[租户A] --> Q1[租户A队列<br/>配额: 80并发] T4[租户B] --> Q2[租户B队列<br/>配额: 20并发] Q1 --> S[调度器] Q2 --> S S --> GPU2[GPU 池] Note2[按配额分配<br/>保证最低资源] -.-> S end

二、多租户隔离的核心机制

2.1 三层隔离模型

多租户隔离分为三个层次：网络隔离（不同租户的流量互不可见）、计算隔离（GPU/CPU 资源按配额分配）、存储隔离（模型权重和数据分区存储）。对于 Agent 产品，计算隔离是核心——推理请求的 GPU 占用是最大的资源瓶颈。

2.2 令牌桶与优先级调度

令牌桶算法是资源配额的经典实现：每个租户有一个令牌桶，请求消耗令牌，令牌以固定速率补充。当令牌耗尽时，新请求被限流或排队。优先级调度则在资源竞争时决定谁先获得 GPU——高优先级租户的请求优先调度，低优先级租户使用剩余资源。

sequenceDiagram participant TA as 租户A participant TB as 租户B participant Scheduler as 调度器 participant Quota as 配额管理器 participant GPU as GPU池 TA->>Scheduler: 提交50个推理请求 Scheduler->>Quota: 查询租户A配额 Quota-->>Scheduler: 配额80，已用30，剩余50 Scheduler->>GPU: 调度50个请求 TB->>Scheduler: 提交30个推理请求 Scheduler->>Quota: 查询租户B配额 Quota-->>Scheduler: 配额20，已用0，剩余20 Scheduler->>GPU: 调度20个请求 Scheduler->>TB: 10个请求排队等待 Note over TB: 等待租户A释放资源<br/>或令牌补充

三、生产级代码实现

3.1 配额管理器

import time import asyncio import logging from dataclasses import dataclass, field from typing import Dict, Optional from enum import Enum logger = logging.getLogger(__name__) class Priority(Enum): LOW = 0 NORMAL = 1 HIGH = 2 CRITICAL = 3 @dataclass class TenantQuota: """租户配额定义""" tenant_id: str max_concurrent: int = 10 # 最大并发推理数 tokens_per_minute: int = 100000 # 每分钟 Token 配额 priority: Priority = Priority.NORMAL gpu_share: float = 1.0 # GPU 份额（0.0-1.0） @dataclass class TokenBucket: """令牌桶：控制每分钟 Token 消耗速率""" capacity: int # 桶容量（最大突发量） tokens: float # 当前令牌数 refill_rate: float # 每秒补充速率 last_refill: float # 上次补充时间 def consume(self, amount: int) -> bool: """尝试消耗令牌，成功返回 True""" self._refill() if self.tokens >= amount: self.tokens -= amount return True return False def _refill(self) -> None: """补充令牌""" now = time.time() elapsed = now - self.last_refill self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate) self.last_refill = now class QuotaManager: """配额管理器：管理多租户的资源配额和令牌桶 设计考量： - 每个租户独立的令牌桶，防止一个租户耗尽全局 Token 配额 - 并发数硬限制：超过最大并发的请求直接排队 - 优先级调度：高优先级租户在资源竞争时优先获得 GPU """ def __init__(self): self._quotas: Dict[str, TenantQuota] = {} self._buckets: Dict[str, TokenBucket] = {} self._current_concurrent: Dict[str, int] = {} self._lock = asyncio.Lock() def register_tenant(self, quota: TenantQuota) -> None: """注册租户配额""" self._quotas[quota.tenant_id] = quota self._buckets[quota.tenant_id] = TokenBucket( capacity=quota.tokens_per_minute, tokens=quota.tokens_per_minute, refill_rate=quota.tokens_per_minute / 60.0, last_refill=time.time(), ) self._current_concurrent[quota.tenant_id] = 0 async def acquire( self, tenant_id: str, estimated_tokens: int = 1000, ) -> bool: """尝试获取资源配额 Returns: True 表示配额充足，可以执行推理 False 表示配额不足，需要排队或拒绝 """ async with self._lock: quota = self._quotas.get(tenant_id) if not quota: logger.error(f"未知租户: {tenant_id}") return False # 检查并发数限制 if self._current_concurrent[tenant_id] >= quota.max_concurrent: logger.info(f"租户 {tenant_id} 并发数已达上限: {quota.max_concurrent}") return False # 检查令牌桶 bucket = self._buckets[tenant_id] if not bucket.consume(estimated_tokens): logger.info(f"租户 {tenant_id} Token 配额不足") return False # 分配成功 self._current_concurrent[tenant_id] += 1 return True async def release(self, tenant_id: str, actual_tokens: int = 0) -> None: """释放资源配额 Args: actual_tokens: 实际消耗的 Token 数，用于修正令牌桶 """ async with self._lock: if tenant_id in self._current_concurrent: self._current_concurrent[tenant_id] = max( 0, self._current_concurrent[tenant_id] - 1 ) # 如果实际消耗与预估差异超过 20%，记录日志用于优化预估 if actual_tokens > 0: estimated = 1000 # 默认预估值 if abs(actual_tokens - estimated) / max(estimated, 1) > 0.2: logger.info( f"租户 {tenant_id} Token 估算偏差: " f"预估={estimated}, 实际={actual_tokens}" ) def get_priority(self, tenant_id: str) -> Priority: """获取租户优先级""" quota = self._quotas.get(tenant_id) return quota.priority if quota else Priority.NORMAL def get_status(self, tenant_id: str) -> Dict: """获取租户当前资源使用状态""" quota = self._quotas.get(tenant_id) bucket = self._buckets.get(tenant_id) return { "tenant_id": tenant_id, "concurrent": self._current_concurrent.get(tenant_id, 0), "max_concurrent": quota.max_concurrent if quota else 0, "tokens_remaining": int(bucket.tokens) if bucket else 0, "priority": quota.priority.value if quota else 0, }

3.2 优先级调度器

@dataclass class InferenceRequest: """推理请求""" request_id: str tenant_id: str prompt: str estimated_tokens: int = 1000 submitted_at: float = field(default_factory=time.time) class PriorityScheduler: """优先级调度器：按租户优先级和提交时间排序请求 设计考量： - 同优先级按 FIFO 排序，保证公平性 - 高优先级请求可以插队，但不能抢占正在执行的请求 - 支持超时丢弃：等待超过阈值的请求自动取消 """ MAX_WAIT_SECONDS = 30 # 最大等待时间 def __init__(self, quota_manager: QuotaManager, max_gpu_slots: int = 8): self.quota_manager = quota_manager self.max_gpu_slots = max_gpu_slots self._active_slots = 0 self._queue: list[InferenceRequest] = [] self._lock = asyncio.Lock() self._condition = asyncio.Condition(self._lock) async def submit(self, request: InferenceRequest) -> bool: """提交推理请求""" async with self._condition: # 尝试直接获取配额 if self._active_slots < self.max_gpu_slots: if await self.quota_manager.acquire(request.tenant_id, request.estimated_tokens): self._active_slots += 1 return True # 配额不足或 GPU 满载，加入等待队列 self._queue.append(request) # 按优先级降序、提交时间升序排序 self._queue.sort( key=lambda r: ( -self.quota_manager.get_priority(r.tenant_id).value, r.submitted_at, ) ) return False async def wait_for_slot(self, request: InferenceRequest) -> bool: """等待可用槽位，超时返回 False""" deadline = request.submitted_at + self.MAX_WAIT_SECONDS while time.time() < deadline: async with self._condition: if await self.quota_manager.acquire(request.tenant_id, request.estimated_tokens): if request in self._queue: self._queue.remove(request) self._active_slots += 1 return True remaining = deadline - time.time() if remaining <= 0: break await asyncio.wait_for( self._condition.wait(), timeout=min(remaining, 1.0), ) # 超时，从队列移除 if request in self._queue: self._queue.remove(request) return False async def complete(self, tenant_id: str, actual_tokens: int = 0) -> None: """请求完成，释放槽位""" async with self._condition: self._active_slots -= 1 await self.quota_manager.release(tenant_id, actual_tokens) self._condition.notify_all()