多租户治理

SaaS 化的 LLM 平台需要支持多租户隔离，每个租户可以有独立的治理策略、规则集和资源配额。

多租户架构

租户配置系统

from dataclasses import dataclass, field
from enum import Enum
class GovernanceLevel(Enum):
STRICT = "严格"     # 金融、医疗
STANDARD = "标准"   # 一般企业
RELAXED = "宽松"    # 内部开发工具
CUSTOM = "自定义"
@dataclass
class TenantQuota:
"""租户资源配额"""
max_requests_per_day: int = 10000
max_tokens_per_request: int = 4096
max_concurrent_requests: int = 50
allowed_models: list[str] = field(
default_factory=lambda: ["gpt-4o-mini", "gpt-4o"]
)
@dataclass
class TenantPolicy:
"""租户治理策略"""
tenant_id: str
name: str
level: GovernanceLevel
quota: TenantQuota
blocked_categories: list[str] = field(default_factory=list)
custom_keywords: list[str] = field(default_factory=list)
require_human_review: bool = False
data_retention_days: int = 90
pii_masking: bool = True
output_disclaimer: str = ""
class TenantManager:
"""多租户管理器"""
# 预定义策略模板
TEMPLATES: dict[GovernanceLevel, dict] = {
GovernanceLevel.STRICT: {
"blocked_categories": [
"violence", "sexual", "self_harm", "hate",
"illegal", "financial_advice", "medical_advice"
],
"require_human_review": True,
"pii_masking": True,
"data_retention_days": 365,
"output_disclaimer": "本回答由AI生成，仅供参考，不构成专业建议。",
},
GovernanceLevel.STANDARD: {
"blocked_categories": [
"violence", "sexual", "self_harm", "hate", "illegal"
],
"require_human_review": False,
"pii_masking": True,
"data_retention_days": 90,
"output_disclaimer": "",
},
GovernanceLevel.RELAXED: {
"blocked_categories": ["illegal", "self_harm"],
"require_human_review": False,
"pii_masking": False,
"data_retention_days": 30,
"output_disclaimer": "",
},
}
def __init__(self):
self.tenants: dict[str, TenantPolicy] = {}
def register_tenant(
self,
tenant_id: str,
name: str,
level: GovernanceLevel,
quota: TenantQuota | None = None,
) -> TenantPolicy:
"""注册新租户"""
template = self.TEMPLATES.get(level, {})
policy = TenantPolicy(
tenant_id=tenant_id,
name=name,
level=level,
quota=quota or TenantQuota(),
blocked_categories=template.get("blocked_categories", []),
require_human_review=template.get("require_human_review", False),
pii_masking=template.get("pii_masking", True),
data_retention_days=template.get("data_retention_days", 90),
output_disclaimer=template.get("output_disclaimer", ""),
)
self.tenants[tenant_id] = policy
return policy
def get_policy(self, tenant_id: str) -> TenantPolicy | None:
return self.tenants.get(tenant_id)
def update_quota(self, tenant_id: str, quota: TenantQuota) -> bool:
policy = self.tenants.get(tenant_id)
if policy:
policy.quota = quota
return True
return False
# 使用示例
manager = TenantManager()
bank = manager.register_tenant(
"t_bank_001", "某银行",
GovernanceLevel.STRICT,
TenantQuota(max_requests_per_day=50000, max_tokens_per_request=8192)
)
print(f"租户: {bank.name}, 治理级别: {bank.level.value}")
print(f"屏蔽类别: {bank.blocked_categories}")

租户隔离执行器

from dataclasses import dataclass
import time
@dataclass
class TenantUsage:
requests_today: int = 0
tokens_today: int = 0
concurrent: int = 0
last_reset: float = 0.0
class TenantIsolationEnforcer:
"""租户隔离执行器——确保配额和策略隔离"""
def __init__(self, manager: TenantManager):
self.manager = manager
self.usage: dict[str, TenantUsage] = {}
def check_quota(self, tenant_id: str, token_count: int) -> dict:
"""检查租户配额"""
policy = self.manager.get_policy(tenant_id)
if not policy:
return {"allowed": False, "reason": "租户不存在"}
usage = self.usage.setdefault(tenant_id, TenantUsage())
# 每日重置
now = time.time()
if now - usage.last_reset > 86400:
usage.requests_today = 0
usage.tokens_today = 0
usage.last_reset = now
# 检查日请求量
if usage.requests_today >= policy.quota.max_requests_per_day:
return {"allowed": False, "reason": "日请求量已达上限"}
# 检查单次 token 数
if token_count > policy.quota.max_tokens_per_request:
return {
"allowed": False,
"reason": f"Token数 {token_count} 超过限制 "
f"{policy.quota.max_tokens_per_request}"
}
# 检查并发
if usage.concurrent >= policy.quota.max_concurrent_requests:
return {"allowed": False, "reason": "并发请求已达上限"}
return {"allowed": True}
def record_usage(self, tenant_id: str, tokens: int) -> None:
usage = self.usage.setdefault(tenant_id, TenantUsage())
usage.requests_today += 1
usage.tokens_today += tokens
def get_usage_summary(self, tenant_id: str) -> dict:
policy = self.manager.get_policy(tenant_id)
usage = self.usage.get(tenant_id, TenantUsage())
if not policy:
return {}
return {
"tenant": policy.name,
"requests": f"{usage.requests_today}/{policy.quota.max_requests_per_day}",
"tokens": f"{usage.tokens_today}",
"utilization": usage.requests_today / policy.quota.max_requests_per_day
if policy.quota.max_requests_per_day > 0 else 0,
}

治理级别对比

本章小结

• 预定义三级模板——严格/标准/宽松满足不同行业需求
• 租户资源配额——日请求量、token 上限、并发数独立配置
• 策略隔离——每个租户独立的治理规则和屏蔽类别
• 用量追踪——实时统计每个租户的资源消耗
• 免责声明——高风险行业自动在输出尾部添加

下一章：Docker 部署