配额管理

引言 Agent SaaS平台在2026年面临的核心挑战之一是多租户架构设计。不同租户的Agent可能使用不同的模型、不同的工具集、不同的Prompt模板，且对性能、安全性和成本的要求差异巨大。如何在共享基础设施上实现高效的资源隔离和公平的成本分摊，是Agent平台架构师必须解决的问题。 多租户隔离模型 三种隔离级别 隔离程度 ──────────────────────────────────▶ 强 ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ 共享模式 │ │ 混合模式 │ │ 独占模式 │ │ │ │ │ │ │ │ 共享所有资源 │ │ 共享计算资源 │ │ 独立资源栈 │ │ 逻辑隔离数据 │ │ 隔离存储资源 │ │ 物理隔离 │ │ │ │ │ │ │ │ 成本最低 │ │ 平衡 │ │ 隔离最强 │ │ 隔离最弱 │ │ │ │ 成本最高 │ └──────────────┘ └──────────────┘ └──────────────┘ from enum import Enum class IsolationLevel(Enum): SHARED = "shared" # 共享模式：所有租户共享同一Agent实例 HYBRID = "hybrid" # 混合模式：共享计算，隔离存储 DEDICATED = "dedicated" # 独占模式：每个租户独立资源栈 class TenantConfig: """租户配置""" def __init__( self, tenant_id: str, tier: str, # free, pro, enterprise isolation: IsolationLevel, quota: dict, custom_config: dict = None ): self.tenant_id = tenant_id self.tier = tier self.isolation = isolation self.quota = quota self.custom_config = custom_config or {} # 根据tier设置默认配额 if not quota: self.quota = self._default_quota(tier) @staticmethod def _default_quota(tier: str) -> dict: defaults = { "free": { "max_sessions": 10, "max_concurrent": 2, "max_tokens_per_day": 100000, "max_tools": 5, "max_memory_mb": 256, "rate_limit_rpm": 20, # 每分钟请求数 }, "pro": { "max_sessions": 100, "max_concurrent": 10, "max_tokens_per_day": 2000000, "max_tools": 20, "max_memory_mb": 2048, "rate_limit_rpm": 200, }, "enterprise": { "max_sessions": -1, # 无限 "max_concurrent": 100, "max_tokens_per_day": 50000000, "max_tools": -1, "max_memory_mb": 32768, "rate_limit_rpm": 2000, } } return defaults.get(tier, defaults["free"]) 资源隔离实现 计算资源隔离 class TenantResourceManager: """租户资源管理器""" def __init__(self, k8s_client): self.k8s = k8s_client self.tenant_pools = {} # tenant_id -> resource pool async def get_or_create_pool( self, tenant: TenantConfig ) -> str: """获取或创建租户资源池""" if tenant.tenant_id in self.tenant_pools: return self.tenant_pools[tenant.tenant_id] if tenant.isolation == IsolationLevel.DEDICATED: # 独占模式：创建独立namespace和资源 namespace = await self._create_dedicated_namespace(tenant) await self._deploy_dedicated_resources(tenant, namespace) self.tenant_pools[tenant.tenant_id] = namespace elif tenant.isolation == IsolationLevel.HYBRID: # 混合模式：使用共享namespace但设置ResourceQuota await self._apply_resource_quota(tenant) self.tenant_pools[tenant.tenant_id] = "shared" else: # 共享模式：仅通过应用层隔离 self.tenant_pools[tenant.tenant_id] = "shared" return self.tenant_pools[tenant.tenant_id] async def _apply_resource_quota(self, tenant: TenantConfig): """应用K8s ResourceQuota""" quota_yaml = { "apiVersion": "v1", "kind": "ResourceQuota", "metadata": { "name": f"quota-{tenant.tenant_id}", "namespace": "agent-shared" }, "spec": { "hard": { "requests.cpu": f"{tenant.quota['max_cpu']}", "requests.memory": f"{tenant.quota['max_memory_mb']}Mi", "pods": str(tenant.quota["max_pods"]), } } } await self.k8s.apply_resource(quota_yaml) class TenantRateLimiter: """租户级限流器""" def __init__(self, redis_client): self.redis = redis_client async def check_and_consume( self, tenant_id: str, resource: str, # "api_call", "token", "tool_exec" amount: int = 1 ) -> bool: """检查配额并消费""" # 滑动窗口限流 key = f"quota:{tenant_id}:{resource}:{datetime.now().strftime('%Y%m%d%H%M')}" pipe = self.redis.pipeline() pipe.incr(key, amount) pipe.expire(key, 3600) # 1小时TTL results = await pipe.execute() current_usage = results[0] limit = await self._get_limit(tenant_id, resource) if current_usage > limit: # 回滚消费 await self.redis.decr(key, amount) return False return True 数据隔离 class TenantDataIsolation: """租户数据隔离管理""" def __init__(self, db_client): self.db = db_client async def execute_for_tenant( self, tenant_id: str, query: str, params: tuple = None ): """在租户上下文中执行查询""" # 方式1：Row-Level Security (PostgreSQL RLS) await self.db.execute( f"SET app.current_tenant = '{tenant_id}'" ) try: result = await self.db.fetch(query, *(params or ())) return result finally: await self.db.execute("RESET app.current_tenant") async def get_vector_store_for_tenant( self, tenant_id: str, collection_name: str ): """获取租户专属的向量存储""" # 使用租户ID作为namespace前缀 namespaced_collection = f"tenant_{tenant_id}_{collection_name}" return VectorStore( collection=namespaced_collection, metadata_filter={"tenant_id": tenant_id} # 双重保障 ) 成本分摊模型 class CostAllocator: """成本分摊器——精确追踪每租户的资源消耗""" # 2026年典型成本基准（美元） COST_RATES = { "llm_token_input": 0.00001, # per token "llm_token_output": 0.00003, # per token "embedding_token": 0.0000001, # per token "vector_search": 0.0001, # per 1k queries "tool_execution": 0.001, # per execution "memory_storage_gb_month": 0.10, "gpu_hour": 2.50, # per GPU hour "cpu_hour": 0.05, } def __init__(self, metrics_store): self.metrics = metrics_store async def record_usage( self, tenant_id: str, resource: str, amount: float, session_id: str = None ): """记录资源使用""" cost = amount * self.COST_RATES.get(resource, 0) await self.metrics.insert({ "tenant_id": tenant_id, "resource": resource, "amount": amount, "cost": cost, "session_id": session_id, "timestamp": datetime.now() }) async def calculate_bill( self, tenant_id: str, period_start: datetime, period_end: datetime ) -> dict: """计算租户账单""" usage = await self.metrics.aggregate( tenant_id=tenant_id, start=period_start, end=period_end ) bill = { "tenant_id": tenant_id, "period": f"{period_start.date()} to {period_end.date()}", "items": [], "total": 0 } for resource, amount in usage.items(): rate = self.COST_RATES.get(resource, 0) cost = amount * rate bill["items"].append({ "resource": resource, "amount": amount, "rate": rate, "cost": round(cost, 4) }) bill["total"] += cost # 应用tier折扣 tier = await self._get_tenant_tier(tenant_id) discount = {"free": 0, "pro": 0.1, "enterprise": 0.25}.get(tier, 0) bill["discount"] = round(bill["total"] * discount, 2) bill["final_total"] = round(bill["total"] - bill["discount"], 2) return bill 实时成本监控 class RealtimeCostMonitor: """实时成本监控与告警""" async def monitor_tenant(self, tenant_id: str): """监控租户实时成本""" while True: daily_cost = await self._get_daily_cost(tenant_id) budget = await self._get_budget(tenant_id) utilization = daily_cost / budget if budget > 0 else 0 if utilization > 0.9: await self._alert( tenant_id=tenant_id, level="critical", message=f"Budget at {utilization:.0%}: ${daily_cost:.2f}/${budget:.2f}" ) # 触发降级或限流 if utilization > 1.0: await self._throttle_tenant(tenant_id) elif utilization > 0.7: await self._alert( tenant_id=tenant_id, level="warning", message=f"Budget at {utilization:.0%}" ) await asyncio.sleep(60) # 每分钟检查 租户级配置管理 class TenantConfigManager: """租户配置管理器""" async def get_agent_config(self, tenant_id: str) -> dict: """获取租户专属的Agent配置""" base_config = { "model": "gpt-4o-mini", "temperature": 0.7, "max_tokens": 4096, "tools": ["search", "calculator"], "system_prompt": "You are a helpful assistant.", "safety_level": "standard" } # 合并租户自定义配置 tenant_overrides = await self._load_tenant_overrides(tenant_id) config = {**base_config, **tenant_overrides} # 应用tier限制 tier = await self._get_tier(tenant_id) if tier == "free": config["model"] = "gpt-4o-mini" # 限制免费用户使用小模型 config["max_tokens"] = min(config["max_tokens"], 2048) return config async def validate_config_change( self, tenant_id: str, new_config: dict ) -> dict: """验证租户配置变更""" tier = await self._get_tier(tenant_id) tier_limits = self.TIER_LIMITS[tier] errors = [] # 检查模型权限 if new_config.get("model") not in tier_limits["allowed_models"]: errors.append(f"Model {new_config['model']} not available for {tier} tier") # 检查工具数量 if len(new_config.get("tools", [])) > tier_limits["max_tools"]: errors.append(f"Too many tools for {tier} tier") # 检查安全级别 if new_config.get("safety_level") == "none" and tier != "enterprise": errors.append("Safety level 'none' requires enterprise tier") return {"valid": len(errors) == 0, "errors": errors} 安全边界设计 class TenantSecurityBoundary: """租户安全边界""" async def enforce_boundary(self, tenant_id: str, request: dict): """执行安全边界检查""" # 1. 防止跨租户数据访问 if request.get("target_tenant") and request["target_tenant"] != tenant_id: raise SecurityViolation("Cross-tenant access denied") # 2. 工具白名单检查 allowed_tools = await self._get_allowed_tools(tenant_id) for tool in request.get("tools", []): if tool not in allowed_tools: raise SecurityViolation(f"Tool '{tool}' not allowed for tenant") # 3. 出站请求域名白名单 if request.get("api_endpoints"): allowed_domains = await self._get_allowed_domains(tenant_id) for endpoint in request["api_endpoints"]: domain = urllib.parse.urlparse(endpoint).hostname if domain not in allowed_domains: raise SecurityViolation(f"Domain '{domain}' not allowed") # 4. 敏感操作审计 if request.get("action") in ["file_write", "code_exec", "network_access"]: await self._audit_log(tenant_id, request) 总结 Agent多租户架构的核心是在资源共享与租户隔离之间找到平衡点。共享模式成本最低但隔离最弱，独占模式隔离最强但成本最高，混合模式是大多数SaaS平台的最佳选择。无论选择哪种模式，都必须建立完善的配额管理、成本分摊和安全边界机制。 ...