为什么多租户 LLM 架构不同于传统 SaaS

传统 SaaS 多租户解决的是数据隔离问题——数据库加个 tenant_id 列基本就搞定了。LLM SaaS 不一样，它涉及计算资源隔离、模型权重隔离、Prompt 安全、Token 计费四层问题，每一层都有坑。

核心矛盾：LLM 推理极其昂贵，你不可能给每个租户分配独立 GPU，但又不能让大租户的请求挤占小租户的资源。

租户隔离的四个层次

层次一：数据隔离

数据隔离是最基础的要求。用户输入的 Prompt、对话历史、上传的文档，必须严格按租户隔离。

# 租户上下文注入中间件
class TenantContextMiddleware:
    def __init__(self, db_pool, redis):
        self.db_pool = db_pool
        self.redis = redis

    async def __call__(self, request):
        tenant_id = request.headers.get("X-Tenant-Id")
        if not tenant_id:
            raise HTTPException(401, "Missing tenant context")

        # 从 Redis 加载租户配置（带缓存）
        tenant_config = await self.redis.get(f"tenant:{tenant_id}:config")
        if not tenant_config:
            tenant_config = await self.load_tenant_config(tenant_id)
            await self.redis.setex(f"tenant:{tenant_id}:config", 300, tenant_config)

        request.state.tenant = json.loads(tenant_config)
        request.state.tenant_id = tenant_id
        return request

    async def load_tenant_config(self, tenant_id):
        async with self.db_pool.acquire() as conn:
            row = await conn.fetchrow(
                "SELECT * FROM tenants WHERE id = $1 AND status = 'active'",
                tenant_id
            )
            if not row:
                raise HTTPException(403, "Tenant not found or inactive")
            return json.dumps(dict(row))

关键原则： 所有数据查询必须强制带 tenant_id 条件，最好在 ORM 层面做硬约束，而不是靠开发者自觉。

层次二：模型隔离

不同租户可能需要不同的模型配置：

隔离级别描述成本适用场景
共享模型所有租户用同一模型低通用聊天、标准化任务
共享模型 + 独立 Prompt同一模型，各租户独立 System Prompt低品牌定制、角色设定
LoRA 适配器共享共享基座模型 + 租户专属 LoRA中领域定制（法律/医疗）
独立模型实例租户独占模型部署高合规要求、数据不出域
独立 GPU 集群物理隔离极高政务/军工级别
# 模型路由器：根据租户级别选择部署策略
class ModelRouter:
    def __init__(self):
        self.shared_pool = SharedModelPool()
        self.lora_manager = LoRAManager()
        self.dedicated_manager = DedicatedModelManager()

    async def route(self, tenant_id: str, request: ChatRequest):
        tenant = await self.get_tenant_config(tenant_id)

        if tenant["tier"] == "enterprise" and tenant.get("dedicated_model"):
            # 企业版：独立模型实例
            return await self.dedicated_manager.inference(tenant_id, request)

        if tenant.get("lora_adapter"):
            # Pro 版：共享基座 + 租户 LoRA
            base_model = self.shared_pool.acquire("llama-3-70b")
            return await self.lora_manager.inference(
                base_model, tenant["lora_adapter"], request
            )

        # 标准版：纯共享模型
        return await self.shared_pool.inference("llama-3-70b", request)

层次三：Prompt 隔离

Prompt 隔离防止租户 A 的对话内容泄漏到租户 B 的上下文中。这不仅是数据库层面的问题，还涉及：

  • KV Cache 隔离：不同租户的请求不能复用 KV Cache
  • 上下文窗口清理：请求结束后立即清理上下文
  • System Prompt 保护：租户的定制 Prompt 不能被终端用户提取
# KV Cache 管理器：确保租户间不共享缓存
class KVCacheManager:
    def __init__(self):
        self.cache_keys = {}  # tenant_id -> set of cache_keys

    def allocate(self, tenant_id: str, session_id: str) -> str:
        cache_key = f"{tenant_id}:{session_id}:{uuid4()}"
        self.cache_keys.setdefault(tenant_id, set()).add(cache_key)
        return cache_key

    def invalidate_tenant(self, tenant_id: str):
        """租户注销或密钥轮换时，清除所有缓存"""
        for key in self.cache_keys.pop(tenant_id, set()):
            vllm_server.drop_cache(key)

层次四：资源隔离

GPU 是最稀缺的资源。资源隔离确保一个租户的流量洪峰不会饿死其他租户。

# 基于令牌桶的租户级限流
class TenantRateLimiter:
    def __init__(self, redis):
        self.redis = redis

    async def acquire(self, tenant_id: str, tokens: int) -> bool:
        # 多维度限流：TPM (tokens per minute) + RPM (requests per minute)
        tpm_key = f"rl:{tenant_id}:tpm"
        rpm_key = f"rl:{tenant_id}:rpm"

        pipe = self.redis.pipeline()
        pipe.incrby(tpm_key, tokens)
        pipe.expire(tpm_key, 60)
        pipe.incr(rpm_key)
        pipe.expire(rpm_key, 60)
        results = await pipe.execute()

        tpm_used, _, rpm_used, _ = results
        limits = await self.get_tenant_limits(tenant_id)

        if tpm_used > limits["tpm"] or rpm_used > limits["rpm"]:
            return False
        return True

共享模型 vs 专用模型：成本与性能权衡

这是多租户 LLM 架构最核心的架构决策。

共享模型池设计

class SharedModelPool:
    """所有标准租户共享的模型推理池"""

    def __init__(self, model_name: str, gpu_count: int):
        self.model = model_name
        # vLLM continuous batching 引擎
        self.engine = AsyncLLMEngine.from_engine_args(
            model=model_name,
            tensor_parallel_size=gpu_count,
            max_num_seqs=256,          # 最大并发序列
            max_num_batched_tokens=8192,
            gpu_memory_utilization=0.90,
            enable_prefix_caching=True,  # 前缀缓存优化
        )
        self.tenant_metrics = defaultdict(lambda: {
            "tokens_in": 0, "tokens_out": 0, "latency_p99": []
        })

    async def inference(self, tenant_id: str, request: ChatRequest):
        # 注入租户标识用于计费和监控
        sampling_params = SamplingParams(
            temperature=request.temperature,
            max_tokens=request.max_tokens,
            # 禁止跨租户 prefix cache 共享
        )
        result = await self.engine.generate(
            request.to_prompt(),
            sampling_params,
            request_id=f"{tenant_id}-{uuid4()}"
        )
        self._record_metrics(tenant_id, result)
        return result

专用模型部署

对于企业级租户，提供独立模型实例：

维度共享模型专用模型
GPU 成本摊薄到所有租户租户独付
延迟受其他租户影响稳定可预测
隐私理论上有泄漏风险完全隔离
定制化仅 Prompt/LoRA可微调全模型
弹性自动扩缩容需预留资源
月费$99-$999$5,000-$50,000

配额管理与计费

Token 级精确计费

class TokenBillingSystem:
    def __init__(self, db_pool):
        self.db_pool = db_pool
        # 计费费率表（每百万 Token 的价格，单位：美元）
        self.pricing = {
            "gpt-4-turbo": {"input": 10.0, "output": 30.0},
            "llama-3-70b": {"input": 0.6, "output": 0.8},
            "claude-3-opus": {"input": 15.0, "output": 75.0},
        }

    async def record_usage(self, tenant_id, model, input_tokens, output_tokens):
        rate = self.pricing[model]
        cost = (input_tokens / 1_000_000 * rate["input"] +
                output_tokens / 1_000_000 * rate["output"])

        async with self.db_pool.acquire() as conn:
            await conn.execute("""
                INSERT INTO usage_records (tenant_id, model, input_tokens, output_tokens, cost, created_at)
                VALUES ($1, $2, $3, $4, $5, NOW())
            """, tenant_id, model, input_tokens, output_tokens, cost)

            # 扣减配额
            await conn.execute("""
                UPDATE tenant_quotas
                SET used_tokens = used_tokens + $2,
                    used_cost = used_cost + $3
                WHERE tenant_id = $1
            """, tenant_id, input_tokens + output_tokens, cost)

    async def check_quota(self, tenant_id, estimated_tokens):
        async with self.db_pool.acquire() as conn:
            quota = await conn.fetchrow(
                "SELECT * FROM tenant_quotas WHERE tenant_id = $1 FOR UPDATE",
                tenant_id
            )
            if quota["used_tokens"] + estimated_tokens > quota["max_tokens"]:
                raise QuotaExceededError(
                    f"Tenant {tenant_id} quota exceeded: "
                    f"used={quota['used_tokens']}, max={quota['max_tokens']}"
                )

分层配额策略

# tenant_quota_tiers.yaml
tiers:
  free:
    max_tokens_per_day: 100_000
    max_rpm: 10
    models_allowed: ["llama-3-8b"]
    features: ["basic_chat"]

  pro:
    max_tokens_per_day: 5_000_000
    max_rpm: 60
    models_allowed: ["llama-3-8b", "llama-3-70b"]
    features: ["basic_chat", "function_calling", "rag"]

  enterprise:
    max_tokens_per_day: 100_000_000
    max_rpm: 600
    models_allowed: ["*"]
    features: ["*"]
    dedicated_sla:
      uptime: 99.9
      p99_latency_ms: 2000

安全边界设计

Prompt 注入防御

多租户场景下，一个租户的恶意 Prompt 不能影响其他租户：

class PromptSanitizer:
    """多租户 Prompt 安全过滤器"""

    # 检测跨租户信息泄漏的模式
    LEAKAGE_PATTERNS = [
        r"ignore.*previous.*tenant",
        r"show.*other.*users?",
        r"system.*prompt.*for.*tenant",
        r"<tenant_id>.*</tenant_id>",
    ]

    def sanitize(self, tenant_id: str, messages: list[dict]) -> list[dict]:
        sanitized = []
        for msg in messages:
            content = msg["content"]
            # 1. 移除可能的租户标识泄漏
            content = re.sub(r'tenant[_-]?id[:\s]\w+', '[REDACTED]', content)
            # 2. 检测注入模式
            for pattern in self.LEAKAGE_PATTERNS:
                if re.search(pattern, content, re.IGNORECASE):
                    logger.warning(f"Potential prompt injection from tenant {tenant_id}")
                    content = "[FILTERED]"
                    break
            # 3. 限制单条消息长度
            if len(content) > 32_000:
                content = content[:32_000] + "...[truncated]"
            sanitized.append({**msg, "content": content})
        return sanitized

租户级审计日志

class TenantAuditLogger:
    async def log_inference(self, tenant_id, request, response, metadata):
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "tenant_id": tenant_id,
            "user_id": metadata["user_id"],
            "model": request["model"],
            "input_tokens": response["usage"]["prompt_tokens"],
            "output_tokens": response["usage"]["completion_tokens"],
            "latency_ms": metadata["latency_ms"],
            "request_hash": hashlib.sha256(
                json.dumps(request["messages"]).encode()
            ).hexdigest()[:16],
            # 不记录完整 Prompt 内容，仅记录 hash 用于审计追踪
        }
        await self.audit_queue.put(log_entry)

案例分析：某法律 AI SaaS 平台

背景： 服务 200+ 律所，每个律所是独立租户，数据高度敏感。

架构选择：

  • 基座模型共享（Llama-3-70B）
  • 每个律所独立 LoRA 适配器（法律领域微调）
  • 独立向量数据库（每个律所一个 Qdrant collection）
  • 租户级 GPU 时间片调度（保证 SLA）

关键指标：

  • 租户间数据泄漏测试：0 次（通过自动化红队验证）
  • 推理延迟 P99：< 1.2s（共享池）
  • 成本：比独立部署低 87%
  • 合规：通过 SOC2 Type II 审计

实战建议

  1. 从共享模型开始，当某个租户的 QPS 超过总流量的 30% 时，考虑迁移到专用实例
  2. Token 计费必须实时，异步计费会导致超额使用无法控制
  3. KV Cache 是性能关键，开启 prefix caching 可以降低 40%+ 的推理成本，但必须确保租户间不共享
  4. 安全边界要多层防御，不要只依赖 Prompt 过滤，还要在模型输出层做检测
  5. 监控粒度到租户，每个租户的 QPS、延迟、错误率、Token 使用量都要独立看板

加入讨论

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

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