CAMEL 记忆块(Memory Blocks)深度解析:ChatHistoryBlock 与 VectorDBBlock 架构与实战
【免费下载链接】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 提供了可插拔的记忆系统。本文聚焦于camel.memories.blocks包——记忆系统中最基础的组成单元:ChatHistoryBlock(对话历史块)与VectorDBBlock(向量数据库块)。你将理解这两种记忆块如何协同工作,掌握窗口化检索、keep_rate 衰减评分、向量相似度召回等核心机制,并学会如何基于MemoryBlock抽象基类在真实代码中组合出会话记忆、向量记忆与长期记忆三种 Agent 记忆形态。
记忆块在 CAMEL 记忆架构中的位置
在 CAMEL 中,记忆系统采用分层设计,其核心依赖关系如下(对应 camel/memories/init.py 的导出结构):
MemoryBlock:抽象基类,定义记忆块的最小接口(写记录、清空、可选删除);ChatHistoryBlock与VectorDBBlock:两个具体实现,即本文主角(位于 camel/memories/blocks/);AgentMemory:面向 Agent 的更高层抽象,负责retrieve与get_context_creator;ChatHistoryMemory、VectorDBMemory、LongtermAgentMemory:三种可直接接入 Agent 的记忆封装,内部委托给上面的记忆块(见 camel/memories/agent_memories.py)。
MemoryBlock的接口定义在 camel/memories/base.py:
write_records(records):批量写入(抽象方法);write_record(record):单条写入的便捷封装;pop_records(count):移除并返回最近的 N 条记录(基类默认抛NotImplementedError);remove_records_by_indices(indices):按索引移除记录(默认抛NotImplementedError);clear():清空全部记录(抽象方法)。
值得注意的是,MemoryBlock刻意不定义统一的retrieve接口——因为不同记忆块的检索语义差异很大(滑动窗口 vs. 向量相似度),这一设计让每个记忆块可以自由定义最适合自己的检索方式。
ChatHistoryBlock:对话历史的窗口化记忆
ChatHistoryBlock位于 camel/memories/blocks/chat_history_block.py,负责维护 Agent 的完整对话历史,核心能力是按最近消息数量(窗口)检索,并通过 keep_rate 为每条历史消息计算"重要性分数"。
构造参数与默认值
from camel.memories import ChatHistoryBlock # 全部使用默认值 block = ChatHistoryBlock() # 指定存储后端与保留率 from camel.storages.key_value_storages.in_memory import InMemoryKeyValueStorage block = ChatHistoryBlock( storage=InMemoryKeyValueStorage(), # 键值存储,默认 InMemoryKeyValueStorage keep_rate=0.9, # 历史消息分数衰减率,默认 0.9 )| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
storage | BaseKeyValueStorage | None(自动用InMemoryKeyValueStorage) | 对话历史的键值存储后端 |
keep_rate | float | 0.9 | 历史消息分数衰减率,必须在 [0, 1] 区间,否则构造时抛出ValueError |
keep_rate的语义:在历史消息中,最后一条消息的分数恒为 1.0,每向前回溯一步,分数乘以一次keep_rate。keep_rate越高,越倾向于在上下文构建时保留更多历史消息。这一分数随后会被ScoreBasedContextCreator等上下文构建器消费,用于决定哪些历史记录进入最终上下文(其实现见 camel/memories/context_creators/score_based.py)。
retrieve:窗口化检索与系统消息保护
retrieve(window_size)是块的核心方法:
window_size=None:返回全部历史记录;window_size=0:返回空列表(但系统/开发者消息除外,见下);window_size=N:返回最近 N 条非系统消息,并在开头保留首条 SYSTEM/DEVELOPER 消息(如果存在)。
源码中的处理逻辑(chat_history_block.py)可以用两个示例概括:
- 场景一:首条消息为 SYSTEM,共 5 条,
window_size=2→[system_msg] + [user_msg3, user_msg4]; - 场景二:首条消息为 USER,共 5 条,
window_size=3→[user_msg3, user_msg4, user_msg5]。
系统消息保护意味着:无论窗口多小,Agent 的 system prompt(角色设定)都不会被截掉,保证 Agent 行为设定始终在上下文中。
检索打分:越近的消息分数越高
retrieve返回的是List[ContextRecord](每条包含memory_record、score、timestamp三要素,结构定义见 camel/memories/records.py)。打分规则(chat_history_block.py):
- 系统消息固定
score=1.0,永不衰减; - 其余消息从最近到最远,分数依次乘以
keep_rate(即最近一条为 1.0、上一条为 0.9、再上一条为 0.81……); - 最终按时间正序返回。
写入、清空与记录删除
write_records(records):将MemoryRecord序列化为 dict 后交给键值存储持久化;clear():清空全部消息;pop_records(count):从末尾移除最近count条记录并返回被移除的记录(按时间正序)。实现同样保护首条系统消息,且要求count为非负整数;remove_records_by_indices(indices):按当前记录列表的 0 基索引批量删除记录(索引 0 的系统/开发者消息受保护,不会被删除)。
VectorDBBlock:基于向量相似度的语义记忆
VectorDBBlock位于 camel/memories/blocks/vectordb_block.py,负责将消息转为向量并存入向量数据库,检索时按语义相似度召回最相关的历史记录——它不依赖消息的新旧,而是依赖内容的相关性。
构造参数与默认值
from camel.memories import VectorDBBlock # 全部使用默认值 block = VectorDBBlock() # 指定向量存储与嵌入模型 from camel.embeddings import OpenAIEmbedding from camel.storages.vectordb_storages import QdrantStorage block = VectorDBBlock( storage=QdrantStorage(vector_dim=1536), # 默认按 embedding 输出维度自动创建 embedding=OpenAIEmbedding(), # 默认 OpenAIEmbedding )| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
storage | BaseVectorStorage | None(自动用QdrantStorage) | 向量数据库存储后端 |
embedding | BaseEmbedding | None(自动用OpenAIEmbedding) | 将消息内容转为向量表示的嵌入模型 |
构造时,块会通过self.embedding.get_output_dim()获取向量维度,并以此初始化默认的QdrantStorage(vector_dim=...),保证维度一致性(vectordb_block.py)。
retrieve:以关键词查相似记录
records = block.retrieve(keyword="如何配置缓存", limit=3)keyword:查询字符串,会被嵌入为向量后发起相似度检索;limit:最多返回的相似消息条数,默认3。
底层流程(vectordb_block.py):
self.embedding.embed(keyword)将查询词转成查询向量;- 构造
VectorDBQuery(query_vector=..., top_k=limit)提交给存储后端(VectorDBQuery定义于 camel/storages/vectordb_storages/base.py,top_k默认 1); - 对每条查询结果构造
ContextRecord,其score直接取自向量检索的similarity(余弦相似度等),timestamp来自记录 payload。
write_records:过滤空内容再向量化
写入时会先过滤掉 content 为空或全空白的记录,避免为无意义文本浪费向量空间;随后将每条有效记录包装为VectorRecord(向量 + payload + UUID)批量写入存储(vectordb_block.py)。
clear()则清空整个向量库。注意:向量记忆不支持按索引删除,因为向量的分布特性决定了无法简单按位置回滚。
从记忆块到 Agent 记忆:三种组合方式
记忆块本身不直接接入 Agent,它们通过AgentMemory封装后使用(实现全部在 camel/memories/agent_memories.py)。
ChatHistoryMemory:纯对话历史
对ChatHistoryBlock的一层薄封装,额外支持window_size参数。当实际取回的记录数达到窗口上限时,会发出UserWarning提示历史消息被截断(agent_memories.py)。它还实现了clean_tool_calls(),可清理 FUNCTION/TOOL 角色消息以及带tool_calls的 ASSISTANT 消息,用于节省 token。
VectorDBMemory:语义检索记忆
对VectorDBBlock的封装。关键设计:写入时假设最后一条 USER 输入即为"当前话题"(_current_topic),检索时用当前话题作为查询词,因此"最近的用户问题会驱动相似历史召回"(agent_memories.py)。该类明确不支持pop_records与remove_records_by_indices,因为向量库无法做顺序回滚。
LongtermAgentMemory:长期记忆(二者兼得)
from camel.memories import LongtermAgentMemory, ChatHistoryBlock, VectorDBBlock memory = LongtermAgentMemory( context_creator=context_creator, # BaseContextCreator 实例 chat_history_block=ChatHistoryBlock(),# 默认自动创建 vector_db_block=VectorDBBlock(), # 默认自动创建 retrieve_limit=3, # 向量召回条数上限 agent_id="my_agent", # 关联的 Agent ID )LongtermAgentMemory.retrieve()的合并策略非常巧妙(agent_memories.py):
chat_history = self.chat_history_block.retrieve() vector_db_retrieve = self.vector_db_block.retrieve(self._current_topic, self.retrieve_limit) return chat_history[:1] + vector_db_retrieve + chat_history[1:]即:首条系统消息 + 向量召回的语义相关记录 + 其余对话历史。这样既保证了角色设定始终在场,又让 Agent 既能"记住最近说了什么"(短期窗口),又能"想起很久以前的相关知识"(长期语义),这正是 CAMEL 长期记忆能力的核心设计。
write_records会把记录同时写入两个块,并同步更新当前话题;pop_records与remove_records_by_indices只作用于对话历史块,向量记忆保持不变。
配套数据结构:MemoryRecord 与 ContextRecord
理解记忆块需要先认识两条核心数据模型(camel/memories/records.py):
MemoryRecord:记忆系统的基本存储单元,字段包括message(BaseMessage 内容)、role_at_backend(后端角色枚举,如 SYSTEM/USER/ASSISTANT)、uuid(唯一标识)、extra_info(附加键值)、timestamp(纳秒精度时间戳)、agent_id。提供to_dict()/from_dict()完成序列化往返——ChatHistoryBlock正是通过它们与键值存储交互,VectorDBBlock则把 dict 作为 payload 存入向量库;ContextRecord:检索结果,包装memory_record并附带score(供上下文构建器权衡取舍)与timestamp。
实战:完整可运行的组合示例
将记忆块接入一个 CAMEL Agent 的典型流程如下(参考 examples/memories 目录下的官方示例):
from camel.memories import ( ChatHistoryBlock, LongtermAgentMemory, MemoryRecord, ScoreBasedContextCreator, VectorDBBlock, ) from camel.messages import BaseMessage from camel.types import OpenAIBackendRole, RoleType from camel.utils import OpenAITokenCounter # 1. 创建两条记忆块(也可直接使用默认构造) chat_block = ChatHistoryBlock(keep_rate=0.9) # 对话历史,衰减率 0.9 vector_block = VectorDBBlock() # 向量记忆,默认 OpenAIEmbedding + QdrantStorage # 2. 组装长期记忆:上下文构建器负责把记录按 token 预算拼装成 OpenAI 消息 memory = LongtermAgentMemory( context_creator=ScoreBasedContextCreator( token_counter=OpenAITokenCounter(model="gpt-4o"), token_limit=4096, ), chat_history_block=chat_block, vector_db_block=vector_block, retrieve_limit=3, agent_id="demo_agent", ) # 3. 写入记录(底层同时写入两个块) record = MemoryRecord( message=BaseMessage( role_name="user", role_type=RoleType.USER, meta_dict=None, content="CAMEL 的向量记忆如何工作?", ), role_at_backend=OpenAIBackendRole.USER, ) memory.write_records([record]) # 4. 构建最终上下文(OpenAIMessage 列表 + token 总数) context_messages, total_tokens = memory.get_context()get_context()是AgentMemory提供的便捷方法(camel/memories/base.py):它调用context_creator.create_context(memory.retrieve()),最终返回可直接发给 LLM 的消息列表与 token 统计。而ScoreBasedContextCreator还带 token 计数缓存:当消息数量与上次 LLM 响应一致时直接复用缓存值,新增消息时用字符级估算增量累加,显著减少重复的 token 计数开销(score_based.py)。
测试验证:记忆块的正确性保障
仓库在 test/memories/test_blocks.py 中为两个记忆块提供了完整单元测试,是理解其行为的权威参考:
- ChatHistoryBlock:默认/自定义存储注入、
window_size窗口检索(retrieve(window_size=5)恰好返回 5 条)、零窗口返回空列表、无窗口返回全部历史、空历史返回空列表、写入与清空调用存储对应方法; - VectorDBBlock:默认组件自动创建、自定义
storage/embedding注入、检索结果顺序与 payload 内容还原、写入调用storage.add、清空调用storage.clear。
测试还通过create_autospec(BaseKeyValueStorage)/create_autospec(BaseVectorStorage)模拟存储层,印证了两个记忆块对存储后端的依赖仅限定于抽象接口,因此可以无缝替换为 Redis、MongoDB 等键值存储或 Chroma、PGVector 等向量存储(存储实现见 camel/storages/key_value_storages/ 与 camel/storages/vectordb_storages/)。更上层的封装测试见 test/memories/test_agent_memories.py。
总结与选型建议
| 需求场景 | 推荐记忆块/记忆封装 |
|---|---|
| 仅需最近的对话上下文,关注 token 开销 | ChatHistoryBlock+window_size(或ChatHistoryMemory) |
| 需要跨长对话召回语义相似的历史知识 | VectorDBBlock(或VectorDBMemory) |
| 既要短期对话、又要长期语义记忆 | LongtermAgentMemory(内部组合两个块) |
ChatHistoryBlock与VectorDBBlock一"近"一"远"、一"序"一"义",共同构成了 CAMEL 记忆系统的底层基石。理解它们的构造参数、检索策略与删除语义,是深入定制 Agent 记忆行为(如调整keep_rate控制历史保留倾向、替换存储后端实现持久化、更换嵌入模型适配多语言检索)的前提。更多记忆相关模块可继续阅读 camel/memories/ 与 docs/key_modules/memory.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),仅供参考