企业官网建设流程全解析

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：抽象基类BaseKeyValueStorage；
• in_memory.py：内存实现InMemoryKeyValueStorage；
• json.py：JSON 文件实现JsonStorage与序列化工具CamelJSONEncoder；
• redis.py：Redis 实现RedisStorage；
• mem0_cloud.py：Mem0 云记忆实现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）：

正是这套极简契约，让上层代码（如 Agent 记忆模块）可以用完全一致的方式操作内存缓存、磁盘文件或分布式缓存，这正是“统一接口”价值的体现。官方指南 docs/key_modules/storages.md 亦将键值存储定位为“从简单键值记录到高性能向量数据库、图引擎的统一存取层”中最基础的一环。

InMemoryKeyValueStorage：轻量快速的临时存储

实现原理

in_memory.py 中的InMemoryKeyValueStorage使用一个 Python 列表self.memory_list作为底层容器。其实现有三个值得注意的细节：

1. save使用deepcopy(records)深拷贝后再extend进内存列表；
2. load同样返回deepcopy(self.memory_list)；
3. 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：人可读的持久化文件存储

核心实现

JsonStorage（json.py）将记录以JSON Lines（每行一个 JSON 对象）的形式写入磁盘文件，构造参数path默认为./chat_history.json，构造时会自动touch()创建文件。三个操作的实现细节：

• save：以追加模式（"a"）打开文件，将每条记录json.dumps(..., cls=CamelJSONEncoder) + "\n"逐行写入；
• load：以读取模式逐行json.loads(..., object_hook=self._json_object_hook)还原；
• clear：以写模式（"w"）打开文件后不写入任何内容，从而清空文件。

由于是追加写入，多次save调用之间的记录会共存，load返回全部历史记录，天然适配“对话历史持续累积”的场景。

CamelJSONEncoder：CAMEL 专属类型的序列化

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：面向分布式缓存的异步实现

构造与连接

RedisStorage（redis.py）面向“需要持久化与高可用的分布式缓存系统”设计。其构造参数为：

构造器被@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_ascii=False)后写入sid对应的 key；若传入可选的expire（秒），则使用SETEX设置过期时间，否则用SET永久保存；
• load：GET sid并json.loads还原，key 不存在或异常时返回[]；
• clear：DELETE sid。

此外，RedisStorage实现了上下文管理器协议（__enter__/__exit__），with块退出时自动异步关闭客户端连接。

使用示例

from camel.storages.key_value_storages import RedisStorage storage = RedisStorage(sid="chat_session_001", url="redis://localhost:6379") storage.save([{'key1': 'value1'}], expire=3600) # 1 小时后过期 print(storage.load()) # >>> [{'key1': 'value1'}] storage.clear()

适用于多进程/多实例共享、需要跨 Agent 共享状态或持久化的分布式场景。

Mem0Storage：面向 Agent 的云记忆后端

虽然 RST 未单列该子模块，但 __init__.py 已将其作为一级导出，docs/reference/camel.storages.key_value_storages.mem0_cloud.md 亦有对应参考文档。Mem0Storage（mem0_cloud.py）基于 Mem0 云服务实现“带上下文地存储、搜索、管理文本”，构造参数包括：

• agent_id（必填）：默认关联记忆的 Agent ID；
• api_key：认证密钥，缺省时从环境变量MEM0_API_KEY读取；
• user_id：默认关联记忆的用户 ID；
• metadata：随记忆附带的默认元数据。

构造器由@dependencies_required('mem0')与@api_keys_required([("api_key", "MEM0_API_KEY")])双重守护，缺依赖或缺密钥会直接报错，防止误用。其save从每条记录中提取message.content与role_at_backend组装成消息列表调用 Mem0 的add；load通过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 对键值存储的定位与各实现的特性，可按如下原则选型：

实战要点与源码验证

1. 接口一致性：四种实现均继承BaseKeyValueStorage，上层只需面向save / load / clear编程即可平滑切换后端，这从 test_key_value_storages.py 用同一套断言参数化测试 in-memory 与 json 两种后端可见一斑。
2. 深拷贝防篡改：InMemoryKeyValueStorage的存/取都经过deepcopy，保证外部修改不影响已存数据；若你的场景追求零拷贝性能，可自行评估取舍。
3. 枚举无损往返：JsonStorage+CamelJSONEncoder组合可持久化RoleType等枚举与 Pydantic 结构化对象，这是将 CAMEL 内部消息直接落盘的关键能力。
4. Redis 的异步调度：RedisStorage在 asyncio 环境中使用run_coroutine_threadsafe，在同步环境中使用run_until_complete，二者兼得；善用上下文管理器可确保连接释放。
5. 依赖与密钥守护：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