记忆系统：让智能体拥有持久记忆

Simon included in AI智能体开发实战

2026-07-05 11157 words 23 minutes

Contents

7.1 引言：为什么Agent需要记忆

人类之所以能持续地学习与成长，核心在于我们拥有记忆。记忆让我们能够从过去的经验中提取规律，避免重复犯错，并在新情境中调用相关知识做出更优决策。对于AI智能体（Agent）而言，记忆同样不可或缺。

打个比方：你让一个智能体帮你管理日程。第一天你告诉它"我每周三下午有固定的部门例会"，第二天它就忘了——每次对话都要从零开始。这样的智能体就像一个每天入职的新员工，无法真正成为你的助手，它只是一个无状态的问答工具。

一个具备实用价值的智能体至少需要解决以下问题：

上下文连贯：在单次对话中记住前面说过的话，避免答非所问。
跨会话持久化：记住用户在之前会话中表达的偏好、事实与任务。
知识复用：从历史交互中提取经验，应用到新任务中。
自我反思：能够回顾过去的决策路径，持续改进。

这些能力都建立在记忆系统之上。本章将从记忆类型分类出发，逐步实现短期记忆、长期记忆与记忆管理策略，最终整合为一个完整的记忆增强智能体。

💡 小贴士：什么是"无状态"？ 无状态指服务端不保存任何会话信息。你每次发请求都要带上全部上下文，服务器处理完就忘。大多数网页API默认是无状态的，所以"原生"的大模型API调用，每次都像初次见面。

还需要算一笔经济账：没有记忆系统时，为了让模型"记住"前面说过的话，每轮对话都得把全部历史消息重发一遍，Token消耗随对话长度线性增长，既慢又贵。引入长期记忆后，每轮只需注入少量"被召回的相关片段"，就能在控制成本的同时保留关键信息——这正是记忆系统的工程价值所在。

本章代码依赖：pip install openai chromadb tiktoken

7.2 记忆类型分类

在动手实现之前，先厘清智能体记忆的几种典型类型。借鉴认知科学的划分，我们可以将Agent记忆归纳为以下五类：

记忆类型	作用	存储位置	生命周期
短期记忆	维持当前对话上下文	内存	单次会话
工作记忆	跟踪当前任务状态与中间结果	内存	任务执行期间
长期记忆	跨会话持久化用户事实与偏好	向量数据库	永久
语义记忆	结构化知识库（FAQ、文档）	向量数据库/图数据库	永久
情景记忆	记录历史交互的具体场景	向量数据库	永久（可衰减）

为了让你更好理解，我们用日常工作场景给每种记忆打个比方：

短期记忆就像你桌上的工作台——只放当前正在处理的文件。桌子空间有限，处理完一批就要清掉腾地方。它直接挂在对话消息列表上，受LLM上下文窗口限制。

长期记忆就像你的随身笔记本——会一直跟着你，里面记着"我是谁、喜欢什么、上周做了什么"。它通常依赖向量数据库实现语义检索，即使关机重启也不会丢。

工作记忆像你的草稿纸，关注当前任务执行过程中的中间状态：多步推理的中间变量、工具调用的临时结果等，任务结束就撕掉。

语义记忆偏向"客观知识"，像公司的规章制度手册——产品手册、公司规章，与具体用户无关，人人可查。

情景记忆记录"某时某地发生了什么"，像你的日记本——“用户上周二询问了退货流程”，带有时间与上下文标签。

💡 小贴士：什么是向量数据库？ 向量数据库是一种专门存储"向量"（一串浮点数）并支持"按相似度快速找近邻"的数据库。传统数据库靠精确匹配关键词，向量数据库靠"语义距离"找最相关的内容。常见产品有ChromaDB、Milvus、Qdrant、Pinecone等。本章我们用ChromaDB，因为它轻量、纯本地、对新手最友好。

理解这些分类后，我们开始逐层实现。需要强调的是，工程实践中并不需要把五种都实现一遍，重点是短期记忆+长期记忆的组合，这也是本章的主线。

7.3 短期记忆实现

短期记忆的本质就是对对话消息列表的管理。最朴素的方式是把所有历史消息原封不动地塞进messages数组传给LLM。但当对话变长，这会带来两个问题：一是Token超限导致API报错；二是冗余信息干扰模型注意力，导致回答质量下降。

继续用工作台的比喻：你不能把所有文件都堆在桌上，必须定期清理。常见做法包括：

滑动窗口：只保留最近N轮对话——就像桌上只留最新N份文件。
Token预算：按Token数而非轮数截断，更精确——按"重量"而非"份数"限制。
摘要压缩：把较早的对话压缩成一段摘要，腾出空间——把旧文件归档成一页总结。

💡 小贴士：什么是Token？ Token是LLM处理文本的最小单位，约等于3~4个英文字符或半个汉字。模型上下文窗口按Token计费和限制，例如2000 Token约能装下1500字中文。tiktoken是OpenAI开源的Token计数工具，能在本地快速估算Token数。

下面实现一个支持滑动窗口与Token预算的ConversationBuffer。它的职责是：每次添加新消息后自动裁剪，保证messages数组始终在窗口与预算之内。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75


# conversation_buffer.py
# 短期记忆管理器：滑动窗口 + Token预算控制

import tiktoken
from typing import List, Dict


class ConversationBuffer:
    """对话短期记忆管理器"""

    def __init__(
        self,
        system_prompt: str = "你是一个乐于助人的AI助手。",
        max_messages: int = 20,
        max_tokens: int = 4000,
        model: str = "gpt-5.4",
    ):
        self.system_prompt = system_prompt
        self.max_messages = max_messages  # 最多保留的消息条数（滑动窗口）
        self.max_tokens = max_tokens      # Token预算上限
        self.model = model
        # 自动根据模型选择分词器，找不到则回退到通用编码
        try:
            self.encoding = tiktoken.encoding_for_model(model)
        except KeyError:
            self.encoding = tiktoken.get_encoding("cl100k_base")
        self.messages: List[Dict] = []

    def add_user_message(self, content: str):
        """添加用户消息"""
        self.messages.append({"role": "user", "content": content})
        self._trim()  # 每次添加后自动裁剪

    def add_assistant_message(self, content: str):
        """添加助手消息"""
        self.messages.append({"role": "assistant", "content": content})
        self._trim()

    def add_tool_message(self, content: str, tool_call_id: str):
        """添加工具返回消息（第6章的工具调用会用到）"""
        self.messages.append(
            {"role": "tool", "content": content, "tool_call_id": tool_call_id}
        )
        self._trim()

    def get_messages(self) -> List[Dict]:
        """获取完整消息列表（含system消息），可直接传给OpenAI接口"""
        return [{"role": "system", "content": self.system_prompt}] + self.messages

    def count_tokens(self) -> int:
        """统计当前消息列表的Token数"""
        total = 0
        for msg in self.get_messages():
            # 每条消息约有4个Token的固定开销（角色标记等）
            total += 4
            total += len(self.encoding.encode(msg.get("content", "")))
            if msg.get("role") == "tool":
                total += len(self.encoding.encode(msg.get("tool_call_id", "")))
        return total

    def _trim(self):
        """按消息条数和Token预算裁剪，保证不超限"""
        # 第一步：按条数裁剪（滑动窗口）
        if len(self.messages) > self.max_messages:
            # 保留最近的max_messages条
            self.messages = self.messages[-self.max_messages:]

        # 第二步：按Token预算裁剪
        while self.count_tokens() > self.max_tokens and len(self.messages) > 2:
            # 每次丢弃最早的一条非system消息
            self.messages.pop(0)

    def clear(self):
        """清空记忆（重置会话时调用）"""
        self.messages = []

使用方式非常直观——就像往工作台上放文件，管理器自动帮你控制总量：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14


from conversation_buffer import ConversationBuffer

buffer = ConversationBuffer(
    system_prompt="你是一个私人日程助手。",
    max_messages=10,
    max_tokens=2000,
)

buffer.add_user_message("我每周三下午有部门例会")
buffer.add_assistant_message("好的，已记录。每周三下午14:00-16:00部门例会。")
buffer.add_user_message("那这周三的安排是什么？")

print(buffer.get_messages())
# 输出包含system + 三条对话消息，可直接传给openai chat接口

这套实现的关键点在于_trim方法的双重裁剪：先按条数，再按Token。按Token裁剪时优先丢弃最早的消息，保证最近上下文不丢失。但纯粹的丢弃会丢失早期重要信息——这就要靠长期记忆来补位了。

💡 小贴士：什么是上下文窗口？ 上下文窗口是模型一次能"看到"的Token上限，例如128K窗口约能装下9万字中文。听起来很大，但实际使用中你不会想塞满——窗口越满，推理越慢、越贵，且模型对中段内容的注意力会下降（俗称"中间遗忘"）。所以短期记忆的max_tokens通常只设窗口的1/4到1/2，给系统提示、检索结果和回答预留空间。

7.4 长期记忆实现

短期记忆只在单次会话内有效，一旦会话结束，所有上下文都随之消失。要让智能体"记住"跨会话的事实，必须把信息持久化到外部存储。在LLM应用中，最主流的方案是向量数据库+语义检索：把记忆文本转为Embedding向量存入向量库，查询时用语义相似度召回最相关的记忆。

💡 小贴士：什么是Embedding（向量嵌入）？ Embedding是把一段文本映射成一组浮点数（比如1536维）的过程。语义相近的文本，向量也相近。这样"找相关记忆"就变成了"在几何空间里找最近的点"。OpenAI的text-embedding-3-small模型就能把任意文本转成向量，调用方式和聊天接口类似。

这里我们使用ChromaDB——一个轻量级、本地优先的向量数据库，非常适合入门与原型开发。它不需要单独部署服务，数据直接落在本地文件夹里，重启程序后依然可读。

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131


# long_term_memory.py
# 长期记忆：基于ChromaDB的向量存储与语义检索

import chromadb
import uuid
from datetime import datetime
from typing import List, Dict, Optional
from openai import OpenAI


class LongTermMemory:
    """基于ChromaDB的长期记忆系统"""

    def __init__(
        self,
        collection_name: str = "agent_memory",
        persist_path: str = "./chroma_db",
        embedding_model: str = "text-embedding-3-small",
    ):
        # 初始化ChromaDB持久化客户端：数据写入persist_path目录
        self.client = chromadb.PersistentClient(path=persist_path)
        # 获取或创建集合（类似数据库中的"表"），使用余弦相似度
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"},  # 余弦相似度，适合文本语义
        )
        self.embedding_model = embedding_model
        self.oai = OpenAI()

    def _embed(self, texts: List[str]) -> List[List[float]]:
        """调用OpenAI生成向量"""
        resp = self.oai.embeddings.create(
            model=self.embedding_model,
            input=texts,
        )
        return [d.embedding for d in resp.data]

    def add(
        self,
        content: str,
        memory_type: str = "episodic",
        importance: int = 5,
        metadata: Optional[Dict] = None,
    ) -> str:
        """存入一条记忆

        Args:
            content: 记忆文本内容
            memory_type: 记忆类型，如 factual/episodic/preference
            importance: 重要性评分 1-10
            metadata: 额外的元数据
        Returns:
            记忆的唯一ID
        """
        mem_id = f"mem_{uuid.uuid4().hex[:12]}"
        meta = {
            "type": memory_type,
            "importance": importance,
            "timestamp": datetime.now().isoformat(),
            "content": content,  # 冗余存一份，便于后续按文本排查
        }
        if metadata:
            meta.update(metadata)

        # 生成向量并写入ChromaDB
        embeddings = self._embed([content])
        self.collection.add(
            ids=[mem_id],
            embeddings=embeddings,
            metadatas=[meta],
            documents=[content],  # documents字段方便直接读回原文
        )
        return mem_id

    def search(
        self,
        query: str,
        top_k: int = 5,
        memory_type: Optional[str] = None,
        min_importance: int = 0,
    ) -> List[Dict]:
        """语义检索相关记忆

        Args:
            query: 查询文本
            top_k: 召回数量
            memory_type: 可选的类型过滤
            min_importance: 最低重要性过滤
        Returns:
            记忆列表，按相关度降序
        """
        # 把查询文本也转成向量
        query_embedding = self._embed([query])[0]

        # 构建过滤条件（ChromaDB的where语法）
        where_filter = None
        conditions = []
        if memory_type:
            conditions.append({"type": memory_type})
        if min_importance > 0:
            conditions.append({"importance": {"$gte": min_importance}})
        if len(conditions) == 1:
            where_filter = conditions[0]
        elif len(conditions) > 1:
            where_filter = {"$and": conditions}

        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=top_k,
            where=where_filter,
        )

        # 整理返回结构，方便上层使用
        memories = []
        if results["ids"] and results["ids"][0]:
            for i, mem_id in enumerate(results["ids"][0]):
                memories.append({
                    "id": mem_id,
                    "content": results["documents"][0][i],
                    "metadata": results["metadatas"][0][i],
                    "distance": results["distances"][0][i],  # 距离越小越相关
                })
        return memories

    def delete(self, mem_id: str):
        """按ID删除记忆"""
        self.collection.delete(ids=[mem_id])

    def count(self) -> int:
        """返回记忆总数"""
        return self.collection.count()

💡 小贴士：余弦相似度 vs 欧氏距离 衡量两个向量"有多像"有两种常见方式：欧氏距离看绝对位置差，余弦相似度只看方向夹角。文本语义检索中，我们更关心"方向是否一致"而非"长度差异"，所以用余弦相似度。ChromaDB返回的distance字段是相似度转换后的距离，越小越相关。

💡 小贴士：HNSW是什么？为什么检索那么快？ 代码里metadata={"hnsw:space": "cosine"}中的HNSW（分层导航小世界图）是一种近似最近邻搜索算法。暴力检索要拿查询向量和库里每一条逐一比较，数据量一大就慢；HNSW通过预先构建多层"索引地图"，能在海量向量中毫秒级召回近邻，准确率通常在95%以上。这就是向量数据库比"自己写个for循环算距离"快得多的关键。

下面演示如何使用长期记忆——就像往笔记本里写条目，之后能按"意思"翻出来：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32


from long_term_memory import LongTermMemory

memory = LongTermMemory(persist_path="./agent_memory")

# 存入用户偏好
memory.add(
    content="用户偏好用Python编写代码，代码注释使用中文。",
    memory_type="preference",
    importance=8,
)

# 存入事实
memory.add(
    content="用户所在公司名称是智链科技，部门是算法平台组。",
    memory_type="factual",
    importance=9,
)

# 存入历史交互
memory.add(
    content="2026年7月1日，用户询问了ChromaDB的部署方案，建议使用Docker方式。",
    memory_type="episodic",
    importance=5,
)

# 语义检索
results = memory.search("用户用什么编程语言？", top_k=3)
for r in results:
    print(f"[相关度 {1 - r['distance']:.3f}] {r['content']}")
# 输出示例：
# [相关度 0.812] 用户偏好用Python编写代码，代码注释使用中文。
# [相关度 0.421] 用户所在公司名称是智链科技，部门是算法平台组。

注意上面查询词是"用什么编程语言"，而记忆原文是"偏好用Python编写代码"——两者没有共同关键词，纯靠语义相似度匹配上了。这正是向量检索相比关键词检索的优势。

这套实现的核心要点：

持久化：PersistentClient会把数据写入./agent_memory目录，重启进程后记忆依然存在。
语义检索：查询时同样生成Embedding，用余弦相似度召回，比关键词匹配更鲁棒。
元数据过滤：可以按类型、重要性等维度过滤，提升召回精度。
重要性字段：为后续的记忆衰减与优先级排序预留了接口。

7.5 记忆管理策略

随着交互积累，记忆库会越来越大，问题随之而来：

噪声记忆稀释了真正重要的信息。
Token预算有限，不能把所有召回都塞给LLM。
部分记忆已经过时或失去价值。

继续用笔记本的比喻：如果什么都记、永不丢弃，笔记本很快会被琐碎内容塞满，真正重要的反而翻不到。因此需要引入记忆管理策略，让笔记本保持"鲜活"。本节实现一个带摘要与衰减的管理器，主要包含三个机制：

摘要压缩：定期把多条相关情景记忆压缩成一条摘要——把零散日记整理成月度总结。
重要性衰减：随时间推移降低记忆的importance分数——久不翻阅的条目自动褪色。
优先级排序：检索时综合相关度、重要性、时间新鲜度排序——翻笔记本时优先看最相关、最新、最重要的。

💡 小贴士：为什么用指数衰减而不是直接删？ 人类记忆的遗忘曲线接近指数下降：刚记住的事忘得最快，越久远忘得越慢。代码里new_imp = importance × 0.95^(天数)正是模拟这一规律——前一周内重要性迅速下滑，但不会一刀切清零，给"偶尔被重新检索到而重新激活"留了余地。这种"软遗忘"比硬删除更接近真实认知，也方便事后回溯。

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125


# memory_manager.py
# 记忆管理器：摘要、衰减、优先级排序

from datetime import datetime, timedelta
from typing import List, Dict
from openai import OpenAI
from long_term_memory import LongTermMemory


class MemoryManager:
    """记忆生命周期管理"""

    def __init__(self, memory: LongTermMemory, llm_model: str = "gpt-5.4"):
        self.memory = memory
        self.llm_model = llm_model
        self.oai = OpenAI()

    def summarize_and_consolidate(self, query: str, top_k: int = 10) -> str:
        """召回一批记忆并让LLM生成压缩摘要

        适用于：情景记忆较多且高度重复时，把它们融合成一条更高层级的记忆。
        """
        memories = self.memory.search(query, top_k=top_k)
        if not memories:
            return ""

        # 拼接所有记忆内容，带上时间戳方便LLM理解时序
        mem_text = "\n".join(
            f"- [{m['metadata'].get('timestamp', '')}] {m['content']}"
            for m in memories
        )

        prompt = (
            "请把以下多条历史记忆压缩为一段简洁的事实摘要，"
            "去除重复信息，保留关键事实与偏好。只输出摘要本身：\n\n"
            f"{mem_text}"
        )

        # 用LLM做摘要
        resp = self.oai.chat.completions.create(
            model=self.llm_model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3,
        )
        summary = resp.choices[0].message.content.strip()

        # 把摘要作为新的语义记忆存入，原记忆可选择保留或删除
        self.memory.add(
            content=summary,
            memory_type="semantic",
            importance=7,
            metadata={"consolidated": True},
        )
        return summary

    def decay_importance(self, days: float = 1.0, decay_rate: float = 0.95):
        """按时间衰减所有记忆的重要性

        每经过 days 天，importance 乘以 decay_rate。
        低于阈值的记忆视为可遗忘。
        """
        all_data = self.memory.collection.get()
        threshold = 1  # 低于此分数视为可遗忘
        now = datetime.now()

        for i, meta in enumerate(all_data["metadatas"]):
            ts_str = meta.get("timestamp")
            if not ts_str:
                continue
            try:
                ts = datetime.fromisoformat(ts_str)
            except ValueError:
                continue
            elapsed_days = (now - ts).total_seconds() / 86400
            if elapsed_days < days:
                continue
            current_imp = meta.get("importance", 5)
            # 计算衰减后的分数（指数衰减）
            periods = elapsed_days / days
            new_imp = max(0, current_imp * (decay_rate ** periods))
            # 更新元数据：保留原字段，只改importance
            self.memory.collection.update(
                ids=[all_data["ids"][i]],
                metadatas=[{**meta, "importance": round(new_imp, 2)}],
            )

    def rank_memories(self, memories: List[Dict]) -> List[Dict]:
        """对召回的记忆做优先级排序

        综合三个维度：
          - 相关度（distance越小越相关）
          - 重要性
          - 时间新鲜度
        """
        now = datetime.now()
        scored = []
        for m in memories:
            relevance = 1 - m["distance"]  # 距离转相似度
            importance = m["metadata"].get("importance", 5) / 10.0  # 归一化到0~1
            ts_str = m["metadata"].get("timestamp", "")
            try:
                ts = datetime.fromisoformat(ts_str)
                age_days = (now - ts).total_seconds() / 86400
                freshness = 1.0 / (1.0 + age_days / 7)  # 7天半衰
            except (ValueError, TypeError):
                freshness = 0.5

            # 加权综合分：相关度0.5 + 重要性0.3 + 新鲜度0.2
            score = 0.5 * relevance + 0.3 * importance + 0.2 * freshness
            scored.append((score, m))

        # 按综合分降序
        scored.sort(key=lambda x: x[0], reverse=True)
        return [m for _, m in scored]

    def forget(self, threshold: float = 1.0) -> int:
        """遗忘重要性低于阈值的记忆，返回删除条数"""
        all_data = self.memory.collection.get()
        to_delete = [
            mid for mid, meta in zip(all_data["ids"], all_data["metadatas"])
            if meta.get("importance", 5) < threshold
        ]
        if to_delete:
            self.memory.collection.delete(ids=to_delete)
        return len(to_delete)

使用示例：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22


from long_term_memory import LongTermMemory
from memory_manager import MemoryManager

memory = LongTermMemory(persist_path="./agent_memory")
manager = MemoryManager(memory)

# 假设已存入大量情景记忆，现在做摘要合并
summary = manager.summarize_and_consolidate("用户的项目背景", top_k=8)
print("合并后的摘要：", summary)

# 衰减一次旧记忆的重要性
manager.decay_importance(days=1, decay_rate=0.95)

# 检索并按优先级排序
results = memory.search("用户的技术栈", top_k=10)
ranked = manager.rank_memories(results)
for m in ranked[:3]:
    print(m["content"])

# 清理掉已不值得保留的记忆
deleted = manager.forget(threshold=1.0)
print(f"已遗忘 {deleted} 条低价值记忆")

这套管理机制让记忆库保持"鲜活"——重要的、近期的、相关的记忆会被优先呈现给LLM，冗余与过时的则被压缩或清除。这正是让智能体在长期使用中保持高质量回答的关键。

7.6 完整的记忆增强Agent

现在把短期记忆、长期记忆、记忆管理整合为一个完整的MemoryAgent。它的核心流程是：

收到用户输入。
用长期记忆做语义检索，召回相关历史。
把召回结果作为背景信息注入短期记忆。
调用LLM生成回答。
判断本轮是否有值得长期保存的信息，若有则存入长期记忆。

用工作台+笔记本的比喻串起来就是：先翻笔记本找到相关旧条目，把要点抄到工作台上做参考，然后基于工作台上的全部信息思考回答，最后判断这轮有没有值得写回笔记本的新内容。

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105


# memory_agent.py
# 完整的记忆增强智能体

from typing import Optional
from openai import OpenAI
from conversation_buffer import ConversationBuffer
from long_term_memory import LongTermMemory
from memory_manager import MemoryManager


class MemoryAgent:
    """整合短期与长期记忆的智能体"""

    def __init__(
        self,
        system_prompt: str = "你是一个具备持久记忆的个人助手。",
        llm_model: str = "gpt-5.4",
        persist_path: str = "./agent_memory",
        recall_top_k: int = 5,
    ):
        self.llm_model = llm_model
        self.oai = OpenAI()
        # 短期记忆（工作台）
        self.buffer = ConversationBuffer(
            system_prompt=system_prompt,
            max_messages=20,
            max_tokens=4000,
            model=llm_model,
        )
        # 长期记忆（笔记本）与管理器
        self.memory = LongTermMemory(persist_path=persist_path)
        self.manager = MemoryManager(self.memory, llm_model=llm_model)
        self.recall_top_k = recall_top_k

    def chat(self, user_input: str) -> str:
        """一次完整的对话回合"""
        # 第一步：从长期记忆中召回相关内容
        recalled = self.memory.search(user_input, top_k=self.recall_top_k)
        ranked = self.manager.rank_memories(recalled)
        if ranked:
            memory_block = "\n".join(f"- {m['content']}" for m in ranked[:3])
            recall_msg = (
                f"[以下是与你相关的历史记忆，可参考但不必逐字复述：]\n{memory_block}"
            )
            # 临时注入到短期记忆（用user角色，作为背景提示）
            self.buffer.messages.append({"role": "user", "content": recall_msg})
            self.buffer.messages.append(
                {"role": "assistant", "content": "好的，我已回顾相关记忆。"}
            )

        # 第二步：把真实用户输入加入短期记忆
        self.buffer.add_user_message(user_input)

        # 第三步：调用LLM生成回答
        resp = self.oai.chat.completions.create(
            model=self.llm_model,
            messages=self.buffer.get_messages(),
            temperature=0.7,
        )
        reply = resp.choices[0].message.content.strip()

        # 第四步：把回答加入短期记忆
        self.buffer.add_assistant_message(reply)

        # 第五步：判断是否值得存入长期记忆
        self._maybe_remember(user_input, reply)

        return reply

    def _maybe_remember(self, user_input: str, reply: str):
        """判断本轮对话是否包含值得长期保存的信息

        简化策略：用LLM判断是否为"用户偏好/事实/重要任务"，
        若是则抽取后存入长期记忆。
        """
        judge_prompt = (
            "判断以下用户输入是否包含需要长期记住的偏好、事实或重要任务。"
            "如果是，请用一句话抽取核心事实；如果不是，只回答 NONE。\n\n"
            f"用户输入：{user_input}"
        )
        resp = self.oai.chat.completions.create(
            model=self.llm_model,
            messages=[{"role": "user", "content": judge_prompt}],
            temperature=0,
        )
        result = resp.choices[0].message.content.strip()

        if result and result.upper() != "NONE":
            # 根据输入内容粗略分类
            memory_type = "preference" if "喜欢" in user_input or "偏好" in user_input else "factual"
            self.memory.add(
                content=result,
                memory_type=memory_type,
                importance=7,
            )

    def reset_session(self):
        """重置短期记忆，保留长期记忆"""
        self.buffer.clear()

    def maintenance(self):
        """定期维护：摘要合并、衰减、遗忘"""
        # 这里用最近用户输入作为摘要主题的简化处理
        self.manager.decay_importance(days=1, decay_rate=0.95)
        self.manager.forget(threshold=1.0)

完整运行示例：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19


from memory_agent import MemoryAgent

agent = MemoryAgent(
    system_prompt="你是用户的私人技术助手，要善于记住用户说过的偏好与背景。",
    persist_path="./agent_memory",
)

# 第一次会话
print("=== 会话 1 ===")
print(agent.chat("我主要用Python做后端开发，代码注释习惯用中文。"))
print(agent.chat("帮我推荐一个轻量级的向量数据库。"))
# 输出：可能推荐ChromaDB

# 模拟新会话：重置短期记忆，但长期记忆保留
agent.reset_session()

print("=== 会话 2（新会话）===")
print(agent.chat("帮我写一个读取JSON文件的函数。"))
# 智能体会"记得"用户偏好Python与中文注释，直接给出符合风格的代码

第二次会话中，即便短期记忆已被清空，智能体依然能从长期记忆中召回"用户偏好Python+中文注释"这一事实，从而给出符合用户习惯的回答。这就是记忆系统带来的真正价值——跨会话的个性化与连续性。

需要提醒的是，_maybe_remember每轮都会额外调用一次LLM做"是否值得记忆"的判断，这会让单次对话的API开销翻倍。生产环境中可用更轻量的策略替代：比如先用关键词规则做粗筛（出现"我叫/我喜欢/我的公司"等触发词才进入判断），或改用更便宜的小模型专门做记忆抽取，把主模型的算力留给真正的回答生成。换言之，记忆系统的设计本身就是"准确率"与"成本"之间的权衡。

7.7 记忆系统最佳实践

实现记忆系统容易，但用好它是一门工程艺术。以下几条经验值得记牢。

1. 何时存入记忆？

不要把每轮对话都塞进长期记忆，那样只会制造噪声。建议设置一个"记忆门槛"——只有当对话包含以下信息时才持久化：

用户明确表达的偏好（“我喜欢……"、“请不要……"）
客观事实（姓名、公司、项目背景）
重要的任务或待办（“下周三前完成报告”）
工具调用产生的关键结论（例如检索到的最终答案）

对于寒暄、追问、确认等无信息量的话，应当丢弃。

2. 存什么内容？

存的是"提炼后的事实”，而非"原始对话原文”。原始对话冗长且重复，直接入库会稀释检索质量。最佳做法是用LLM先抽取一句核心事实再存入，就像_maybe_remember方法所做的那样。

3. 检索策略怎么选？

简单问答：单轮向量召回即可，top_k=3~5。
多轮复杂任务：结合元数据过滤（按类型、按时间窗口）。
长期跟踪某主题：定期调用summarize_and_consolidate做摘要合并，避免情景记忆爆炸式增长。

4. 记忆与RAG的区别？

记忆偏"用户维度"，记录的是与该用户交互中产生的私有信息；RAG偏"知识维度"，检索的是公共文档与知识库。两者在工程实现上高度相似（都是向量检索），但数据来源与更新策略不同。下一章我们会深入讨论RAG。

5. 隐私与安全

长期记忆里往往包含用户敏感信息，务必注意：

数据库文件做好访问控制。
提供"遗忘某条记忆"与"清空全部记忆"的接口。
在多租户场景下，为每个用户建立独立的collection或用metadata强隔离。

6. 成本与性能的平衡

记忆系统每轮对话至少多触发一次Embedding调用，_maybe_remember还会多一次LLM调用，长期累积成本不可忽视。优化方向有四：一是给Embedding做本地缓存（同一句话不重复向量化）；二是用更便宜的embedding模型（如text-embedding-3-small就比large便宜数十倍）；三是控制recall_top_k，召回3~5条通常足够，再多边际收益递减；四是在低频时段批量跑maintenance()做摘要与遗忘，避免影响实时对话延迟。记住：记忆系统是"长期资产"，要把它的开销摊到长期使用中去看，而不是单轮计较。

7.8 小结

本章围绕"让智能体拥有持久记忆"这一目标，构建了完整的记忆架构：

短期记忆用ConversationBuffer管理对话上下文，通过滑动窗口与Token预算防止上下文溢出，就像工作台上只放当前要处理的文件。
长期记忆用ChromaDB存储Embedding向量，支持跨会话语义检索，让智能体真正"记住"用户，就像一本随身携带的笔记本。
记忆管理通过摘要压缩、重要性衰减、优先级排序与遗忘机制，保持记忆库的高质量与低噪声。
完整Agent将三者整合为MemoryAgent，演示了"召回-注入-生成-记忆"的闭环流程。

核心认知是：记忆不是简单的存储，而是一个有生命周期的系统。好的记忆系统要让重要信息浮现、让冗余信息沉淀，这才是一个能长期陪伴用户的智能体所必需的。

最后用一个画面收束本章：理想的智能体就像一位和你共事多年的老同事——你说上半句它就懂下半句，记得你的偏好与忌讳，会在合适时机主动想起过往约定，却也不会把每次寒暄都当回事。这种"分寸感"正是记忆管理要解决的核心问题，也是本章代码背后真正想传达的设计哲学（不妨回头看看前面的裁剪与衰减逻辑，每一处都在为这种分寸感服务）。

7.9 预告：第8章 RAG增强检索

记忆系统解决了"记住用户"的问题，但智能体还需要"记住世界"——即访问海量的外部知识。第8章我们将深入RAG（Retrieval-Augmented Generation）增强检索技术，讨论文档切分、多路召回、重排序、混合检索等主题，让智能体在面对专业领域问题时，能够从知识库中精准提取依据，生成有据可查的回答。记忆与RAG，一个是"内化的经验"，一个是"外化的知识"，二者共同构成智能体的认知基础。