Mem0 Python SDK 完全参考：MemoryClient 平台客户端与 Memory 开源客户端的 API 详解

本文是 mem0ai Python 包的完整 API 参考指南，基于 mem0-plugin 技能文档 python.md 并结合仓库源码逐节展开。读完本文，你将掌握两类入口（托管平台客户端 MemoryClient 与自托管开源客户端 Memory）的全部方法签名、参数默认值与返回值结构，理解语义检索的过滤算子体系，并能对照 mem0/client/main.py 与 mem0/memory/main.py 的源码验证文档描述与实际行为的一致性。

两个入口：同一个包里的两套客户端

mem0ai 包同时提供托管 API 客户端与本地自托管内存库。从 mem0/init.py 的导出可以看出顶层入口只有四个类：

from mem0.client.main import AsyncMemoryClient, MemoryClient
from mem0.memory.main import AsyncMemory, Memory

• Platform（托管）：MemoryClient（同步）/ AsyncMemoryClient（异步），所有操作走 https://api.mem0.ai 的 HTTP 调用；
• OSS（自托管）：Memory（同步）/ AsyncMemory（异步），本地运行 LLM、Embedding 与向量库。

注意 import 区分：from mem0 import Memory 是开源客户端，from mem0 import MemoryClient 是平台客户端，两者不可混用。

Platform 客户端：安装与构造行为

安装

pip install mem0ai
export MEM0_API_KEY="m0-your-api-key"

MemoryClient（同步）

from mem0 import MemoryClient

client = MemoryClient(api_key="m0-xxx")

MemoryClient(api_key=None) 的构造逻辑在 mem0/client/main.py 中可以逐行印证：

• 未传 api_key 时回退读取 MEM0_API_KEY 环境变量；两者都缺失则抛出 ValueError("Mem0 API Key not provided...")；
• HTTP 库为 httpx，默认 timeout=300 秒（源码中 httpx.Client(..., timeout=300)）；
• Base URL 默认 https://api.mem0.ai，可通过 host 参数覆盖（也支持注入自定义 httpx.Client 实例）；
• 请求头携带 Authorization: Token <api_key> 与 Mem0-User-ID（API key 的 MD5 摘要）。

AsyncMemoryClient（异步）

from mem0 import AsyncMemoryClient

client = AsyncMemoryClient(api_key="m0-xxx")

# Or use as context manager
async with AsyncMemoryClient(api_key="m0-xxx") as client:
    results = await client.search("query", filters={"user_id": "alice"})

AsyncMemoryClient 的方法与 MemoryClient 一一对应，全部为 async/await。异步上下文管理器在源码中确认存在：aenter/aexit 会在退出时调用 async_client.aclose() 释放连接池。

Memory 核心方法（Platform）

以下方法同步存在于 MemoryClient 与 AsyncMemoryClient（异步版为 async def，见 async 方法定义）。

add(messages, **kwargs)

从消息中抽取并存储新记忆：

messages = [
    {"role": "user", "content": "I'm a vegetarian and allergic to nuts."},
    {"role": "assistant", "content": "Got it! I'll remember that."}
]
client.add(messages, user_id="alice")

参数	类型	默认值	说明
messages	str | dict | list[dict]	必填	消息内容；字符串会被自动转换为 user 消息
user_id	str	None	用户标识
agent_id	str	None	Agent 标识
app_id	str	None	应用标识
run_id	str	None	会话/运行标识
metadata	dict	None	自定义键值对
infer	bool	True	为 False 时跳过 LLM 抽取，直接存储原始文本
custom_categories	list	None	覆盖项目级分类定义
custom_instructions	str	None	覆盖抽取提示词
timestamp	int | float | str	None	自定义时间戳（Unix epoch 或 ISO 8601）

返回： dict——事件列表，形如 [{"id": "...", "event": "ADD", "data": {"memory": "..."}}]。源码中实体参数通过 ENTITY_PARAMS = frozenset({"user_id", "agent_id", "app_id", "run_id"})（mem0/client/main.py）统一识别并以 kwargs 形式注入请求载荷。

search(query, **kwargs)

按语义相似度检索记忆：

results = client.search("dietary preferences", filters={"user_id": "alice"})
for mem in results.get("results", []):
    print(mem["memory"], mem["score"])

参数	类型	默认值	说明
query	str	必填	自然语言查询
filters	dict	None	过滤对象：实体 ID 与/或 AND/OR/NOT 条件组合
top_k	int	10	返回条数
rerank	bool	False	启用深度语义重排（增加约 150–200ms 延迟）
threshold	float	0.1	最低相似度分数
fields	list	None	指定返回字段
categories	list	None	按分类过滤

返回： dict——{"results": [{id, memory, user_id, categories, score, created_at, ...}]}。查询字符串会经过 _validate_and_trim_search_query 校验：非字符串或纯空白直接抛 ValueError。

get(memory_id)

按 ID 取单条记忆，返回完整 memory 对象：

memory = client.get(memory_id="ea925981-...")

get_all(**kwargs)

带过滤地列出全部记忆，至少需要一个实体标识：

memories = client.get_all(filters={"user_id": "alice"})
# 复合过滤
memories = client.get_all(filters={"AND": [{"user_id": "alice"}, {"categories": {"contains": "health"}}]})

参数	类型	默认值	说明
filters	dict	None	实体 ID 与/或 AND/OR/NOT 条件组合
top_k	int	None	限制返回条数
page	int	None	页码
page_size	int	None	每页条数

返回： dict——{"results": [...]}。

update(memory_id, text=None, metadata=None, timestamp=None)

更新记忆内容、元数据或时间戳，至少提供一个参数：

client.update("ea925981-...", text="Updated: vegan since 2024")
client.update("ea925981-...", metadata={"verified": True})

delete(memory_id)

永久删除单条记忆：

client.delete("ea925981-...")

delete_all(**kwargs)

删除所有匹配过滤条件的记忆，不可恢复：

client.delete_all(user_id="alice")

history(memory_id)

查询记忆的变更历史：

history = client.history("ea925981-...")
# Returns: [{previous_value, new_value, action, timestamps}]

批量方法

方法	说明
batch_update(memories)	单请求最多更新 1000 条记忆
batch_delete(memories)	单请求最多删除 1000 条记忆

client.batch_update([
    {"memory_id": "uuid-1", "text": "Updated text"},
    {"memory_id": "uuid-2", "text": "Another update", "metadata": {"verified": True}},
])

client.batch_delete([
    {"memory_id": "uuid-1"},
    {"memory_id": "uuid-2"},
])

两个方法在 mem0/client/main.py 中均有对应实现。批量操作仅 Platform 可用，OSS 客户端不提供。

用户 / 实体管理

# 列出所有拥有记忆的用户、agent 与 session
users = client.users()
# Returns: {"results": [{"type": "user", "name": "alice"}, ...]}

# 删除指定实体及其全部记忆
client.delete_users(user_id="alice")

# 重置整个项目：删除所有用户、agent、session 与记忆
client.reset()

delete_users(user_id=None, agent_id=None, app_id=None, run_id=None) 支持按任一实体维度清理。reset() 是破坏性操作，会清空项目内全部数据，生产环境请谨慎调用。

导出与摘要

create_memory_export(schema, **kwargs)

按 JSON Schema 定义的结构导出记忆：

import json

schema = json.dumps({
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "preferences": {"type": "array", "items": {"type": "string"}},
    }
})
export = client.create_memory_export(schema=schema, user_id="alice")

get_memory_export(**kwargs)

取回之前创建的导出结果：

result = client.get_memory_export(memory_export_id=export["id"])

get_summary(filters=None)

获取记忆的摘要视图：

summary = client.get_summary(filters={"user_id": "alice"})

Feedback

对单条记忆提交质量反馈，用于闭环优化抽取质量：

client.feedback(
    memory_id="mem-123",
    feedback="POSITIVE",  # POSITIVE | NEGATIVE | VERY_NEGATIVE | None (清除反馈)
    feedback_reason="Accurately captured preference"
)

Webhooks

Webhook 的完整 CRUD（源码实现）：

# 列表
webhooks = client.get_webhooks(project_id="proj_123")

# 创建
webhook = client.create_webhook(
    url="https://your-app.com/webhook",
    name="Memory Logger",
    project_id="proj_123",
    event_types=["memory_add", "memory_update"]
)

# 更新
client.update_webhook(webhook_id=123, name="Updated", url="https://new-url.com")

# 删除
client.delete_webhook(webhook_id=123)

项目管理

项目级设置通过 client.project.* 访问（mem0/client/project.py 实现）：

# 读取项目配置
config = client.project.get(fields=["custom_categories", "custom_instructions"])

# 更新项目设置
client.project.update(
    custom_instructions="Extract dietary preferences and health info",
    custom_categories=[{"health": "Medical and dietary info"}],
    multilingual=True,
)

# 创建 / 删除项目
client.project.create(name="My Project", description="...")
client.project.delete()

# 成员管理
members = client.project.get_members()
client.project.add_member(email="[e-mail]", role="READER")  # READER 或 OWNER
client.project.update_member(email="[e-mail]", role="OWNER")
client.project.remove_member(email="[e-mail]")

自定义抽取提示（custom_instructions）与分类（custom_categories）是平台侧控制记忆抽取行为的两个核心旋钮。

Open Source / 自托管客户端

安装与构造

pip install mem0ai

from mem0 import Memory

m = Memory()  # 使用默认配置（OpenAI embedder + 内存型向量存储）

配置

Memory.from_config(config_dict) 会用 Pydantic 的 MemoryConfig 校验配置，非法配置直接抛出 ValidationError（源码）：

config = {
    "llm": {
        "provider": "openai",        # openai, groq, azure, ollama, lmstudio, google, anthropic, mistral
        "config": {
            "model": "gpt-5-mini",
            "api_key": "sk-xxx",
        }
    },
    "embedder": {
        "provider": "openai",        # openai, ollama, azure, lmstudio, google, huggingface
        "config": {
            "model": "text-embedding-3-small",
            "api_key": "sk-xxx",
        }
    },
    "vector_store": {
        "provider": "qdrant",        # faiss, qdrant, pgvector, redis, supabase, azure_ai_search, memory
        "config": {
            "collection_name": "my_memories",
            "host": "localhost",
            "port": 6333,
        }
    },
    "history_db_path": "history.db",              # 变更历史的 SQLite 路径
    "custom_instructions": "...",                  # 自定义抽取提示词
}

m = Memory.from_config(config)

仓库中 mem0/configs/llms/、mem0/configs/embeddings/ 与 mem0/configs/vector_stores/ 三个目录分别列出了当前支持的全部 LLM、Embedding 与向量库 provider 配置类，实际支持范围比上文注释更全（向量库达二十余种，如 mem0/vector_stores/ 中的 chroma、milvus、pgvector、weaviate 等）。

核心方法（OSS）

OSS 客户端的方法与 Platform 客户端同名，但全部在本地执行。

add(messages, *, user_id, agent_id, run_id, metadata, infer=True)

m.add("I'm a vegetarian", user_id="alice")
m.add([
    {"role": "user", "content": "I like hiking"},
    {"role": "assistant", "content": "Great outdoor activity!"}
], user_id="alice")

user_id、agent_id、run_id 三者至少传一个，否则抛错（add 实现）。当 agent_id 存在且消息中包含 assistant 角色时，源码会切换为 agent 记忆抽取模式（_should_use_agent_memory_extraction）。

返回： {"results": [...], "relations": [...]}。

源码中 add() 还额外暴露 expiration_date（过期记忆会从 search/get_all 中隐藏，除非 show_expired=True）、memory_type（"procedural_memory" 创建程序性记忆）、prompt 等参数。另外 timestamp 参数在 OSS 端不被支持，传入会直接抛 ValueError（源码）——这是与 Platform 端的一个关键差异。

search(query, *, filters=None, top_k=20, threshold=0.1, rerank=False)

results = m.search("dietary preferences", filters={"user_id": "alice"}, top_k=5)

实体 ID 必须放在 filters 字典内：search()/get_all() 会以 ValueError 拒绝顶层 user_id/agent_id/run_id 参数（_reject_top_level_entity_params）。这与 add() 接受顶层实体参数形成对比，是最常见的迁移踩坑点。

过滤算子在 search 文档字符串与实现 中完整列出：

算子	语义	示例
eq / ne	等于 / 不等于	{"key": {"ne": "x"}}
in / nin	在列表中 / 不在列表中	{"key": {"in": ["a", "b"]}}
gt / gte / lt / lte	数值比较	{"count": {"gte": 10}}
contains / icontains	包含 / 忽略大小写包含	{"key": {"contains": "text"}}
* 通配	匹配任意值	{"key": "*"}
AND / OR / NOT	逻辑组合	{"AND": [f1, f2]}

默认值与技能文档一致且可源码印证：top_k=20、threshold=0.1、rerank=False（search 签名）。

其余方法

get(memory_id) / get_all(**kwargs) / update(memory_id, data, metadata=None) / delete(memory_id) / delete_all(**kwargs) / history(memory_id) 与 Platform 客户端接口对齐（实现位置）。

reset()

清空向量库 collection 与历史数据库，并重建向量存储：

m.reset()

reset 源码 的执行顺序：重置并关闭 SQLite 历史库 → 重建 SQLiteManager → 通过 VectorStoreFactory.reset()（或降级为 delete_col() + 重新 create）重建向量库 → 若存在实体存储则一并重置。

close() 与资源释放

m.close()  # 释放 SQLite 连接

close 实现 关闭 db 连接并置空。从当前源码结构看，Memory/AsyncMemory 类并未定义 __enter__/__exit__（技能文档中 with Memory(config) as m: 的上下文管理器写法未能从源码中确认），因此在不依赖上下文管理器的场景下，建议在流程结束时显式调用 m.close()，避免 SQLite 连接泄漏。

AsyncMemory

from mem0 import AsyncMemory

m = AsyncMemory(config)
await m.add("text", user_id="alice")
results = await m.search("query", filters={"user_id": "alice"})

Platform vs OSS：关键差异对照

维度	Platform（MemoryClient）	OSS（Memory）
Import	from mem0 import MemoryClient	from mem0 import Memory
鉴权	需要 API key（MEM0_API_KEY）	无需 API key，基于配置
执行位置	HTTP 调用 api.mem0.ai	本地执行
基础设施	全托管	自行管理向量库、Embedding、LLM
实体过滤	filters={"user_id": "..."}	filters={"user_id": "..."}
批量操作	batch_update、batch_delete	不可用
Webhooks	完整 CRUD	不可用
导出	create_memory_export、get_memory_export	不可用
Feedback	feedback()	不可用
项目管理	client.project.*	不可用
用户列表	users()、delete_users()	不可用
自定义提示词	通过项目设置	直接配置 custom_instructions
历史存储	平台托管	本地 SQLite（路径可配）
异步	AsyncMemoryClient	AsyncMemory

v2 兼容性说明

如果你正从 SDK v2.x / v2 API 迁移到 v3：

API 变化：

• search/get_all 的实体 ID 位置：v2 传顶层 kwargs，v3 必须放进 filters：

# v2
results = client.search("query", user_id="alice")
# v3
results = client.search("query", filters={"user_id": "alice"})

• add() 返回值：v2 返回 ADD、UPDATE、DELETE 三类事件；v3 只返回 ADD。

默认值变化：

参数	v2	v3
top_k	100	20
threshold	None	0.1
rerank	True	False

已移除的参数：

• 构造函数：org_id、project_id
• add()：async_mode、output_format、enable_graph、immutable、expiration_date、filter_memories、batch_size、force_add_only、includes、excludes、keyword_search
• search() / get_all()：enable_graph
• 配置：enable_graph、graph_store、custom_fact_extraction_prompt（重命名为 custom_instructions）

完整的迁移说明可参考仓库内的 OSS v2 到 v3 迁移指南。

小结

• 选客户端看部署形态：要托管、要批量/导出/Webhook/项目级治理，用 MemoryClient；要数据完全留在本地、自选向量库与模型，用 Memory。
• 实体 ID 的位置是最大差异点：add() 顶层传，search()/get_all() 必须进 filters，OSS 端对顶层传参会直接抛 ValueError。
• 默认值决定召回行为：OSS 端 top_k=20、threshold=0.1、rerank=False，从源码签名可直接验证；需要更严的召回质量时再显式打开 rerank。
• 破坏性操作集中：delete_all、reset、delete_users、client.reset() 都会造成不可逆数据丢失，建议先 history()/get_all() 留档再执行。