---
id: scenario.multi-tenant-saas
title: 多租户 SaaS：数据隔离方案、租户上下文、配额与限流、AI 成本分摊和租户级配置
type: scenario
domain: cross
stack: [postgresql, vector-db, redis, fastapi, gin, llm-serving, k8s]
tags: [多租户, 数据隔离, 行级安全, 租户上下文, 配额, 限流, 成本分摊, 噪声邻居]
level: advanced
prerequisites: [backend.postgresql.index, ai.vector-db.index]
est_hours: 3
related: [ai.vector-db.scenario.multi-tenant-index, ai.rag.scenario.permission-aware-kb, ai.llm-serving.scenario.multi-provider-gateway]
week: [6, 13, 14]
---

# 多租户 SaaS：数据隔离方案、租户上下文、配额与限流、AI 成本分摊和租户级配置

## 业务背景与问题

AI 客服做成了 SaaS，服务几百家企业客户：

1. 一个开发者漏写了租户条件，A 公司看到了 B 公司的工单（严重事故）；
2. 一个大客户批量导入文档，向量化任务占满资源，其他客户的问答变慢（噪声邻居）；
3. 模型费用无法按客户核算，定价没有依据；
4. 大客户要求数据物理隔离、使用自己的模型密钥。

## 约束

- 大多数客户规模小，单独部署成本太高。
- 少数大客户愿意为隔离付费。

## 方案设计

1. **隔离分级**：
   | 级别 | 方式 | 适合 |
   | --- | --- | --- |
   | 共享表 + `tenant_id` + 行级安全 | 成本最低 | 大多数客户 |
   | 每租户独立 schema / 向量集合 | 中等 | 中型客户 |
   | 独立数据库 / 独立部署 | 最高 | 有合规要求的大客户 |
   通过“租户 → 数据源”的路由表决定访问哪里，代码对上层透明。
2. **租户上下文**：网关从令牌中解析租户，放入请求上下文；数据访问层自动注入（ORM 全局过滤 + PostgreSQL RLS 兜底）；缓存 key、向量检索 filter、对象存储路径、日志都带租户 id。见[多租户向量索引](../topics/ai/vector-db/scenarios/multi-tenant-index.md)。
3. **防噪声邻居**：按租户限流（Redis）；后台任务队列按租户公平调度（每个租户的并发上限）；大批量导入走单独队列。
4. **配额与计费**：每次模型调用记录租户、模型、输入 / 输出 token、成本；按套餐设置月度额度，超额降级到小模型或拒绝；报表给销售和财务。见[多模型网关](../topics/ai/llm-serving/scenarios/multi-provider-gateway.md)。
5. **租户级配置**：提示词、模型选择、品牌样式、功能开关按租户配置；支持客户自带模型密钥（加密存储，按租户使用）。
6. **测试**：跨租户访问的自动化测试（用 A 的令牌访问 B 的每类资源，必须 404 / 403）放进 CI。

## 关键代码

租户上下文 + RLS（FastAPI + PostgreSQL，示意）：

```python
tenant_ctx: ContextVar[str] = ContextVar("tenant_id")

@app.middleware("http")
async def tenant_middleware(request: Request, call_next):
    claims = verify_jwt(request.headers.get("authorization"))
    token = tenant_ctx.set(claims["tenant_id"])
    try:
        return await call_next(request)
    finally:
        tenant_ctx.reset(token)

@contextmanager
def tenant_conn():
    with pool.connection() as conn, conn.transaction():
        conn.execute("SELECT set_config('app.tenant_id', %s, true)", (tenant_ctx.get(),))   # RLS 策略读取它
        yield conn

def cache_key(*parts: str) -> str:
    return ":".join(["t", tenant_ctx.get(), *parts])      # 缓存 key 带租户，防止串数据
```

跨租户访问测试：

```python
@pytest.mark.parametrize("path", ["/tickets/{id}", "/docs/{id}", "/conversations/{id}"])
def test_cross_tenant_access_denied(path, tenant_a_token, tenant_b_resources):
    r = client.get(path.format(id=tenant_b_resources[path]), headers=auth(tenant_a_token))
    assert r.status_code in (403, 404)
```

## 取舍

| 决策 | 选择 | 原因 |
| --- | --- | --- |
| 隔离方式 | 分级（共享为主，大客户独立） | 平衡成本和合规 |
| 租户过滤 | 应用层自动注入 + 数据库 RLS 兜底 | 一层失误还有另一层 |
| 资源公平 | 租户级限流和任务并发上限 | 防噪声邻居 |
| 计费 | 按 token 记录到租户 | 定价有依据 |

## 易错点

- 只依赖开发者记得写 `WHERE tenant_id = ?`。
- 缓存 key、向量检索、文件路径漏了租户维度。
- 异步任务中丢失租户上下文（ContextVar 不会自动跨进程 / 跨消息传递，需要在消息中显式携带）。
- 管理后台的“超级管理员”接口没有审计。

## 面试怎么讲

> “多租户我做了分级隔离：大多数客户共享表加 tenant_id，数据库层用行级安全兜底；有合规要求的大客户独立库，通过路由表透明切换。租户从令牌解析放进上下文，数据访问层自动注入，缓存 key、向量检索、文件路径都带租户。防噪声邻居靠租户级限流和任务并发上限；每次模型调用按租户记 token 和成本，支撑套餐定价。CI 里有跨租户访问测试。”