Memory 系统与消息过滤

学习目标

理解 MetaGPT 的三级记忆架构，以及 Memory 如何支持多维度检索和任务级隔离。

---

项目实践

三级记忆架构

┌─────────────────────────────────────────────────┐
│  rc.news (临时)                                  │
│  本次 observe 到的新消息                         │
│  生命周期：一轮 run()                            │
├─────────────────────────────────────────────────┤
│  rc.memory (持久)                                │
│  所有处理过的消息，按 cause_by 索引              │
│  生命周期：Role 存活期间                         │
├─────────────────────────────────────────────────┤
│  rc.working_memory (任务级）                      │
│  当前任务的上下文，任务完成后 clear               │
│  生命周期：单个 Task                             │
└─────────────────────────────────────────────────┘

Memory 基础类

class Memory(BaseModel):
    storage: list[Message]                    # 消息列表（按时间排序）
    index: DefaultDict[str, list[Message]]    # 按 cause_by 索引的字典

    def add(self, message: Message):
        """添加消息，同时更新 index"""
        self.storage.append(message)
        if message.cause_by:
            self.index[message.cause_by].append(message)

    def get(self, k=0) -> list[Message]:
        """返回最近 k 条消息，k=0 返回全部"""
        return self.storage[-k:]

    def get_by_actions(self, actions: Set) -> list[Message]:
        """返回指定 action 触发的所有消息"""
        rsp = []
        for action in any_to_str_set(actions):
            if action in self.index:
                rsp += self.index[action]
        return rsp

    def get_by_role(self, role: str) -> list[Message]: ...
    def get_by_content(self, content: str) -> list[Message]: ...
    def try_remember(self, keyword: str) -> list[Message]: ...

消息过滤流程

Role._observe() 是消息进入 memory 的关口：

async def _observe(self) -> int:
    # 1. 从 msg_buffer 弹出新消息
    news = self.rc.msg_buffer.pop_all()

    # 2. 过滤：只保留关注的消息
    self.rc.news = [
        n for n in news
        if (n.cause_by in self.rc.watch or self.name in n.send_to)  # 关键过滤条件
        and n not in old_messages
    ]

    # 3. 写入 memory（可选）
    if self.observe_all_msg_from_buffer:
        self.rc.memory.add_batch(news)          # 所有消息
    else:
        self.rc.memory.add_batch(self.rc.news)   # 只保留关注的

rc.watch 的作用：

每个角色通过 _watch() 声明关注的 Action：

# ProductManager 关注 UserRequirement 和 PrepareDocuments
self._watch([UserRequirement, PrepareDocuments])

# Architect 关注 WritePRD
self._watch([WritePRD])

# Engineer 关注 WriteDesign
self._watch([WriteDesign])

这意味着 ProductManager 只会处理由 UserRequirement 或 PrepareDocuments 触发的消息，忽略其他消息。

working_memory：任务级隔离

working_memory 是 RoleContext 中的独立 Memory 实例，用于：

1. Plan-and-act 模式：Planner 使用 working_memory 存储当前任务的上下文
2. 任务完成后清理：self.working_memory.clear() 在任务完成后调用
3. 避免历史污染：每个任务有独立的工作空间

# Planner 使用 working_memory
class Planner(BaseModel):
    working_memory: Memory = Field(default_factory=Memory)

    async def update_plan(self, goal: str = ""):
        rsp = await WritePlan().run(context, max_tasks=max_tasks)
        self.working_memory.add(Message(content=rsp, role="assistant"))
        # ...
        self.working_memory.clear()  # 计划确认后清理

RoleZero 长短期记忆

RoleZero 支持可选的长短期记忆（RoleZeroLongTermMemory），基于 ChromaDB 向量存储：

def set_longterm_memory(self) -> "RoleZero":
    if self.config.role_zero.enable_longterm_memory:
        self.rc.memory = RoleZeroLongTermMemory(
            **self.rc.memory.model_dump(),
            persist_path=self.config.role_zero.longterm_memory_persist_path,
            collection_name=self.name.replace(" ", ""),
            memory_k=self.config.role_zero.memory_k,
            similarity_top_k=self.config.role_zero.similarity_top_k,
            use_llm_ranker=self.config.role_zero.use_llm_ranker,
        )

工作流程：

1. 新消息同时写入短期 memory（列表）和长期 memory（ChromaDB）
2. _think() 时通过 rc.memory.get(memory_k) 获取最近 200 条上下文
3. ChromaDB 提供相似度检索，补充关键词匹配

---

问题与规避

enable_memory=False 的影响

• ProductManager 在 fixed SOP 模式下设置 enable_memory=False，不写入 memory
• 目的：节省 memory 空间和 token（_get_prefix() 中的 history 不会被使用）
• 注意：这不影响 _observe() 的 rc.news 过滤，只影响 rc.memory.add()

long-term memory 检索延迟

• ChromaDB 向量检索比列表遍历慢，尤其在大量记录时
• 对策：similarity_top_k 限制返回数量，use_llm_ranker 控制是否用 LLM 重排序

Memory index 不一致

• 手动删除 storage 中的消息但忘记更新 index 会导致 get_by_actions() 返回已删除消息
• 对策：始终使用 Memory.delete() 方法，不要直接操作 storage

---

设计取舍

列表 + 字典 vs 向量数据库

• 基础 Memory 使用列表+字典，简单高效，适合小规模（<1000 条消息）
• 长短期记忆引入 ChromaDB，支持语义搜索但增加依赖
• 分层设计：用户可以根据场景选择是否启用长短期记忆

cause_by 索引 vs 全文索引

• 按 cause_by 索引支持按 action 检索，是 MetaGPT 消息路由的核心
• 不支持全文搜索（try_remember 使用 keyword in content 线性扫描）
• 长短期记忆弥补了这一点，通过向量嵌入支持语义搜索

---

参考来源

• 源码验证: metagpt/memory/memory.py:20 — Memory 基类
• 源码验证: metagpt/roles/role.py:399 — _observe() 消息过滤
• 源码验证: metagpt/roles/role.py:116 — important_memory 属性
• 源码验证: metagpt/roles/di/role_zero.py:173 — set_longterm_memory()
• 源码验证: metagpt/memory/role_zero_memory.py — RoleZeroLongTermMemory 实现