Agent Memory 架构：短期、长期与共享记忆的设计取舍

没有记忆的 Agent 只是一个"金鱼"——每次对话都从零开始，无法建立真正的连续性。Agent Memory 架构决定了系统能否跨会话学习、跨 Agent 共享知识、在长期任务中保持一致。本文从工程实战出发，系统对比 Agent Memory 的三大类别——短期记忆、长期记忆、共享记忆，以及它们在不同场景下的设计取舍。

为什么 Memory 是 Agent 系统的核心

传统 LLM 是无状态的——每个 prompt 都是独立的请求，没有历史。但 Agent 系统需要解决三类根本问题：

1. 单会话内的上下文延续。用户和 Agent 的多轮对话中，前面的消息决定了后面的回答。没有"短期记忆"，Agent 无法理解指代、省略、上下文。

2. 跨会话的用户记忆。客服 Agent 应该在用户下次来访时记得上次的问题；个人助理 Agent 应该记得用户的偏好、历史决策。跨会话的"长期记忆"是 Agent 个性化的基础。

3. 多 Agent 间的知识共享。在多 Agent 系统中，一个 Agent 检索到的知识应该能被其他 Agent 复用；某个 Agent 学到的经验应该沉淀到共享知识库。

这三种需求对应三种 Memory 架构：短期记忆（Working Memory）、长期记忆（Long-term Memory）、共享记忆（Shared Memory）。它们的存储介质、检索方式、生命周期都截然不同。

Memory 类别 1：短期记忆（Working Memory）

短期记忆存储当前会话的上下文，是 Agent 最基础的 Memory 形式。

from dataclasses import dataclass, field
from typing import Literal
from datetime import datetime

@dataclass
class Message:
    role: Literal["user", "assistant", "system", "tool"]
    content: str
    timestamp: datetime = field(default_factory=datetime.now)
    metadata: dict = field(default_factory=dict)
    
    # 计算 token 估算（粗略）
    @property
    def tokens(self) -> int:
        return len(self.content) // 4

class WorkingMemory:
    def __init__(self, max_tokens: int = 8000, max_messages: int = 50):
        self.messages: list[Message] = []
        self.max_tokens = max_tokens
        self.max_messages = max_messages
        self.summary: str = ""
    
    def add(self, message: Message) -> None:
        self.messages.append(message)
        self._maybe_compress()
    
    def get_context(self) -> list[dict]:
        context = []
        if self.summary:
            context.append({
                "role": "system",
                "content": f"[Conversation summary so far] {self.summary}",
            })
        for msg in self.messages:
            context.append({
                "role": msg.role,
                "content": msg.content,
            })
        return context
    
    def total_tokens(self) -> int:
        return sum(m.tokens for m in self.messages) + (len(self.summary) // 4)
    
    def _maybe_compress(self) -> None:
        if self.total_tokens() > self.max_tokens or len(self.messages) > self.max_messages:
            # 保留最近 4 条 + 总结旧的
            recent = self.messages[-4:]
            old = self.messages[:-4]
            self.summary = self._summarize(old, self.summary)
            self.messages = recent
    
    def _summarize(self, old_messages, prev_summary) -> str:
        # 调用 LLM 做摘要（生产环境实现）
        return prev_summary  # 简化

短期记忆的设计要点：

• 窗口大小：常见 4K-32K tokens，受限于 LLM 上下文窗口
• 压缩策略：超出窗口时按"摘要 + 近期 N 轮"压缩
• 结构化存储：不只是字符串，保留 role、metadata、tool_call_id
• 生命周期：会话结束即销毁，不持久化

短期记忆的痛点：

• 窗口瓶颈：长对话会迅速耗尽
• 信息丢失：摘要必然丢失细节
• 会话间割裂：用户换设备/新会话就完全失忆

Memory 类别 2：长期记忆（Long-term Memory）

长期记忆存储跨会话的用户偏好、历史、知识，让 Agent 真正"认识"用户。

import numpy as np
from datetime import datetime

class LongTermMemory:
    def __init__(self, embedding_model, vector_store):
        self.embedding_model = embedding_model
        self.vector_store = vector_store
        self.embeddings_dim = 1536  # text-embedding-3-small
    
    async def remember(
        self,
        content: str,
        user_id: str,
        memory_type: str = "fact",  # fact / preference / event / skill
        importance: float = 0.5,
        metadata: dict = None,
    ) -> str:
        embedding = await self.embedding_model.aembed(content)
        memory_id = f"{user_id}:{datetime.now().isoformat()}:{hash(content) % 10000}"
        
        await self.vector_store.upsert(
            id=memory_id,
            vector=embedding,
            payload={
                "content": content,
                "user_id": user_id,
                "memory_type": memory_type,
                "importance": importance,
                "created_at": datetime.now().isoformat(),
                "access_count": 0,
                **(metadata or {}),
            },
        )
        return memory_id
    
    async def recall(
        self,
        query: str,
        user_id: str,
        top_k: int = 5,
        memory_types: list[str] | None = None,
        min_importance: float = 0.0,
    ) -> list[dict]:
        query_emb = await self.embedding_model.aembed(query)
        
        filters = {
            "user_id": user_id,
            "importance": {"$gte": min_importance},
        }
        if memory_types:
            filters["memory_type"] = {"$in": memory_types}
        
        results = await self.vector_store.search(
            vector=query_emb,
            top_k=top_k,
            filter=filters,
        )
        
        # 更新访问计数
        for r in results:
            await self.vector_store.update(
                r["id"],
                {"$inc": {"access_count": 1}},
            )
        
        return results

长期记忆的设计模式：

1. 事实记忆（Fact Memory）：用户静态信息（"我叫 Alex"，"我在上海"）
2. 偏好记忆（Preference Memory）：用户偏好（"我喜欢简洁回答"）
3. 事件记忆（Episodic Memory）：重要事件（"上周买了机票"）
4. 技能记忆（Skill Memory）：用户行为模式（"通常用中文提问"）

长期记忆的关键挑战：

• 写入时机：什么时候从短期记忆"升级"为长期记忆？过于频繁写入会污染，过于稀少会丢失
• 去重合并：同一事实多次出现（"我家在北京""住在北京"），需要合并而非重复存储
• 过期机制：某些记忆会过时（"明天的会议"过了明天就无意义）
• 冲突处理：用户的偏好可能改变（"我以前喜欢咖啡，现在喜欢茶"），需要更新而非追加

Mem0 框架专门解决这些问题：它提供了 LLM 驱动的 memory 提取、合并、冲突解决机制。

Memory 类别 3：共享记忆（Shared Memory）

共享记忆解决多 Agent 间的知识共享问题。一个 Agent 检索到的知识，另一个 Agent 应该能用；某个 Agent 学到的经验，其他 Agent 应该能查询。

class SharedMemory:
    def __init__(self, vector_store, namespace: str = "shared"):
        self.vector_store = vector_store
        self.namespace = namespace
        self.access_log: list[dict] = []
    
    async def publish(
        self,
        content: str,
        source_agent: str,
        knowledge_type: str = "general",  # fact / procedure / tool_result / observation
        confidence: float = 0.8,
        tags: list[str] = None,
    ) -> str:
        embedding = await embed(content)
        doc_id = f"{self.namespace}:{source_agent}:{datetime.now().isoformat()}"
        
        await self.vector_store.upsert(
            id=doc_id,
            vector=embedding,
            payload={
                "content": content,
                "source_agent": source_agent,
                "knowledge_type": knowledge_type,
                "confidence": confidence,
                "tags": tags or [],
                "created_at": datetime.now().isoformat(),
                "access_count": 0,
            },
        )
        return doc_id
    
    async def query(
        self,
        query: str,
        requesting_agent: str,
        top_k: int = 5,
        min_confidence: float = 0.6,
    ) -> list[dict]:
        query_emb = await embed(query)
        results = await self.vector_store.search(
            vector=query_emb,
            top_k=top_k,
            filter={
                "namespace": self.namespace,
                "confidence": {"$gte": min_confidence},
            },
        )
        
        # 记录访问
        for r in results:
            r["_meta"] = {
                "requested_by": requesting_agent,
                "requested_at": datetime.now().isoformat(),
            }
        
        return results

共享记忆的应用场景：

1. 研究 Agent 群：研究员 Agent 找到的资料，写作 Agent 应该能直接引用
2. 客服 Agent 群：订单 Agent 查到的物流信息，售后 Agent 应该能查询
3. 代码 Agent 群：架构师 Agent 设计的 API 模式，工程师 Agent 应该能复用
4. 跨会话学习：今天的 Agent 学到的经验，明天的 Agent 也能用

共享记忆的设计原则：

• 来源追踪：每条知识必须标记来源 Agent，便于审计
• 置信度标注：Agent 生成的知识有置信度（"我觉得 80% 对"），低置信度不应被其他 Agent 信任
• 访问控制：某些记忆只有特定 Agent 能访问（敏感业务数据）
• 冲突检测：多 Agent 可能产生矛盾的知识，需要仲裁机制

三大 Memory 类别对比

维度	短期记忆	长期记忆	共享记忆
存储介质	内存	向量数据库	向量数据库 + 元数据
生命周期	单会话	跨会话/永久	跨 Agent/永久
检索方式	顺序/FIFO	语义检索	语义检索 + 来源过滤
写入时机	实时	异步（事件触发）	异步（Agent 主动）
典型框架	LangChain Memory	Mem0, Letta, Zep	Zep, Mem0, OpenMemory
容量	4K-32K tokens	百万级条目	百万级条目
一致性	强	弱（需合并）	弱（需仲裁）
隐私	不持久	用户隔离	角色访问控制

实施路径

第 1 阶段：实现短期记忆（Working Memory），解决单会话上下文问题。 第 2 阶段：引入长期记忆（Long-term Memory），存储用户偏好和事实。 第 3 阶段：实现记忆合并和去重机制，避免长期记忆污染。 第 4 阶段：在多 Agent 系统中实现共享记忆（Shared Memory）。 第 5 阶段：建立记忆审计和过期清理流程。 第 6 阶段：引入记忆可视化，让用户能查看、修改、删除自己的记忆。

总结

Agent Memory 不是单一的"上下文窗口"，而是分层的记忆系统：短期记忆维持会话内上下文，长期记忆沉淀用户偏好和事实，共享记忆让多 Agent 协作。三者存储介质、生命周期、检索方式都不同，不能用一套方案解决所有问题。

Mem0、Letta、Zep、OpenMemory 等框架提供了开箱即用的实现。选择时考虑：数据所有权、隐私合规、检索精度、性能开销。

参考工具：Mem0（LLM 驱动的长期记忆框架）、Zep（生产级长期记忆）、Letta（有状态 Agent 框架）、memvid（视频编码记忆）和 OpenMemory（本地优先的共享记忆）覆盖了三大 Memory 类别的工程实现。