7.2. 多租户与命名空间是实现 SaaS 规模化记忆的前提

多租户与命名空间是实现 SaaS 规模化记忆的前提

2026 年 5 月，Reddit 的 r/LangChain 社区出现了一个获得大量讨论的帖子：《Moving LangChain to production: How we solve multi-tenancy, lazy loading》。作者在帖子中写下了一句直击要害的话：“仅依赖元数据过滤来分离租户数据在安全上不够可靠，应采用硬隔离的命名空间方案。”这条来自一线工程师的生产经验，恰好揭示了我们把 AI 智能体记忆推向 SaaS 规模化运营时必须跨越的第一道坎——在多租户环境下彻底杜绝记忆数据的交叉污染。

在上一章，我们已经将记忆系统独立部署为基础设施。现在，当同一个记忆平台需要同时服务几十个、甚至上百个企业客户时，如果只是简单地把所有租户的对话历史、向量嵌入和用户偏好塞进同一张表或同一个索引，单靠 WHERE tenant_id = ? 过滤，不仅会引入严重的安全隐患，还会让每次查询都扫过海量无关数据，查询延迟和存储成本都会指数级攀升。因此，我们需要一套从存储层、索引层到检索层都贯彻的多租户隔离架构，让每个租户就像在使用一套完全独立的记忆系统，而运维团队只需要管理一个统一的集群。

本章的目标，就是与你一起完成这项改造：设计一套基于命名空间的记忆隔离方案，为租户配备可动态调整的配额与限流能力，并在严格隔离的基础上，巧妙地注入一层可选的全局共享知识，提升新租户的冷启动体验。我们将以 Python 和 LangChain 生态为具体载体，但核心思路适用于任何 Agent 记忆平台。

你需要什么

• 运行环境：Python 3.10+，已安装 langchain、langchain_community、langchain_core，并有一个可用的向量数据库（如 FAISS、Chroma 或 Pinecone）及 Key-Value 存储（如 Redis 或本地内存）。
• 示例数据库：本文默认使用 FAISS 作为向量存储、Redis 作为计数与限流存储，但你可以根据需要替换。
• 工具：Postman 或 curl 用于验证 API，以及一个代码编辑器。
• 预计时间：约 40 分钟完成全部理解与实践。

最终成果

你将得到一个可运行的、具备三层隔离与配额管控的 Agent 记忆后端骨架：

• 每个租户拥有完全独立的持久化记忆文件、向量索引和对话历史。
• 可以随时通过简单配置为租户设定最大存储条目数或请求频率，防止单租户膨胀耗尽集群资源。
• 新建租户无需从零开始，可以共享一份全局通用知识库，但彼此完全无感。

这份骨架本身就是一套 SaaS 记忆产品的内核原型，后续只需替换持久化实现（如改为 PostgreSQL 或 DynamoDB），即可快速切换存储策略。

步骤说明

步骤 1：为记忆后端注入租户命名空间

LangChain 的 DeepAgents 和 Message Histories 都提供了原生的 namespace 概念。在官方文档的 Backends 一节中明确指出：“对于多用户部署，始终设置一个 namespace 工厂函数来隔离每个用户或租户的数据。” 这里的命名空间本质上是一个元组，例如 ("tenant_a", "user_123")，会被底层存储实现自动用作键前缀或索引分区。

我们先创建一个支持多租户的记忆存储实例。

# step_1_namespace.py
from langchain.storage import InMemoryStore  # 生产环境应换为 Redis 等持久化存储
from langchain_core.stores import BaseStore

# 模拟一个简单的带命名空间的存储封装
class TenantAwareStore:
    """为每个租户自动添加命名空间前缀的存储适配器"""

    def __init__(self, base_store: BaseStore):
        self._store = base_store

    def _ns_key(self, tenant_id: str, key: str) -> tuple:
        """生产环境建议使用更细粒度的 (tenant_id, user_id, assistant_id, key)"""
        return (f"tenant:{tenant_id}", key)

    def mset(self, tenant_id: str, data: dict):
        """写入数据时自动附加租户前缀"""
        ns_keys = [self._ns_key(tenant_id, k) for k in data.keys()]
        self._store.mset(list(zip(ns_keys, data.values())))

    def mget(self, tenant_id: str, keys: list[str]) -> list:
        ns_keys = [self._ns_key(tenant_id, k) for k in keys]
        return self._store.mget(ns_keys)

# 初始化存储
base_store = InMemoryStore()
store = TenantAwareStore(base_store)

# 写入租户A和租户B的同名记忆，验证隔离
store.mset("tenant_A", {"user_pref": "likes dark mode"})
store.mset("tenant_B", {"user_pref": "likes light mode"})

# 分别读取
print(store.mget("tenant_A", ["user_pref"]))  # 预期输出: [b'likes dark mode'] 或类似
print(store.mget("tenant_B", ["user_pref"]))  # 预期输出: [b'likes light mode']

注意：InMemoryStore 仅可用于测试。生产环境中必须替换为 Redis、PostgreSQL 等持久化存储，否则服务重启后所有记忆丢失，且无法水平扩展。另外，命名空间元组的长度和顺序必须与向量库、关系库中的分区规则保持一致，否则会出现“记忆存进去了但永远读不出来”的诡异现象。

如果使用 LangChain 官方的 Backend 设置，可以直接在配置阶段指定 namespace_factory：

from deepagents.backends import FilesystemBackend

def namespace_factory(tenant_id: str, user_id: str) -> tuple:
    return (tenant_id, user_id)

backend = FilesystemBackend(
    root_dir="./agent_memory",
    namespace_factory=namespace_factory
)

每当我们通过 backend.get(namespace=("tenant_A", "user_1"), ...) 访问文件时，底层的文件路径会自动组织为 ./agent_memory/tenant_A/user_1/...，实现文件系统级别的天然隔离。

预期结果：两个不同租户的数据完全独立存储和读取，即便使用相同的业务键（如 user_pref），也不会互相覆盖。

步骤 2：在向量索引上实现硬隔离

对于 AI 智能体而言，记忆不仅仅是键值对，更重要的是基于语义相似度的长期记忆检索。这一部分必须对每个租户建立独立的向量空间，而不是共用一个索引然后通过元数据过滤。前者是真正的“物理”隔离（不同索引文件或集合），后者仅是“逻辑”隔离。Reddit 分享的生产经验指出了依赖元数据过滤的安全风险：一旦过滤条件因为代码 Bug 或注入攻击而失效，A 租户就能读到 B 租户的私有记忆。

下面我们以 FAISS 为例，为每个租户动态创建或加载独立的向量索引。

# step_2_vector_isolation.py
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
import os

class TenantVectorStore:
    """管理每个租户的独立 FAISS 索引，索引路径在文件系统上用 tenant_id 命名空间隔离"""

    def __init__(self, index_root: str, embedding_model):
        self._root = index_root
        self._embeddings = embedding_model
        os.makedirs(self._root, exist_ok=True)

    def _index_path(self, tenant_id: str) -> str:
        return os.path.join(self._root, f"tenant_{tenant_id}.faiss")

    def get_store(self, tenant_id: str) -> FAISS:
        path = self._index_path(tenant_id)
        if os.path.exists(path):
            return FAISS.load_local(path, self._embeddings, allow_dangerous_deserialization=True)
        else:
            # 新建一个空索引
            return FAISS.from_texts(["placeholder"], self._embeddings)  # 后续可以移除占位文本

    def persist(self, tenant_id: str, store: FAISS):
        store.save_local(self._index_path(tenant_id))

# 使用示例
vs_manager = TenantVectorStore("./vector_indices", OpenAIEmbeddings())
store_a = vs_manager.get_store("tenant_A")
store_a.add_texts(["Tenant A private note: project deadline next Friday"])
vs_manager.persist("tenant_A", store_a)

store_b = vs_manager.get_store("tenant_B")  # 完全独立的索引，不会包含 A 的数据

如果使用托管向量数据库（Pinecone、Qdrant、Weaviate 等），应确保在创建集合（collection）或命名空间时将租户 ID 作为前缀或独立维度，例如在 Pinecone 中为每个租户创建独立的 namespace。

预期结果：租户 A 的语义搜索绝不会检索到租户 B 的记忆，且各租户的索引文件独立存储，便于按需备份、迁移或删除。

步骤 3：配置租户级配额与动态限流

隔离解决了安全问题，但没有解决资源争用。在 SaaS 模式下，你无法控制某个租户会不会突然灌入海量记忆，导致集群磁盘爆满或向量检索性能大幅下降。因此，我们需要在记忆写入和检索两条路径上加入配额检查和速率限制，并且这些限制必须可动态调整，无需重启服务。

这里我们通过 Redis 实现一个轻量级的配额管理器。Redis 的原子操作非常适合计数和 TTL 管理。

# step_3_quota_limiter.py
import redis
import time
from functools import wraps

class TenantQuotaManager:
    """使用 Redis 管理每个租户的存储条目数上限和请求速率"""

    def __init__(self, redis_client: redis.Redis, default_max_items: int = 10000,
                 default_rate_limit: int = 100):  # 每秒请求数
        self.redis = redis_client
        self.default_max_items = default_max_items
        self.default_rate_limit = default_rate_limit

    def _get_quota(self, tenant_id: str) -> dict:
        """从 Redis 中读取或初始化租户配额，支持动态修改"""
        key = f"quota:{tenant_id}"
        data = self.redis.hgetall(key)
        if not data:
            data = {"max_items": str(self.default_max_items),
                    "rate_limit": str(self.default_rate_limit)}
            self.redis.hset(key, mapping=data)
        return {k: int(v) for k, v in data.items()}

    def check_and_increment_items(self, tenant_id: str, delta: int = 1) -> bool:
        """检查存储条目数是否超限，并原子性增加计数"""
        quota = self._get_quota(tenant_id)
        count_key = f"item_count:{tenant_id}"
        current = int(self.redis.get(count_key) or 0)
        if current + delta > quota["max_items"]:
            return False
        self.redis.incrby(count_key, delta)
        return True

    def rate_limit_check(self, tenant_id: str) -> bool:
        """简单的滑动窗口限流：记录最近一秒内的请求数"""
        quota = self._get_quota(tenant_id)
        key = f"rate:{tenant_id}:{int(time.time())}"
        current = self.redis.incr(key)
        self.redis.expire(key, 2)  # 避免永久残留
        return current <= quota["rate_limit"]

# 在记忆写入 API 中使用装饰器
quota_manager = TenantQuotaManager(redis.Redis())

def require_quota(func):
    @wraps(func)
    def wrapper(tenant_id, *args, **kwargs):
        if not quota_manager.check_and_increment_items(tenant_id):
            raise Exception("Tenant item quota exceeded. Please upgrade your plan.")
        if not quota_manager.rate_limit_check(tenant_id):
            raise Exception("Rate limit hit. Please slow down.")
        return func(tenant_id, *args, **kwargs)
    return wrapper

@require_quota
def add_memory(tenant_id: str, memory_text: str):
    # 实际写入记忆的逻辑
    pass

预期结果：当你模拟某个租户疯狂写入时，一旦超过预设条目数或频率，请求会被明确拒绝并返回可读的异常消息。你可以通过修改 Redis 中的 quota:tenant_id 哈希，实时调整某个租户的上限，而无需重新部署代码。

踩坑经验：配额计数器必须与当前实际存储量保持松散一致。当记忆被删除时，务必减少计数器。最好的做法是让计数字段由存储层的真实条目数计算得出（例如通过 COUNT(*) 查询），而非仅依赖简单的 increment，避免因异常情况导致计数偏移。此外，限流窗口的粒度需要根据业务场景调整，秒级窗口比较粗放，生产环境建议使用令牌桶或漏桶算法，可以用 Redis 的 sorted set 实现更精确的控制。

步骤 4：构建可选的跨租户全局知识层

在实现严格隔离之后，一个现实的痛点是：当全新租户接入时，智能体没有任何先验知识，回答会显得单薄且容易出错，这种冷启动问题在垂直领域（如医疗法规、企业内部常识）尤为明显。为了解决它，我们需要在隔离体系之上设计一个只读的全局知识层。该层对所有租户开放查询，但只有系统管理员可以修改。本质上，它是独立于租户记忆的一个共享向量库，检索时可选择性地与租户私有记忆的结果合并。

+----------------+     +----------------+
|  Tenant A      |     |  Tenant B      |
|  Private Index |     |  Private Index |
+-------+--------+     +-------+--------+
        |                      |
        +----------+-----------+
                   |
          +--------v--------+
          |  Global Shared   |
          |  Knowledge Index |
          +-----------------+

我们在向量存储管理层增加一个全局获取逻辑：

# step_4_global_knowledge.py
class TenantWithGlobalStore:
    def __init__(self, tenant_vs: TenantVectorStore, global_index_path: str):
        self.tenant_vs = tenant_vs
        self.global_store = FAISS.load_local(
            global_index_path, OpenAIEmbeddings(), allow_dangerous_deserialization=True
        )

    def hybrid_search(self, tenant_id: str, query: str, k: int = 4, global_k: int = 2):
        # 并行检索私有记忆和全局知识
        private_store = self.tenant_vs.get_store(tenant_id)
        private_results = private_store.similarity_search_with_score(query, k=k)
        global_results = self.global_store.similarity_search_with_score(query, k=global_k)

        # 合并并重新排序（简单示例：按相似度分数排序，生产环境可加入加权策略）
        combined = private_results + global_results
        combined.sort(key=lambda x: x[1], reverse=True)
        return combined

全局索引的构建和维护由管理员专职完成，例如通过定期更新企业通用文档、法律法规等内容。检索时，我们优先展示私有记忆，然后在底部或混合结果中注入全局信息。为了避免全局知识“喧宾夺主”，通常定义较低的 global_k 值，并可以标记 source=global，让上游智能体决定是否采用。

预期结果：新租户在没有积累任何私有记忆时，仍然能够从全局知识中获得合理的参考信息。一旦该租户逐渐积累起自己的记忆，私有结果会自然占据主导。

注意：全局索引一旦构建，通常对租户而言是只读的。任何租户的对话内容都绝不写入全局索引，否则会立刻引发数据泄露事件。如果你希望将某个租户的特定知识（经其授权）提炼到全局，必须走单独的审核流程，由系统管理员从完全独立的管道注入。

回顾

在这一章，我们从零开始为 AI Agent 的记忆基础设施筑起了多租户的防火墙：

1. 命名空间隔离：在键值存储和文件系统中引入基于租户 ID 的命名空间，保证数据硬隔离。
2. 向量索引隔离：为每个租户创建独立的向量空间，杜绝元数据过滤带来的安全隐患。
3. 配额与限流：利用 Redis 实现了存储条目数和请求速率的动态控制，防止资源被单租户吞噬。
4. 全局共享知识：在隔离之上构建了可选的只读知识层，解决新租户冷启动难题。

如果你一步步跟进并执行了上面的代码，你已经将单一的“单机记忆库”升级为一个可以安全服务多个客户的 SaaS 记忆内核。整个操作耗时约 40 分钟，核心改动集中在存储适配层的封装，对于上层 Agent 逻辑几乎无侵入。

但到目前为止，我们一直回避了一个关键问题：实际存储介质该如何选型？Redis 的纯内存特性虽然快，但成本高昂且数据量受限；FAISS 的本地文件索引在分布式下需要自己管理分片；PostgreSQL 提供了事务和持久化，但向量检索需要额外索引；DynamoDB 和 MongoDB 则各有开销与运维复杂度。下一章，我们将直面 持久化策略选型直接影响成本与可靠性，对比 PostgreSQL、Redis、DynamoDB 和向量数据库在记忆存储中的适用性，帮助你做出与业务规模和预算匹配的存储决策。

行动清单

• 立刻检查你的向量库配置：确保每个客户使用独立的 collection 或 namespace，不要仅靠元数据过滤区分租户。
• 为记忆写入路径增加配额检查：先用 Redis 或内存计数器做一个最简单的条目数上限，防止意外膨胀。
• 实现一个轻量限流中间件：基于租户 ID 对记忆写入接口做速率限制，避免单点风暴。
• 准备一份全局知识种子数据：哪怕只是一个 FAQ 文档，将它导入一个只读共享索引，验证冷启动改善效果。
• 列出你的持久化需求清单：包括数据量级、读写 QPS、延迟要求、成本预算，为下一章的存储选型做准备。