为什么 Agent 需要记忆

没有记忆的 Agent 就像金鱼——每次对话都从零开始。你告诉它你的名字，下一轮它就忘了。你纠正过它的错误，它下次照犯。

记忆系统让 Agent 能够：

  • 记住用户偏好和历史交互
  • 从过去经验中学习
  • 保持跨会话的上下文一致性
  • 在长任务中不丢失关键信息

记忆类型体系

人类认知科学启发

记忆类型持续时间容量Agent 对应存储方案
工作记忆秒-分钟4±2 项当前对话上下文LLM Context Window
短期记忆分钟-小时有限当前会话摘要内存/Redis
长期记忆天-永久近无限用户偏好、知识向量库 + 关系库
情景记忆事件级大具体交互记录时序向量库

架构总览

┌──────────────────────────────────────────────┐
│                 Agent 主循环                   │
│                                              │
│  ┌─────────┐  ┌─────────┐  ┌──────────────┐ │
│  │ 工作记忆 │  │ 短期记忆 │  │  长期记忆     │ │
│  │ Context │←→│ Session │←→│  Persistent  │ │
│  │ Window  │  │ Summary │  │  Store       │ │
│  └─────────┘  └─────────┘  └──────────────┘ │
│       │            │              │          │
│       │            │     ┌────────┼────────┐ │
│       │            │     │        │        │ │
│       │            │  ┌──┴──┐ ┌───┴───┐ ┌──┴──┐
│       │            │  │语义  │ │情景   │ │程序  │
│       │            │  │记忆  │ │记忆   │ │记忆  │
│       │            │  │向量  │ │时序   │ │图   │
│       │            │  └─────┘ └───────┘ └─────┘
│       │            │
│  ┌────┴────────────┴────────────────────────┐│
│  │            记忆管理器                       ││
│  │  写入 → 压缩 → 检索 → 遗忘 → 巩固          ││
│  └────────────────────────────────────────────┘│
└──────────────────────────────────────────────┘

工作记忆：上下文窗口管理

from dataclasses import dataclass, field
from typing import List, Dict

@dataclass
class WorkingMemory:
    """工作记忆: 管理 LLM 上下文窗口"""

    max_tokens: int = 128000
    reserved_for_response: int = 4096
    messages: List[dict] = field(default_factory=list)

    def add(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        self._evict_if_needed()

    def add_system(self, content: str):
        """系统消息放在最前面，不被驱逐"""
        self.messages.insert(0, {"role": "system", "content": content})

    def _evict_if_needed(self):
        """当超过 token 限制时，驱逐最旧的对话"""
        while self._count_tokens() > self.max_tokens - self.reserved_for_response:
            if len(self.messages) <= 1:
                break
            # 找到第一个非 system 消息并移除
            for i, msg in enumerate(self.messages):
                if msg["role"] != "system":
                    evicted = self.messages.pop(i)
                    # 触发压缩: 被驱逐的消息需要被压缩到短期记忆
                    self._on_evict(evicted)
                    break

    def _count_tokens(self) -> int:
        import tiktoken
        enc = tiktoken.encoding_for_model("gpt-4o")
        return sum(len(enc.encode(m["content"])) for m in self.messages)

    def _on_evict(self, evicted_msg: dict):
        """被驱逐的消息回调，交给短期记忆处理"""
        # 由 MemoryManager 注册
        pass

    def get_context(self) -> List[dict]:
        """获取当前上下文"""
        return self.messages.copy()

短期记忆：会话级摘要

class ShortTermMemory:
    """短期记忆: 当前会话的压缩摘要"""

    def __init__(self):
        self.summary = ""
        self.key_points: List[str] = []
        self.token_budget = 2000  # 摘要不超 2000 token

    async def compress(self, messages: List[dict]) -> str:
        """将多轮对话压缩为摘要"""
        conversation = "\n".join(
            f"{m['role']}: {m['content'][:200]}" for m in messages
        )

        prompt = f"""请将以下对话压缩为简洁摘要，保留:
1. 用户的核心需求
2. 已确定的关键事实
3. 未解决的问题
4. 用户的明确偏好

对话:
{conversation}

摘要:"""

        response = await llm_call(prompt, max_tokens=self.token_budget)
        self.summary = response
        return response

    async def extract_key_points(self, messages: List[dict]) -> List[str]:
        """提取关键信息点"""
        prompt = f"""从以下对话中提取关键信息点，每点一行:
{chr(10).join(m['content'][:200] for m in messages)}

关键信息:"""

        response = await llm_call(prompt, max_tokens=500)
        self.key_points = response.strip().split('\n')
        return self.key_points

    def to_context(self) -> str:
        """转换为可注入上下文的文本"""
        parts = []
        if self.summary:
            parts.append(f"之前的对话摘要:\n{self.summary}")
        if self.key_points:
            parts.append(f"关键信息:\n" + "\n".join(f"- {p}" for p in self.key_points))
        return "\n\n".join(parts)

长期记忆：持久化存储

语义记忆（向量库）

class SemanticMemory:
    """语义记忆: 存储事实和知识，向量检索"""

    def __init__(self, vector_store, embed_model):
        self.store = vector_store
        self.embed = embed_model

    async def remember(self, content: str, metadata: dict = None):
        """记住一条信息"""
        embedding = await self.embed(content)
        await self.store.upsert({
            "id": self._generate_id(content),
            "values": embedding,
            "metadata": {
                "content": content,
                "type": "semantic",
                "created_at": datetime.now().isoformat(),
                **(metadata or {})
            }
        })

    async def recall(self, query: str, top_k: int = 5) -> List[dict]:
        """检索相关记忆"""
        query_embedding = await self.embed(query)
        results = await self.store.query(
            vector=query_embedding,
            top_k=top_k,
            filter={"type": {"$eq": "semantic"}}
        )
        return [
            {
                "content": r.metadata["content"],
                "score": r.score,
                "metadata": r.metadata
            }
            for r in results.matches
        ]

    async def forget(self, content_id: str):
        """主动遗忘"""
        await self.store.delete(ids=[content_id])

情景记忆（时序事件）

class EpisodicMemory:
    """情景记忆: 记录具体交互事件，支持时间检索"""

    def __init__(self, db):
        self.db = db  # PostgreSQL + pgvector

    async def record(self, event: dict):
        """记录一个交互事件"""
        await self.db.execute(
            """
            INSERT INTO episodic_memory
            (event_id, user_id, timestamp, input, output, outcome,
             context_snapshot, embedding)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
            """,
            event["id"], event["user_id"], datetime.now(),
            event["input"], event["output"], event["outcome"],
            json.dumps(event.get("context", {})),
            event.get("embedding")
        )

    async def recall_by_time(self, user_id: str,
                              start: datetime, end: datetime) -> List[dict]:
        """按时间范围回忆"""
        rows = await self.db.fetch(
            """SELECT * FROM episodic_memory
               WHERE user_id = $1 AND timestamp BETWEEN $2 AND $3
               ORDER BY timestamp DESC""",
            user_id, start, end
        )
        return [dict(r) for r in rows]

    async def recall_similar(self, user_id: str,
                              query_embedding: list, top_k: int = 5) -> List[dict]:
        """检索相似的历史事件"""
        rows = await self.db.fetch(
            """SELECT *, embedding <=> $1 as distance
               FROM episodic_memory
               WHERE user_id = $2
               ORDER BY embedding <=> $1
               LIMIT $3""",
            query_embedding, user_id, top_k
        )
        return [dict(r) for r in rows]

程序记忆（知识图谱）

class ProceduralMemory:
    """程序记忆: 存储技能、流程、因果关系（图结构）"""

    def __init__(self, graph_db):
        self.graph = graph_db  # Neo4j

    async def learn_procedure(self, name: str, steps: List[str],
                               triggers: List[str]):
        """学习一个新流程"""
        await self.graph.execute(
            """
            CREATE (p:Procedure {name: $name})
            WITH p
            UNWIND $steps as step_text
            CREATE (s:Step {text: step_text})
            CREATE (p)-[:HAS_STEP]->(s)
            WITH p
            UNWIND $triggers as trigger
            CREATE (t:Trigger {text: trigger})
            CREATE (t)-[:ACTIVATES]->(p)
            """,
            name=name, steps=steps, triggers=triggers
        )

    async def recall_procedure(self, context: str) -> List[dict]:
        """根据上下文检索相关流程"""
        result = await self.graph.execute(
            """
            MATCH (t:Trigger)-[:ACTIVATES]->(p:Procedure)-[:HAS_STEP]->(s:Step)
            WHERE $context CONTAINS t.text
            RETURN p.name as procedure, collect(s.text) as steps
            """,
            context=context
        )
        return [dict(r) for r in result]

记忆管理器：统一调度

class MemoryManager:
    """统一记忆管理器"""

    def __init__(self):
        self.working = WorkingMemory()
        self.short_term = ShortTermMemory()
        self.semantic = SemanticMemory(...)
        self.episodic = EpisodicMemory(...)
        self.procedural = ProceduralMemory(...)
        self.recall_count = 0

    async def think(self, user_input: str, user_id: str) -> str:
        """带记忆的思考过程"""

        # 1. 从长期记忆检索相关信息
        semantic_results = await self.semantic.recall(user_input, top_k=3)
        episodic_results = await self.episodic.recall_similar(
            user_id, await self.embed(user_input), top_k=3
        )
        procedures = await self.procedural.recall_procedure(user_input)

        # 2. 组装上下文
        context_parts = []
        if self.short_term.summary:
            context_parts.append(self.short_term.to_context())
        if semantic_results:
            context_parts.append("相关知识:\n" +
                "\n".join(f"- {r['content']}" for r in semantic_results))
        if episodic_results:
            context_parts.append("历史交互:\n" +
                "\n".join(f"- {r['input']} → {r['output']}" for r in episodic_results))
        if procedures:
            context_parts.append("可用流程:\n" +
                "\n".join(f"- {p['procedure']}: {p['steps']}" for p in procedures))

        context = "\n\n".join(context_parts)

        # 3. 注入工作记忆
        if context:
            self.working.add("system", f"记忆上下文:\n{context[:4000]}")

        self.working.add("user", user_input)

        # 4. 生成回复
        response = await llm_call(self.working.get_context())
        self.working.add("assistant", response)

        # 5. 异步写入记忆
        asyncio.create_task(self._consolidate(user_input, response, user_id))

        return response

    async def _consolidate(self, input: str, output: str, user_id: str):
        """记忆巩固: 筛选重要信息写入长期记忆"""
        # 判断是否值得记住
        importance = await self._assess_importance(input, output)

        if importance > 0.7:
            # 高重要性: 写入语义记忆 + 情景记忆
            await self.semantic.remember(
                f"用户说了: {input}\n助手回答: {output}",
                metadata={"importance": importance, "user_id": user_id}
            )

        # 总是记录情景记忆
        await self.episodic.record({
            "id": str(uuid4()),
            "user_id": user_id,
            "input": input,
            "output": output,
            "outcome": "success",
            "embedding": await self.embed(input),
        })

    async def _assess_importance(self, input: str, output: str) -> float:
        """评估信息重要性 0-1"""
        prompt = f"""评估以下交互的信息重要性(0-1):
输入: {input[:200]}
输出: {output[:200]}

评分标准:
- 0.9+: 用户明确要求记住/涉及核心偏好
- 0.7-0.8: 包含重要事实或决策
- 0.5-0.6: 有一定参考价值
- 0.3-0.4: 闲聊但含少量信息
- 0.0-0.2: 纯闲聊无信息

只输出数字:"""
        return float(await llm_call(prompt, max_tokens=10))

遗忘机制

class ForgettingMechanism:
    """遗忘机制: 防止记忆膨胀"""

    async def decay(self, user_id: str):
        """基于时间的记忆衰减"""
        await self.db.execute(
            """
            UPDATE semantic_memory
            SET importance = importance * 0.95  -- 每天衰减5%
            WHERE user_id = $1
              AND last_accessed < NOW() - INTERVAL '1 day'
            """,
            user_id
        )

        # 删除低重要性记忆
        await self.db.execute(
            """
            DELETE FROM semantic_memory
            WHERE user_id = $1 AND importance < 0.1
            """,
            user_id
        )

    async def deduplicate(self, user_id: str):
        """去重: 合并语义相似的记忆"""
        # 找到相似度 > 0.9 的记忆对
        duplicates = await self.db.fetch(
            """
            SELECT a.id as id_a, b.id as id_b,
                   a.content as content_a, b.content as content_b
            FROM semantic_memory a
            JOIN semantic_memory b ON a.id < b.id
            WHERE a.user_id = $1 AND b.user_id = $1
              AND a.embedding <=> b.embedding < 0.1
            """,
            user_id
        )

        for dup in duplicates:
            # 合并: 保留更详细的，删除另一个
            longer = dup['content_a'] if len(dup['content_a']) > len(dup['content_b']) else dup['content_b']
            shorter_id = dup['id_b'] if len(dup['content_a']) > len(dup['content_b']) else dup['id_a']
            await self.db.execute("DELETE FROM semantic_memory WHERE id = $1", shorter_id)

    async def compress_old_memories(self, user_id: str, days: int = 30):
        """压缩旧记忆: 把多条相关记忆合并为一条摘要"""
        old_memories = await self.db.fetch(
            """
            SELECT content FROM semantic_memory
            WHERE user_id = $1 AND created_at < NOW() - INTERVAL '$2 days'
            ORDER BY created_at
            """,
            user_id, days
        )

        if len(old_memories) < 10:
            return

        # 用 LLM 合并
        memories_text = "\n".join(m['content'] for m in old_memories)
        summary = await llm_call(
            f"将以下记忆合并为简洁摘要，保留关键信息:\n{memories_text}"
        )

        # 删除旧记忆，写入摘要
        await self.db.execute(
            "DELETE FROM semantic_memory WHERE user_id = $1 AND created_at < NOW() - INTERVAL '$2 days'",
            user_id, days
        )
        await self.semantic.remember(summary, metadata={"compressed": True})

MemGPT / Letta 架构参考

MemGPT (现 Letta) 的核心思想是把操作系统的虚拟内存概念引入 LLM：

┌────────────────────────────────────┐
│         LLM Context Window         │ ← "RAM"
│  ┌──────────┐  ┌────────────────┐ │
│  │ System   │  │ 工作记忆 (对话)  │ │
│  │ Prompt   │  │ + 分页加载的记忆  │ │
│  └──────────┘  └────────────────┘ │
├────────────────────────────────────┤
│         外部记忆存储                 │ ← "Disk"
│  ┌──────────┐  ┌────────────────┐ │
│  │ 主记忆    │  │ 归档记忆        │ │
│  │ (活跃)   │  │ (冷存储)       │ │
│  └──────────┘  └────────────────┘ │
└────────────────────────────────────┘

class MemGPTStyleMemory:
    """MemGPT 风格的记忆管理"""

    def __init__(self):
        self.main_context_limit = 8000   # 主上下文 token 限制
        self.working_limit = 4000        # 工作记忆限制
        self.recall_limit = 2000         # 检索记忆限制
        self.archival_limit = 2000       # 归档记忆限制

    async def process(self, user_input: str, user_id: str) -> str:
        # 1. 检查是否需要搜索归档记忆
        if self._needs_recall(user_input):
            results = await self._search_archival(user_input, top_k=3)
            await self._page_in(results)  # 分页加载到主上下文

        # 2. 生成回复
        response = await llm_call(self._build_context())

        # 3. 检查是否需要将工作记忆溢出到归档
        if self._working_tokens() > self.working_limit:
            await self._page_out()  # 将旧对话移到归档

        return response

    async def _page_in(self, memories: list):
        """从归档记忆分页加载到主上下文"""
        for m in memories:
            self.working.add("system", f"[Recalled] {m['content']}")

    async def _page_out(self):
        """将工作记忆中较旧的内容移到归档存储"""
        old_messages = self.working.messages[1:-4]  # 保留首尾
        for msg in old_messages:
            await self.archival.save({
                "content": msg["content"],
                "type": "paged_out",
                "timestamp": datetime.now().isoformat()
            })
        self.working.messages = [
            self.working.messages[0],  # system
            *self.working.messages[-4:]  # 最近4条
        ]

总结

Agent 记忆系统的设计核心是分层和取舍：工作记忆管当前上下文，短期记忆管会话摘要，长期记忆分语义/情景/程序三种。检索要快但不能全召回——token 预算有限。遗忘和压缩比记住更重要——不遗忘的记忆库会变成垃圾场。MemGPT 的分页机制是目前最优雅的解决方案，值得参考但不必照搬，根据你的场景选择合适的记忆层次即可。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。