CAMEL 键值存储Key-Value Storage模块解析从 InMemory、JSON 到 Redis 的统一数据存取接口【免费下载链接】camel CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel导读在 CAMEL 多智能体框架中Agent 的记忆Memory、对话历史Chat History与各类结构化中间数据需要一套统一、可插拔的存取机制。camel.storages.key_value_storages正是为此设计的键值存储模块它通过一个抽象基类BaseKeyValueStorage定义了save / load / clear三个标准接口并提供内存、JSON 文件、Redis、Mem0 云服务四类开箱即用的实现。阅读完本文你将掌握 CAMEL 键值存储层的抽象设计、各实现的参数语义与底层原理并能依据场景临时缓存、持久化、分布式、云记忆正确选型与组合使用。本文以 docs/camel.storages.key_value_storages.rst 的 API 结构为主线结合 camel/storages/key_value_storages 目录下的源码实现、docs/key_modules/storages.md 的官方指南以及 test/storages/key_value_storages/test_key_value_storages.py 的测试用例展开。模块概览统一接口下的多后端存储包结构与导出面camel.storages.key_value_storages包位于 camel/storages/key_value_storages从包结构看包含 5 个实现文件base.py抽象基类BaseKeyValueStoragein_memory.py内存实现InMemoryKeyValueStoragejson.pyJSON 文件实现JsonStorage与序列化工具CamelJSONEncoderredis.pyRedis 实现RedisStoragemem0_cloud.pyMem0 云记忆实现Mem0Storage。其中 RST 文档通过automodule指令明确列出了base、in_memory、json、redis四个子模块而包的 __init__.py 进一步将CamelJSONEncoder与Mem0Storage一并导出因此实际可用的导入面为from camel.storages.key_value_storages import ( BaseKeyValueStorage, InMemoryKeyValueStorage, JsonStorage, RedisStorage, CamelJSONEncoder, Mem0Storage, )三个抽象方法的契约BaseKeyValueStorage是一个抽象基类ABC其核心设计是“通过 Python 字典交互”官方文档将其定位为可被 JSON 文件存储、NoSQL 数据库MongoDB、Redis以及内存字典等多种后端继承的公共契约。它只声明了三个抽象方法base.py方法签名语义savesave(records: List[Dict[str, Any]]) - None批量保存记录每条记录是一个字典loadload() - List[Dict[str, Any]]读取全部已保存记录clearclear() - None清空所有记录正是这套极简契约让上层代码如 Agent 记忆模块可以用完全一致的方式操作内存缓存、磁盘文件或分布式缓存这正是“统一接口”价值的体现。官方指南 docs/key_modules/storages.md 亦将键值存储定位为“从简单键值记录到高性能向量数据库、图引擎的统一存取层”中最基础的一环。InMemoryKeyValueStorage轻量快速的临时存储实现原理in_memory.py 中的InMemoryKeyValueStorage使用一个 Python 列表self.memory_list作为底层容器。其实现有三个值得注意的细节save使用deepcopy(records)深拷贝后再extend进内存列表load同样返回deepcopy(self.memory_list)clear直接调用memory_list.clear()。深拷贝的存在意味着写入存储后外部对原字典的后续修改不会影响已保存的数据测试用例 test_key_value_storages.py 专门验证了这一行为保存msg1后对msg1.clear()再load()出的记录仍保持原值确保“无信息丢失”的契约。由于数据仅驻留进程内存程序结束即丢失它适用于原型验证、单元测试、进程内的临时缓存与快速迭代场景。使用示例from camel.storages.key_value_storages import InMemoryKeyValueStorage memory_storage InMemoryKeyValueStorage() memory_storage.save([{key1: value1}, {key2: value2}]) records memory_storage.load() print(records) # [{key1: value1}, {key2: value2}] memory_storage.clear() print(memory_storage.load()) # []JsonStorage人可读的持久化文件存储核心实现JsonStoragejson.py将记录以JSON Lines每行一个 JSON 对象的形式写入磁盘文件构造参数path默认为./chat_history.json构造时会自动touch()创建文件。三个操作的实现细节save以追加模式a打开文件将每条记录json.dumps(..., clsCamelJSONEncoder) \n逐行写入load以读取模式逐行json.loads(..., object_hookself._json_object_hook)还原clear以写模式w打开文件后不写入任何内容从而清空文件。由于是追加写入多次save调用之间的记录会共存load返回全部历史记录天然适配“对话历史持续累积”的场景。CamelJSONEncoderCAMEL 专属类型的序列化JsonStorage与普通 JSON 存储的关键区别在于 CamelJSONEncoder 这个自定义编码器它解决了 CAMEL 生态中两类对象无法直接被json.dumps序列化的问题CAMEL 枚举类型通过CAMEL_ENUMS类变量注册了RoleType、TaskType、ModelType、OpenAIBackendRole四种枚举序列化时统一转为{__enum__: RoleType.USER}这样的标记字典Pydantic BaseModel结构化输出structured outputs产生的模型对象序列化为obj.model_dump()的结果。与之配套JsonStorage._json_object_hook在读取时识别__enum__标记并还原为对应的枚举成员从而保证枚举与结构化对象能够无损地写入、读出。测试用例中也验证了包含RoleType.USER等枚举字段的字典可以完整往返test_key_value_storages.py。使用示例from pathlib import Path from camel.storages.key_value_storages import JsonStorage json_storage JsonStorage(Path(my_data.json)) json_storage.save([{key1: value1}, {key2: value2}]) records json_storage.load() print(records) # [{key1: value1}, {key2: value2}] json_storage.clear()适合用于配置文件、小型持久化数据集、导出/导入流程以及需要人工检视的日志型数据。RedisStorage面向分布式缓存的异步实现构造与连接RedisStorageredis.py面向“需要持久化与高可用的分布式缓存系统”设计。其构造参数为参数默认值说明sid必填存储实例 ID用于标识记录空间对应 Redis 的 keyurlredis://localhost:6379Redis 连接 URLloopNone事件循环缺省时取当前asyncio.get_event_loop()**kwargs—透传给 Redis 客户端的其他配置构造器被dependencies_required(redis)装饰未安装redis包时会抛出ImportError底层通过redis.asyncio.from_url(self._url, **kwargs)创建异步客户端可通过client属性获取原生的redis.asyncio.Redis实例做更底层的操作。同步外壳 异步内核值得注意的实现手法是对外暴露的save / load / clear是同步方法内部通过_run_async调度器将请求转发给_async_save / _async_load / _async_clear三个协程redis.py当事件循环未运行时使用loop.run_until_complete(coro)同步执行当事件循环正在运行如处于 asyncio 应用内时改用asyncio.run_coroutine_threadsafe(coro, loop)投递执行。具体的数据语义为save将整个记录列表json.dumps(records, ensure_asciiFalse)后写入sid对应的 key若传入可选的expire秒则使用SETEX设置过期时间否则用SET永久保存loadGET sid并json.loads还原key 不存在或异常时返回[]clearDELETE sid。此外RedisStorage实现了上下文管理器协议__enter__/__exit__with块退出时自动异步关闭客户端连接。使用示例from camel.storages.key_value_storages import RedisStorage storage RedisStorage(sidchat_session_001, urlredis://localhost:6379) storage.save([{key1: value1}], expire3600) # 1 小时后过期 print(storage.load()) # [{key1: value1}] storage.clear()适用于多进程/多实例共享、需要跨 Agent 共享状态或持久化的分布式场景。Mem0Storage面向 Agent 的云记忆后端虽然 RST 未单列该子模块但 __init__.py 已将其作为一级导出docs/reference/camel.storages.key_value_storages.mem0_cloud.md 亦有对应参考文档。Mem0Storagemem0_cloud.py基于 Mem0 云服务实现“带上下文地存储、搜索、管理文本”构造参数包括agent_id必填默认关联记忆的 Agent IDapi_key认证密钥缺省时从环境变量MEM0_API_KEY读取user_id默认关联记忆的用户 IDmetadata随记忆附带的默认元数据。构造器由dependencies_required(mem0)与api_keys_required([(api_key, MEM0_API_KEY)])双重守护缺依赖或缺密钥会直接报错防止误用。其save从每条记录中提取message.content与role_at_backend组装成消息列表调用 Mem0 的addload通过get_all拉取后转换为 CAMEL 的MemoryRecord包含uuid、message、role_at_backend、timestamp、agent_id等字段再以字典形式返回clear则按user_id/agent_id过滤调用delete_users删除记忆。它面向需要跨会话、跨 Agent 持久记忆的 LLM 应用场景。选型指南四种后端怎么选结合官方指南 docs/key_modules/storages.md 对键值存储的定位与各实现的特性可按如下原则选型后端持久性适用场景关键约束InMemoryKeyValueStorage否进程结束即失原型、测试、进程内缓存数据不跨进程共享JsonStorage是磁盘文件配置、日志、小型数据集、导入导出需保证 JSON 可序列化枚举用CamelJSONEncoderRedisStorage是取决于配置分布式缓存、多实例共享、高可用需安装redis包并运行 Redis 服务支持expire过期Mem0Storage是云端Agent 跨会话持久记忆、记忆搜索需安装mem0包并配置MEM0_API_KEY实战要点与源码验证接口一致性四种实现均继承BaseKeyValueStorage上层只需面向save / load / clear编程即可平滑切换后端这从 test_key_value_storages.py 用同一套断言参数化测试 in-memory 与 json 两种后端可见一斑。深拷贝防篡改InMemoryKeyValueStorage的存/取都经过deepcopy保证外部修改不影响已存数据若你的场景追求零拷贝性能可自行评估取舍。枚举无损往返JsonStorageCamelJSONEncoder组合可持久化RoleType等枚举与 Pydantic 结构化对象这是将 CAMEL 内部消息直接落盘的关键能力。Redis 的异步调度RedisStorage在 asyncio 环境中使用run_coroutine_threadsafe在同步环境中使用run_until_complete二者兼得善用上下文管理器可确保连接释放。依赖与密钥守护Redis 与 Mem0 实现分别通过dependencies_required与api_keys_required做前置校验test/storages/test_storages_decorators.py 覆盖了这类装饰器行为在缺失依赖/密钥时给出明确错误而非运行时模糊异常。小结CAMEL 的键值存储模块用 30 行左右的抽象基类约束出了一套“字典进、字典出”的稳定契约并在其下提供了从进程内存、磁盘 JSON、异步 Redis 到云端 Mem0 的完整梯度实现。无论你是为 Agent 搭建临时缓存、为对话历史寻找可持久化落盘方案还是构建跨实例共享的分布式状态都可以在 camel/storages/key_value_storages 中找到对应的现成组件进一步阅读 docs/camel.storages.key_value_storages.rst 可获取该包完整的 API 成员清单而 docs/key_modules/storages.md 则给出了它与向量存储、图存储并列的宏观定位。【免费下载链接】camel CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 CAMEL 键值存储（Key-Value Storage）模块解析：从 InMemory、JSON 到 Redis 的统一数据存取接口
CAMEL 键值存储Key-Value Storage模块解析从 InMemory、JSON 到 Redis 的统一数据存取接口【免费下载链接】camel CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel导读在 CAMEL 多智能体框架中Agent 的记忆Memory、对话历史Chat History与各类结构化中间数据需要一套统一、可插拔的存取机制。camel.storages.key_value_storages正是为此设计的键值存储模块它通过一个抽象基类BaseKeyValueStorage定义了save / load / clear三个标准接口并提供内存、JSON 文件、Redis、Mem0 云服务四类开箱即用的实现。阅读完本文你将掌握 CAMEL 键值存储层的抽象设计、各实现的参数语义与底层原理并能依据场景临时缓存、持久化、分布式、云记忆正确选型与组合使用。本文以 docs/camel.storages.key_value_storages.rst 的 API 结构为主线结合 camel/storages/key_value_storages 目录下的源码实现、docs/key_modules/storages.md 的官方指南以及 test/storages/key_value_storages/test_key_value_storages.py 的测试用例展开。模块概览统一接口下的多后端存储包结构与导出面camel.storages.key_value_storages包位于 camel/storages/key_value_storages从包结构看包含 5 个实现文件base.py抽象基类BaseKeyValueStoragein_memory.py内存实现InMemoryKeyValueStoragejson.pyJSON 文件实现JsonStorage与序列化工具CamelJSONEncoderredis.pyRedis 实现RedisStoragemem0_cloud.pyMem0 云记忆实现Mem0Storage。其中 RST 文档通过automodule指令明确列出了base、in_memory、json、redis四个子模块而包的 __init__.py 进一步将CamelJSONEncoder与Mem0Storage一并导出因此实际可用的导入面为from camel.storages.key_value_storages import ( BaseKeyValueStorage, InMemoryKeyValueStorage, JsonStorage, RedisStorage, CamelJSONEncoder, Mem0Storage, )三个抽象方法的契约BaseKeyValueStorage是一个抽象基类ABC其核心设计是“通过 Python 字典交互”官方文档将其定位为可被 JSON 文件存储、NoSQL 数据库MongoDB、Redis以及内存字典等多种后端继承的公共契约。它只声明了三个抽象方法base.py方法签名语义savesave(records: List[Dict[str, Any]]) - None批量保存记录每条记录是一个字典loadload() - List[Dict[str, Any]]读取全部已保存记录clearclear() - None清空所有记录正是这套极简契约让上层代码如 Agent 记忆模块可以用完全一致的方式操作内存缓存、磁盘文件或分布式缓存这正是“统一接口”价值的体现。官方指南 docs/key_modules/storages.md 亦将键值存储定位为“从简单键值记录到高性能向量数据库、图引擎的统一存取层”中最基础的一环。InMemoryKeyValueStorage轻量快速的临时存储实现原理in_memory.py 中的InMemoryKeyValueStorage使用一个 Python 列表self.memory_list作为底层容器。其实现有三个值得注意的细节save使用deepcopy(records)深拷贝后再extend进内存列表load同样返回deepcopy(self.memory_list)clear直接调用memory_list.clear()。深拷贝的存在意味着写入存储后外部对原字典的后续修改不会影响已保存的数据测试用例 test_key_value_storages.py 专门验证了这一行为保存msg1后对msg1.clear()再load()出的记录仍保持原值确保“无信息丢失”的契约。由于数据仅驻留进程内存程序结束即丢失它适用于原型验证、单元测试、进程内的临时缓存与快速迭代场景。使用示例from camel.storages.key_value_storages import InMemoryKeyValueStorage memory_storage InMemoryKeyValueStorage() memory_storage.save([{key1: value1}, {key2: value2}]) records memory_storage.load() print(records) # [{key1: value1}, {key2: value2}] memory_storage.clear() print(memory_storage.load()) # []JsonStorage人可读的持久化文件存储核心实现JsonStoragejson.py将记录以JSON Lines每行一个 JSON 对象的形式写入磁盘文件构造参数path默认为./chat_history.json构造时会自动touch()创建文件。三个操作的实现细节save以追加模式a打开文件将每条记录json.dumps(..., clsCamelJSONEncoder) \n逐行写入load以读取模式逐行json.loads(..., object_hookself._json_object_hook)还原clear以写模式w打开文件后不写入任何内容从而清空文件。由于是追加写入多次save调用之间的记录会共存load返回全部历史记录天然适配“对话历史持续累积”的场景。CamelJSONEncoderCAMEL 专属类型的序列化JsonStorage与普通 JSON 存储的关键区别在于 CamelJSONEncoder 这个自定义编码器它解决了 CAMEL 生态中两类对象无法直接被json.dumps序列化的问题CAMEL 枚举类型通过CAMEL_ENUMS类变量注册了RoleType、TaskType、ModelType、OpenAIBackendRole四种枚举序列化时统一转为{__enum__: RoleType.USER}这样的标记字典Pydantic BaseModel结构化输出structured outputs产生的模型对象序列化为obj.model_dump()的结果。与之配套JsonStorage._json_object_hook在读取时识别__enum__标记并还原为对应的枚举成员从而保证枚举与结构化对象能够无损地写入、读出。测试用例中也验证了包含RoleType.USER等枚举字段的字典可以完整往返test_key_value_storages.py。使用示例from pathlib import Path from camel.storages.key_value_storages import JsonStorage json_storage JsonStorage(Path(my_data.json)) json_storage.save([{key1: value1}, {key2: value2}]) records json_storage.load() print(records) # [{key1: value1}, {key2: value2}] json_storage.clear()适合用于配置文件、小型持久化数据集、导出/导入流程以及需要人工检视的日志型数据。RedisStorage面向分布式缓存的异步实现构造与连接RedisStorageredis.py面向“需要持久化与高可用的分布式缓存系统”设计。其构造参数为参数默认值说明sid必填存储实例 ID用于标识记录空间对应 Redis 的 keyurlredis://localhost:6379Redis 连接 URLloopNone事件循环缺省时取当前asyncio.get_event_loop()**kwargs—透传给 Redis 客户端的其他配置构造器被dependencies_required(redis)装饰未安装redis包时会抛出ImportError底层通过redis.asyncio.from_url(self._url, **kwargs)创建异步客户端可通过client属性获取原生的redis.asyncio.Redis实例做更底层的操作。同步外壳 异步内核值得注意的实现手法是对外暴露的save / load / clear是同步方法内部通过_run_async调度器将请求转发给_async_save / _async_load / _async_clear三个协程redis.py当事件循环未运行时使用loop.run_until_complete(coro)同步执行当事件循环正在运行如处于 asyncio 应用内时改用asyncio.run_coroutine_threadsafe(coro, loop)投递执行。具体的数据语义为save将整个记录列表json.dumps(records, ensure_asciiFalse)后写入sid对应的 key若传入可选的expire秒则使用SETEX设置过期时间否则用SET永久保存loadGET sid并json.loads还原key 不存在或异常时返回[]clearDELETE sid。此外RedisStorage实现了上下文管理器协议__enter__/__exit__with块退出时自动异步关闭客户端连接。使用示例from camel.storages.key_value_storages import RedisStorage storage RedisStorage(sidchat_session_001, urlredis://localhost:6379) storage.save([{key1: value1}], expire3600) # 1 小时后过期 print(storage.load()) # [{key1: value1}] storage.clear()适用于多进程/多实例共享、需要跨 Agent 共享状态或持久化的分布式场景。Mem0Storage面向 Agent 的云记忆后端虽然 RST 未单列该子模块但 __init__.py 已将其作为一级导出docs/reference/camel.storages.key_value_storages.mem0_cloud.md 亦有对应参考文档。Mem0Storagemem0_cloud.py基于 Mem0 云服务实现“带上下文地存储、搜索、管理文本”构造参数包括agent_id必填默认关联记忆的 Agent IDapi_key认证密钥缺省时从环境变量MEM0_API_KEY读取user_id默认关联记忆的用户 IDmetadata随记忆附带的默认元数据。构造器由dependencies_required(mem0)与api_keys_required([(api_key, MEM0_API_KEY)])双重守护缺依赖或缺密钥会直接报错防止误用。其save从每条记录中提取message.content与role_at_backend组装成消息列表调用 Mem0 的addload通过get_all拉取后转换为 CAMEL 的MemoryRecord包含uuid、message、role_at_backend、timestamp、agent_id等字段再以字典形式返回clear则按user_id/agent_id过滤调用delete_users删除记忆。它面向需要跨会话、跨 Agent 持久记忆的 LLM 应用场景。选型指南四种后端怎么选结合官方指南 docs/key_modules/storages.md 对键值存储的定位与各实现的特性可按如下原则选型后端持久性适用场景关键约束InMemoryKeyValueStorage否进程结束即失原型、测试、进程内缓存数据不跨进程共享JsonStorage是磁盘文件配置、日志、小型数据集、导入导出需保证 JSON 可序列化枚举用CamelJSONEncoderRedisStorage是取决于配置分布式缓存、多实例共享、高可用需安装redis包并运行 Redis 服务支持expire过期Mem0Storage是云端Agent 跨会话持久记忆、记忆搜索需安装mem0包并配置MEM0_API_KEY实战要点与源码验证接口一致性四种实现均继承BaseKeyValueStorage上层只需面向save / load / clear编程即可平滑切换后端这从 test_key_value_storages.py 用同一套断言参数化测试 in-memory 与 json 两种后端可见一斑。深拷贝防篡改InMemoryKeyValueStorage的存/取都经过deepcopy保证外部修改不影响已存数据若你的场景追求零拷贝性能可自行评估取舍。枚举无损往返JsonStorageCamelJSONEncoder组合可持久化RoleType等枚举与 Pydantic 结构化对象这是将 CAMEL 内部消息直接落盘的关键能力。Redis 的异步调度RedisStorage在 asyncio 环境中使用run_coroutine_threadsafe在同步环境中使用run_until_complete二者兼得善用上下文管理器可确保连接释放。依赖与密钥守护Redis 与 Mem0 实现分别通过dependencies_required与api_keys_required做前置校验test/storages/test_storages_decorators.py 覆盖了这类装饰器行为在缺失依赖/密钥时给出明确错误而非运行时模糊异常。小结CAMEL 的键值存储模块用 30 行左右的抽象基类约束出了一套“字典进、字典出”的稳定契约并在其下提供了从进程内存、磁盘 JSON、异步 Redis 到云端 Mem0 的完整梯度实现。无论你是为 Agent 搭建临时缓存、为对话历史寻找可持久化落盘方案还是构建跨实例共享的分布式状态都可以在 camel/storages/key_value_storages 中找到对应的现成组件进一步阅读 docs/camel.storages.key_value_storages.rst 可获取该包完整的 API 成员清单而 docs/key_modules/storages.md 则给出了它与向量存储、图存储并列的宏观定位。【免费下载链接】camel CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考