05-Mem0 的三层记忆作用域实现
Mem0 的三层记忆作用域实现
学习目标
本章分析 Mem0 如何通过 user_id、agent_id、run_id 三层标识符实现记忆的作用域隔离。你将了解:
_build_filters_and_metadata的双通道构建器实现- 实体 ID 的验证与规范化规则
- Actor 过滤的实现逻辑与应用场景
- 作用域隔离在多租户场景中的重要性
前置知识
本章涉及多实体作用域的通用原理,建议先阅读:
下文假设你已理解作用域隔离的基本概念,直接聚焦 Mem0 的具体实现。
项目实践
双通道构建器
_build_filters_and_metadata() 是作用域隔离的核心函数,同时构建两个输出:
# 伪代码base_metadata_template, effective_query_filters = _build_filters_and_metadata( user_id="alice", agent_id="assistant-v1", run_id="run-123", input_metadata={"source": "web"}, input_filters={"topic": "programming"},)输出 1 - 存储元数据模板(用于写入新记忆):
{ "user_id": "alice", "agent_id": "assistant-v1", "run_id": "run-123", "source": "web", # 来自 input_metadata}输出 2 - 查询过滤器(用于检索已有记忆):
{ "user_id": "alice", "agent_id": "assistant-v1", "run_id": "run-123", "topic": "programming", # 来自 input_filters "actor_id": "John", # 可选,来自 actor_id 参数}实体 ID 验证
def _validate_and_trim_entity_id(value, name): trimmed = value.strip() if trimmed == "": raise ValueError(f"Invalid {name}: cannot be empty") if any(c.isspace() for c in trimmed): raise ValueError(f"Invalid {name}: cannot contain whitespace") return trimmed三层标识符都必须:
- 非空(去除首尾空白后)
- 不包含内部空白(
"user 123"被拒绝)
为什么禁止内部空白:如果允许 "user 123" 作为标识符,向量存储的过滤查询可能将空格解析为特殊字符,导致过滤异常。
强制至少一个标识符
if not session_ids_provided: raise Mem0ValidationError( message="At least one of 'user_id', 'agent_id', or 'run_id' must be provided.", error_code="VALIDATION_001", )Mem0 v3 使用自定义异常类 Mem0ValidationError,包含 error_code 和 details 字段,便于调用方精确处理。
Actor 过滤
actor_id 是一个额外的过滤维度,用于区分同一会话中的不同发言者:
# 解析优先级:显式 actor_id 参数 → filters["actor_id"]resolved_actor_id = actor_id or effective_query_filters.get("actor_id")if resolved_actor_id: effective_query_filters["actor_id"] = resolved_actor_idactor_id 不加入存储元数据模板——因为 actor 通常从消息内容中派生(如 message.get("name")),而非预先指定。
应用场景:多 speaker 对话中,消息的 role 都是 “user” 或 “assistant”,但实际发言者不同:
messages = [ {"role": "user", "content": "I have a dog named Max", "name": "John"}, {"role": "assistant", "content": "I just got a cat named Bailey", "name": "Maria"},]Mem0 在 _add_to_vector_store() 的 infer=False 模式下,将 name 字段提取为 actor_id 存入元数据:
actor_name = message_dict.get("name")if actor_name: per_msg_meta["actor_id"] = actor_name敏感字段保护
_safe_deepcopy_config() 在深拷贝配置时,对敏感字段进行红action:
_SENSITIVE_FIELDS_EXACT = frozenset({ "api_key", "secret_key", "private_key", "password", "token", "client_secret", ...})
_SENSITIVE_SUFFIXES = ("_password", "_secret", "_token", "_credential")运行时字段(http_auth、ssl_context)通过 allowlist 保护,不会被红action。
问题与规避
| 问题 | 对策 |
|---|---|
| 未提供标识符导致全局泄漏 | Mem0ValidationError 强制拦截 |
| 标识符含空格导致过滤异常 | _validate_and_trim_entity_id 拒绝含空格的 ID |
| 深拷贝非序列化对象失败 | _safe_deepcopy_config 回退到 dict 克隆 |
| 敏感字段泄漏到日志 | 深拷贝时红action _SENSITIVE_FIELDS_EXACT 中的字段 |
设计取舍
为什么用 filters 字典而非顶层参数?
优势:
- 灵活扩展:新增过滤条件不需要修改函数签名
- 支持嵌套条件:
{"AND": [...]}、{"OR": [...]} - 统一接口:所有 API 方法(add、search、get_all、delete_all)使用相同的 filters 格式
代价:
- 新手需要学习 filters 字典的构造规则
- IDE 自动补全不如顶层参数友好
为什么禁止标识符含空格?
如果允许 "user 123" 作为标识符,在向量存储的过滤查询中,空格可能被解析为分隔符或需要转义。禁止空格避免了这种歧义,同时保证了标识符的 URL-safe 特性(可在 API 路径中直接使用)。
参考来源
- Mem0 论文:arXiv:2504.19413
- Mem0 源码:
mem0/memory/main.py