11.5 · 批量操作（Batch Operations）

批量操作

批量操作

相关源文件

本章引用的主要源码文件：

• mem0-ts/jest.integration.config.js
• mem0-ts/src/client/tests/integration/batch.test.ts
• mem0-ts/src/client/tests/integration/crud.test.ts
• mem0-ts/src/client/tests/integration/global-setup.ts
• mem0-ts/src/client/tests/integration/global-teardown.ts
• mem0-ts/src/client/tests/integration/helpers.ts
• mem0-ts/src/client/tests/integration/initialization.test.ts
• mem0-ts/src/client/tests/integration/management.test.ts
• mem0-ts/src/client/tests/integration/search.test.ts
• mem0-ts/src/client/tests/memoryClient.crud.test.ts
• mem0-ts/src/client/tests/memoryClient.project.test.ts
• mem0-ts/src/client/tests/memoryClient.search.test.ts
• mem0/__init__.py
• mem0/configs/prompts.py
• mem0/memory/main.py
• mem0/memory/storage.py
• mem0/memory/utils.py
• tests/configs/test_prompts.py
• tests/memory/test_main.py
• tests/test_chatty_llm_parsing.py
• tests/test_main.py
• tests/test_memory.py
• tests/test_proxy.py

批量操作提供了高效的机制，可以在单个 API 请求中对多个记忆执行批量更新和删除。这些操作仅通过 MemoryClient 和 AsyncMemoryClient 类在 Mem0 平台 API 中可用，与逐个执行操作相比，可以显著减少网络开销。

关于单个记忆操作，请参阅记忆操作。关于平台 API 的常规用法，请参阅平台与 API。

概述

批量操作可以让你：

• 更新多个记忆：在单个请求中更新多个记忆的文本或元数据。
• 删除多个记忆：通过 ID 删除多个记忆，无需迭代开销。
• 减少 API 调用：将 N 个独立请求减少为单个批量请求。
• 提升性能：通过最小化网络往返次数来提高性能。

这两种操作都基于平台 API，并且需要有效的现有记忆 ID。

来源： mem0/client/main.py:515-565，mem0/client/main.py:1086-1138

架构

请求流程

下图展示了 SDK 与平台 API 之间批量操作的数据流。

图：批量操作序列

来源： mem0/client/main.py:515-565，mem0/memory/storage.py:193-215

客户端组件

MemoryClient 将这些操作实现为高级方法，封装了底层的 HTTP 逻辑。

图：SDK 批量组件关系

来源： mem0/client/main.py:515-565，mem0/client/main.py:136-160

批量更新

batch_update 方法可以在单个 API 调用中更新多个记忆的文本内容或元数据。

方法签名

# 同步版本
def batch_update(self, memories: List[Dict[str, Any]]) -> Dict[str, Any]

# 异步版本
async def batch_update(self, memories: List[Dict[str, Any]]) -> Dict[str, Any]

参数：

记忆字典结构：

返回： Dict[str, Any]，包含服务器返回的更新结果。

来源： mem0/client/main.py:515-539，mem0/client/main.py:1086-1111

实现细节

该方法执行以下步骤：

1. 构建请求 - 将记忆列表包装成 JSON 载荷：{"memories": memories} mem0/client/main.py:535。
2. 发送 HTTP 请求 - 使用内部客户端向 /v1/batch/ 端点发送 PUT 请求 mem0/client/main.py:536。
3. 校验响应 - 如果请求失败，通过 response.raise_for_status() 抛出 HTTP 状态错误 mem0/client/main.py:537。
4. 遥测 - 捕获 client.batch_update 事件用于分析 mem0/client/main.py:538。
5. 返回结果 - 返回服务器解析后的 JSON 响应 mem0/client/main.py:539。

来源： mem0/client/main.py:515-539

批量删除

batch_delete 方法可以在单个 API 调用中通过 ID 删除多个记忆。

方法签名

# 同步版本
def batch_delete(self, memories: List[Dict[str, Any]]) -> Dict[str, Any]

# 异步版本
async def batch_delete(self, memories: List[Dict[str, Any]]) -> Dict[str, Any]

参数：

记忆字典结构：

返回： Dict[str, Any]，包含服务器返回的删除确认信息。

来源： mem0/client/main.py:542-565，mem0/client/main.py:1113-1138

实现细节

该方法执行以下步骤：

1. 构建请求 - 将记忆列表包装成 JSON 载荷：{"memories": memories} mem0/client/main.py:561。
2. 发送 HTTP 请求 - 向 /v1/batch/ 端点发送包含 JSON 请求体的 DELETE 请求。注意：DELETE 请求包含 JSON 请求体，通过使用 self.client.request("DELETE", ...) 处理 mem0/client/main.py:562。
3. 校验响应 - 如果请求失败，抛出 HTTP 状态错误 mem0/client/main.py:563。
4. 遥测 - 捕获 client.batch_delete 事件用于分析 mem0/client/main.py:564。
5. 返回结果 - 返回服务器解析后的 JSON 响应 mem0/client/main.py:565。

来源： mem0/client/main.py:542-565

错误处理

两种批量操作都使用了 @api_error_handler 装饰器 mem0/client/main.py:136，该装饰器提供了 HTTP 状态码与 Mem0 异常之间的一致映射。

异常类型

来源： mem0/client/main.py:136-160

性能考量

何时使用批量操作

在以下情况下使用批量操作：

• 在单个逻辑操作中更新/删除多个记忆（通常为 5 个以上）。
• 网络延迟是关注点（高延迟连接）。
• 执行批量数据迁移或清理操作。

在以下情况下使用单个操作：

• 需要对每个特定记忆修改进行实时反馈/确认。
• 操作因用户交互而自然分散。

性能对比

来源： mem0/client/main.py:515-565

限制与约束

仅限平台 API

批量操作在自托管/开源版本的 Memory 类中不可用。它们仅通过 MemoryClient 和 AsyncMemoryClient 在平台 API 中提供 mem0/client/main.py:36。自托管用户必须逐个迭代执行 update 或 delete 调用。

原子性

批量操作在数据库层面通常不是原子性的：

• 可能出现部分失败（部分记忆更新/删除成功，部分失败）。
• 客户端应检查响应载荷以验证批次中各个项目的状态。
• 如果同一批次中的后续项目失败，不会自动回滚已成功的更新。

来源： mem0/client/main.py:515-565，mem0/memory/storage.py:193-215