Mem0 Platform 如何导出用户记忆用于审查与合规
【免费下载链接】embedchainThe Memory Layer for AI Agents - Drop-in memory infrastructure for AI agents and apps. Context that persists. Built for production.项目地址: https://gitcode.com/GitHub_Trending/em/embedchain
当 Mem0 Platform 上的应用积累了大量用户记忆后,你经常需要把某个用户(或某段时间)的记忆完整取出来,交给合规团队审查、用于数据迁移,或满足 GDPR 类的数据可携要求。Mem0 Platform 的MemoryClient提供两条互补的路径:get_all()直接拉取原始记忆记录用于审计,create_memory_export()按自定义 Pydantic schema 把记忆结构化成一份可下载的导出文件。本文以 Python SDK 为主路径,走通从准备 API key、取数、提交导出任务到下载结果的完整流程,并说明导出的有效期限制。
准备:API key 与客户端初始化
使用 Platform 的导出功能需要一个具备导出权限的 API key。文档明确提示:如果导出操作因认证错误失败,需要到项目的 dashboard 设置中确认该 key 拥有 export 权限。
from mem0 import MemoryClient client = MemoryClient(api_key="your-api-key")上面的your-api-key需要替换为你自己的 Platform API key。初始化成功后,可以先写入少量样例记忆用于演练(文档中的示例数据):
# Dev's work history client.add( "Dev works at TechCorp as a senior engineer", user_id="dev", metadata={"type": "professional"} ) # Arjun's preferences client.add( "Arjun prefers morning meetings and async communication", user_id="arjun", metadata={"type": "preference"} ) # Carl's project notes client.add( "Carl is leading the API redesign project, targeting Q2 launch", user_id="carl", metadata={"type": "project"} )用 get_all() 做全量拉取:审计原始记录
合规审查通常第一步是把某个用户的全部记忆原样取出。get_all()返回所有匹配过滤条件的记录:无语义检索、无排序,就是原始数据,适合导出和审计。
dev_memories = client.get_all( filters={"user_id": "dev"}, page=1, page_size=50 ) print(f"Total memories: {dev_memories['count']}") print(f"First memory: {dev_memories['results'][0]['memory']}")文档示例输出(仅用于演示,实际记录数取决于你的数据):
Total memories: 1 First memory: Dev works at TechCorp as a senior engineer返回体是分页信封,包含count、next、previous、results四个字段,通过page和page_size参数翻页。有几个过滤规则需要留意:
- 实体 ID(
user_id、agent_id、app_id、run_id)必须放在filters对象内部,放在顶层会返回 400。 - 已过期(
expiration_date已过)的记忆默认不出现在get_all()结果里;需要包含它们时传show_expired: true(TypeScript 中是showExpired)。 filters支持 AND、OR、NOT 逻辑运算和in、gte、lte、gt、lt、ne、icontains、*等比较操作符,可以按元数据和时间范围圈定审查对象,例如:
memories = client.get_all( filters={ "AND": [ {"user_id": "alex"}, {"created_at": {"gte": "2024-07-01", "lte": "2024-07-31"}} ] }, show_expired=False, page=1, page_size=50 )如果只是针对某个具体问题做语义检索,可以用search()替代,但批量审查仍以get_all()为准:
results = client.search( query="What does Dev do for work?", filters={"user_id": "dev"}, top_k=5 )用 create_memory_export() 生成结构化导出
当审查或迁移需要把记忆映射到固定字段(而不是逐条原文)时,使用结构化导出。整个流程分三步:定义 schema、提交导出任务、下载导出结果。
第 1 步:定义导出 schema
Schema 是标准 JSON schema 结构,描述导出文件里的字段。文档给出的职业档案示例:
professional_profile_schema = { "properties": { "full_name": { "type": "string", "description": "The person's full name" }, "current_role": { "type": "string", "description": "Current job title or role" }, "company": { "type": "string", "description": "Current employer" } }, "title": "ProfessionalProfile", "type": "object" }Memory Export 功能文档中还展示了更完整的写法:用$defs定义枚举(如EducationLevel、EmploymentStatus),用anyOf允许字段为null。如果希望缺失字段返回 null 而不是让模型猜测,可在 schema 中把字段设为可空。
第 2 步:提交导出任务
export_job = client.create_memory_export( schema=professional_profile_schema, filters={"user_id": "dev"} ) print(f"Export ID: {export_job['id']}") print(f"Message: {export_job['message']}")文档示例输出:
Export ID: 550e8400-e29b-41d4-a716-446655440000 Message: Memory export request received. The export will be ready in a few seconds.导出是异步任务,记忆量大或 schema 复杂时耗时更长。可用的过滤条件包括user_id、agent_id、app_id、run_id以及按created_at的时间范围。
可选地,通过export_instructions参数指导导出过程中冲突的解决方式和字段格式:
export_with_instructions = client.create_memory_export( schema=professional_profile_schema, filters={"user_id": "arjun"}, export_instructions=""" 1. Use the most recent information if there are conflicts 2. Only include confirmed facts, not speculation 3. Return null for missing fields rather than guessing """ )export_instructions是自由文本,文档示例用它约定"冲突取最新、不写推测、缺字段返回 null"这类规则;具体措辞由你按审查要求自行调整。
第 3 步:下载导出结果
拿到导出 ID 后轮询get_memory_export():
# Get by ID export_data = client.get_memory_export( memory_export_id=export_job['id'] ) print(export_data)任务仍在处理时,该调用会返回 404 和{"error": "No memory export request found"}。文档给出的判断方式:短间隔重试,直到调用成功,而不是去轮询某个状态字段。成功后返回按 schema 结构化的数据,文档示例结果:
{ "full_name": "Dev", "current_role": "senior engineer", "company": "TechCorp" }也可以用过滤条件直接取"最新一次匹配的导出",不必记住 ID:
# Get latest export matching filters export_by_filters = client.get_memory_export( filters={"user_id": "dev"} )通过 Platform UI 手动导出
一次性导出或人工审查时,也可以直接在 dashboard 操作:
- 进入项目 dashboard 的Memory Exports;
- 点击Create Export;
- 选择过滤条件和 schema;
- 导出完成后下载 JSON 文件。
这条路径和 SDK 路径使用同一套过滤与 schema 机制,区别只是无需写代码。
有效期限制与验证方式
- 导出数据 7 天后过期。文档以警告形式明确:导出过期后需要重新创建导出任务;如果要做长期归档,必须尽快把导出的 JSON 下载到本地保存。这是流程中最容易踩的坑:任务提交成功不等于可以慢慢取。
- 成功判断:
create_memory_export()返回包含id和message的对象,说明任务已被受理;get_memory_export()不再返回 404、并按你的 schema 返回字段,说明导出已完成可取。 - 认证失败:导出操作报认证错误时,检查 API key 是否具有 export 权限(见项目 dashboard 设置),而不是反复重试。
- 取数范围核对:审查前可用
get_all()先确认过滤条件命中的count,与导出范围一致后再提交导出任务。
SDK 调用对应的接口定义可参考 Create Memory Export(post /v1/exports/)与 Get Memory Export,get_all()的完整过滤与分页规则见 Get Memories。
适用边界
- 本文所有
MemoryClient用法仅适用于 Mem0 Platform,不适用于 Open Source 的本地Memory类。 - 结构化导出由 Mem0 按 schema 从记忆内容生成,文档强调"Only include confirmed facts, not speculation"应通过
export_instructions显式约定,而不是默认行为。 - 过期记忆默认不进入
get_all()列表;如果合规审查要求覆盖已过期记忆,记得传show_expired: true。
完整代码与输出示例见 Export Stored Memories cookbook 和 Memory Export 功能文档。
【免费下载链接】embedchainThe Memory Layer for AI Agents - Drop-in memory infrastructure for AI agents and apps. Context that persists. Built for production.项目地址: https://gitcode.com/GitHub_Trending/em/embedchain
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考