引言

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平台的最佳选择。无论选择哪种模式，都必须建立完善的配额管理、成本分摊和安全边界机制。

核心原则：多租户系统的成功不仅取决于技术架构，更取决于成本透明度和安全可信度。租户需要清楚知道自己的资源消耗和费用，平台需要确保租户间的数据安全和行为隔离。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。