上一篇文章讲了 Tools，给 Agent 安上了"手"。今天咱们聊聊 Memory，给 Agent 装上"脑子"。核心就是解决 LLM 的"失忆症"问题——短期记忆管上下文，长期记忆管跨会话。记住一件事：别一上来就搞向量库，简单对话用滑动窗口就够了，够用就行。

你有没有发现一个很尴尬的事情？

你跟 LLM 说"我叫张三"，它会说"你好张三"。

五分钟后再问"我叫什么？“，它会说"不知道，你还没告诉我”。

这就是 LLM 的"失忆症"——它是无状态的。

LLM 的 Context Window 确实能存储对话历史，但问题在于：

1. 长度限制：大部分模型的 Context Window 只有 4k-128k tokens，对话一长就爆了
2. 跨会话丢失：你关掉对话，下次打开，LLM 完全不记得你说过什么
3. 检索困难：历史对话全都混在一起，想找半年前说的话，大海捞针

这些问题，Memory 都能解决。

---

Memory 有哪几类？

Memory 这个东西，从不同角度看，分法不一样。

┌─────────────────────────────────────────┐
  系统分析维度（论文元框架） 
          
  形式(Form) / 功能(Function) / 动态(Dynamics) 
  → 回答："怎么评价一个记忆系统的设计？"      
├─────────────────────────────────────────┤
  认知内容维度（4类记忆）  
            
  工作 / 情景 / 语义 / 程序                 
  → 回答："Agent 应该记住什么类型的东西？"     
├─────────────────────────────────────────┤
  工程时间维度（2类记忆）  
            
  短期记忆 / 长期记忆                        
  → 回答："数据存在哪里、存多久？"            
└─────────────────────────────────────────┘

从时间维度看

短期记忆：当前对话的上下文，存在 LLM 的 Context Window 里。

就像人的"工作记忆"——你正在想的事情，暂时存在脑子里，过一会儿就忘了。

优点：响应快，直接用就行。

缺点：有长度限制，跨会话就丢了。

长期记忆：跨会话持久化，存储在外部数据库中。

就像人的"长期记忆"——你记得去年去过的景区、十年前见过的人。

优点：理论上无限，跨会话保留。

缺点：需要额外存储，检索有开销。

从认知内容维度看

工作记忆（Working Memory）：当前对话上下文、激活的事实、中间推理步骤。

就像你做数学题时，脑子里正在算的那一步。

情景记忆（Episodic Memory）：具体对话事件、用户行为历史、时间线。

就像你记得"上周三和客户开会，客户提了三个需求"。

语义记忆（Semantic Memory）：用户画像、领域知识、概念关系。

就像你知道"用户是技术团队负责人"、“行业是电商”。

程序记忆（Procedural Memory）：工具调用流程、审批规则、标准操作 SOP。

就像你记得"遇到退款申请，先查订单，再算金额，最后发邮件"。

---

短期记忆怎么实现？

短期记忆的实现方式有四种：

1. 全量保留

什么都不做，把所有历史对话直接拼接到 Prompt 中。

class NaiveMemory:
    def __init__(self):
        self.messages = []
    
    def add(self, role, content):
        self.messages.append({"role": role, "content": content})
    
    def get(self):
	    # 原封不动返回所有历史
        return self.messages

适用场景：极短对话、测试阶段、Demo 演示。

优点：简单粗暴，什么都不用做。

缺点：对话稍长就会崩，Token 费用高。

2. 滑动窗口

只保留最近 K 轮对话，丢弃更早的。

class SlidingWindowMemory:
    def __init__(self, max_rounds=3):
        self.max_rounds = max_rounds  # 保留的轮数（1轮=1问1答）
        self.history = []

    def add(self, role: str, content: str):
        self.history.append({"role": role, "content": content})
        # 维持窗口大小，超过则切除最旧的消息
        # max_rounds * 2 代表 max_rounds 轮的问答消息数
        if len(self.history) > self.max_rounds * 2:
            self.history = self.history[-self.max_rounds * 2:]

    def get_prompt(self):
        return self.history

适用场景：需要限制 Token 成本、对话时长可控的场景。

优点：实现简单，能控制 Token 用量。

缺点：记忆力"金鱼化"，关键信息丢了就找不回。

比如你跟 Agent 说"我下周要出差，帮我安排行程"，然后聊了 10 轮别的，最后问"下周去哪出差？"——它忘了，因为你刚才说的内容被滑动窗口切掉了。

3. Token 感知截断

设定一个 Token 上限，每次获取记忆时，如果超限，保留高重要性消息，从最旧的消息开始逐条删除。

怎么评分？ 有几种策略：

• 位置评分：最近的消息得分高
• 关键词评分：包含用户明确提到的信息（如"我姓王"）得分高
• 情感评分：用户表达强烈意见（如"我讨厌这个"）得分高
• 引用评分：被后续消息引用过的得分高

import tiktoken

class ProTokenAwareMemory:
    def __init__(self, max_tokens=4000, model="gpt-4o"):
        self.messages = []
        self.max_tokens = max_tokens
        self.encoding = tiktoken.encoding_for_model(model)
        self._seq_counter = 0  # 显式的全局递增序列号，替代 id(m)

    def count(self, text):
        return len(self.encoding.encode(text))

    def add(self, role, content, importance=1):
        self._seq_counter += 1
        msg = {
            "role": role,
            "content": content,
            "importance": importance,  # 1=普通, 2=重要, 3=系统指令
            "tokens": self.count(content) + 4, # +4 是格式开销
            "seq": self._seq_counter   # 明确记录插入的时间顺序
        }
        self.messages.append(msg)
        self._truncate()

    def _truncate(self):
        total = sum(m["tokens"] for m in self.messages)
        
        while total > self.max_tokens and len(self.messages) > 2:
            # 1. 绝不删除第一条和最后一条
            candidates = self.messages[1:-1]
            if not candidates:
                # 如果中间部分为空，说明是最后一条消息太长导致超限
                # 此时只能忍痛截断最后一条消息的内容，或者丢弃（根据业务决定）
                # 这里采用简单丢弃策略，防止死循环
                if total > self.max_tokens and len(self.messages) > 1:
                    removed = self.messages.pop(1) # 丢弃第2条
                    total -= removed["tokens"]
                continue

            # 2. 找到优先级最低（importance最小）且最旧（seq最小）的消息
            to_remove = min(candidates, key=lambda m: (m["importance"], m["seq"]))
            
            # 3. 使用索引删除，比 list.remove() 更高效且安全
            idx = self.messages.index(to_remove)
            self.messages.pop(idx)
            
            # 4. 重新计算总 Token
            total = sum(m["tokens"] for m in self.messages)

    def get(self):
        # 返回标准格式，去掉内部字段
        return [{"role": m["role"], "content": m["content"]} for m in self.messages]

适用场景：生产环境、需要平衡记忆容量和成本的场景。

优点：比滑动窗口更智能，能保留重要信息。

缺点：需要设计评分策略，有一定复杂度。

4. 总结压缩

当对话达到一定长度，调用一次 LLM，把前面的对话总结成一段摘要。下次发送 Prompt 时，用"历史摘要+近期对话"代替全量历史。

import tiktoken

class UltimateSummaryMemory:
    def __init__(self, llm_client, max_raw=6, max_summary_tokens=500, model="gpt-4"):
        self.llm = llm_client
        self.summary = ""                # 压缩后的历史摘要
        self.recent = []                 # 最近未压缩的原始消息
        self.max_raw = max_raw           # 触发压缩的近期消息阈值
        self.max_summary_tokens = max_summary_tokens
        
        # 初始化 Token 计算器
        try:
            self.encoder = tiktoken.encoding_for_model(model)
        except KeyError:
            self.encoder = tiktoken.get_encoding("cl100k_base") # 默认回退

    def _count_tokens(self, text):
        return len(self.encoder.encode(text))

    def _safe_truncate(self, text, max_tokens):
        """安全的 Token 截断：优先保留完整句子"""
        tokens = self.encoder.encode(text)
        if len(tokens) <= max_tokens:
            return text
        
        # 预留 3 个 token 给 "..."
        truncated_tokens = tokens[:max_tokens - 3]
        decoded_text = self.encoder.decode(truncated_tokens)
        
        # 寻找最近的句子结束符，避免语义撕裂
        last_punct = max(
            decoded_text.rfind('。'), 
            decoded_text.rfind('. '), 
            decoded_text.rfind('？'), 
            decoded_text.rfind('? ')
        )
        # 如果在尾部的 30% 范围内找到了标点，就在这里切断
        if last_punct > len(decoded_text) * 0.7:
            decoded_text = decoded_text[:last_punct + 1]
            
        return decoded_text + "..."

    def add(self, role, content):
        self.recent.append({"role": role, "content": content})
        if len(self.recent) > self.max_raw:
            self._compress()

    def _compress(self):
        # 1. 剥离需要被压缩的历史（保留最近 2 轮原貌）
        to_summarize = self.recent[:-2]
        self.recent = self.recent[-2:]

        # 2. 将历史消息格式化为文本
        text = "\n".join([f"{m['role']}: {m['content']}" for m in to_summarize])

        # 3. 防御一：输入端防爆（Prompt 不能无限长）
        # 粗略估算 Prompt 模板本身占 100 Token，预留空间
        max_input_tokens = 3000 - 100  
        
        # 如果原始对话太长，从最旧的开始丢弃（滑动窗口过滤）
        while self._count_tokens(text) > max_input_tokens and len(to_summarize) > 1:
            to_summarize.pop(0) # 移除最旧的一条
            text = "\n".join([f"{m['role']}: {m['content']}" for m in to_summarize])

        # 4. 动态构建 Prompt（核心优化：把压力给到 LLM，而不是工程截断）
        summary_tokens = self._count_tokens(self.summary)
        
        # 动态计算要求 LLM 输出的长度
        # 如果旧摘要已经很长了，就要求 LLM 这次极简输出，强行压缩
        if summary_tokens > self.max_summary_tokens * 0.6:
            target_words = "极简提炼(不超过50字)"
        else:
            target_words = "简明扼要的摘要(不超过200字)"

        prompt = f"""你是一个记忆压缩助手。
已有的历史摘要：{self.summary}
新发生的对话：\n{text}

请结合已有摘要和新对话，生成一份全新的历史摘要。
要求：保留所有关键事实、用户偏好和重要指令，丢弃闲聊和无用细节。
请生成{target_words}。"""

        # 5. 调用大模型
        new_summary = self.llm.generate(prompt)

        # 6. 防御二：输出端兜底（万一 LLM 失控，强行安全截断）
        self.summary = self._safe_truncate(new_summary, self.max_summary_tokens)

    def get(self):
        """返回最终给 LLM 的 Prompt 列表"""
        result = []
        if self.summary:
            result.append({"role": "system", "content": f"[历史摘要] {self.summary}"})
        result.extend([{"role": m["role"], "content": m["content"]} for m in self.recent])
        return result

适用场景：长会话、复杂任务、需要长期追踪的场景。

优点：既保留了历史信息，又控制了 Token 用量。

缺点：有信息损失风险，需要额外调用 LLM 成本。

---

长期记忆怎么实现？

长期记忆有三种主要实现方式：

1. 键值存储

从对话中提取结构化属性（如姓名、年龄、饮食禁忌），以 Key-Value 形式存入数据库。

"""
长期记忆 - 键值存储（结构化画像）
用 Key-Value 存储用户结构化信息，支持精确查询和关键词搜索
"""

import json
from pathlib import Path


class KeyValueMemory:
    """
    存储结构: { user_id: { category: { key: value } } }
    例: { "user_001": { "profile": {"name": "张三"}, "prefs": {"lang": "Python"} } }
    """

    def __init__(self, path="kv_memory.json"):
        self.path = Path(path)
        self.store = {}
        if self.path.exists():
            try:
                with open(self.path, "r", encoding="utf-8") as f:
                    self.store = json.load(f)
            except (json.JSONDecodeError, OSError) as e:
                # 文件损坏时：记录警告，从空状态启动（生产环境应报警/备份恢复）
                print(f"  [KV存储] 加载失败({e})，从空状态启动")

    def _save(self):
        # 原子写入：先写临时文件再 rename，避免写入中断导致文件损坏
        tmp = self.path.with_suffix(".tmp")
        with open(tmp, "w", encoding="utf-8") as f:
            json.dump(self.store, f, ensure_ascii=False, indent=2)
        tmp.replace(self.path)

    def save(self, user_id, category, key, value):
        """保存一条结构化记忆"""
        self.store.setdefault(user_id, {}).setdefault(category, {})[key] = value
        self._save()

    def get(self, user_id, category, key, default=None):
        """精确查询一条记忆"""
        return self.store.get(user_id, {}).get(category, {}).get(key, default)

    def get_all(self, user_id):
        """获取某用户全部记忆"""
        return self.store.get(user_id, {})

    def delete(self, user_id, category, key):
        """删除一条记忆，空分类和空用户自动清理"""
        if user_id not in self.store:
            return
        if category not in self.store[user_id]:
            return
        self.store[user_id][category].pop(key, None)
        # 清理空分类
        if not self.store[user_id][category]:
            del self.store[user_id][category]
        # 清理空用户（防止长期运行内存泄漏）
        if not self.store[user_id]:
            del self.store[user_id]
        self._save()

    def search(self, user_id, keyword):
        """按关键词搜索（简单字符串匹配，遍历所有键值对）"""
        results = []
        for cat, kvs in self.store.get(user_id, {}).items():
            for k, v in kvs.items():
                if keyword.lower() in str(k).lower() or keyword.lower() in str(v).lower():
                    results.append({"category": cat, "key": k, "value": v})
        return results

    def build_prompt(self, user_id):
        """将用户记忆格式化为 LLM 可用的上下文提示词"""
        data = self.get_all(user_id)
        if not data:
            return ""
        lines = ["[用户记忆画像]"]
        for cat, kvs in data.items():
            lines.append(f"## {cat}")
            for k, v in kvs.items():
                lines.append(f"- {k}: {v}")
        return "\\n".join(lines)


# 使用
m = KeyValueMemory("/tmp/kv_memory.json")
m.save("user_001", "profile", "name", "张三")
m.save("user_001", "prefs", "framework", "FastAPI")
print(m.get("user_001", "prefs", "framework"))    # "FastAPI"
print(m.search("user_001", "FastAPI"))             # [{"category":"prefs",...}]
print(m.build_prompt("user_001"))                   # 格式化上下文

适用场景：用户偏好设置、个人档案、明确的配置项。

优点：极其精准，查询快，存储高效。

缺点：只能存预先定义好的字段，灵活性差。

比如你定义了字段表：name, age, preferences。突然用户说"我之前跟你说过我是个程序员"，你的 KV 存储就尴尬了——你表里没这个字段。

2. 向量存储

把文本通过 Embedding 模型转化为高维向量，存入向量数据库。

检索时，计算问题向量和所有记忆向量的余弦相似度，返回得分最高的 Top-K 条记忆。

"""
长期记忆 - 向量存储（语义相似检索）
使用 OpenAI Embedding API 生成向量 + ChromaDB 存储和检索
"""

import uuid
import chromadb
from openai import OpenAI


class VectorMemory:
    def __init__(self, openai_client, embedding_model="text-embedding-3-small",
                 persist_dir="./chroma_memory"):
        """
        openai_client:    OpenAI 客户端实例
        embedding_model:  Embedding 模型（text-embedding-3-small: 1536维，性价比最高）
        persist_dir:      ChromaDB 数据持久化目录
        """
        self.openai_client = openai_client
        self.embedding_model = embedding_model
        self._embed_cache = {}  # 查询向量缓存，避免同一 query 重复调 API

        # ChromaDB 持久化客户端：数据自动落盘，重启后自动加载
        self.chroma = chromadb.PersistentClient(path=persist_dir)
        self.collection = self.chroma.get_or_create_collection(
            name="agent_memory",
            metadata={"hnsw:space": "cosine"},  # 余弦距离度量
        )

    def _embed(self, text):
        """调用 OpenAI Embedding API 获取文本向量"""
        if text in self._embed_cache:
            return self._embed_cache[text]
        resp = self.openai_client.embeddings.create(
            model=self.embedding_model,
            input=text,
        )
        vector = resp.data[0].embedding
        self._embed_cache[text] = vector
        return vector

    def add(self, content, metadata=None):
        """
        添加一条记忆：生成向量 → 存入 ChromaDB
        ChromaDB 同时存储原文、向量、元数据，检索时可直接返回原文
        """
        entry_id = f"mem_{uuid.uuid4().hex[:8]}"
        embedding = self._embed(content)
        self.collection.add(
            ids=[entry_id],
            documents=[content],
            embeddings=[embedding],
            metadatas=[metadata or {}],
        )
        return entry_id

    def search(self, query, top_k=5, threshold=0.5):
        """
        语义检索：将 query 转为向量 → ChromaDB HNSW 索引检索 → 返回最相似的记忆
        ChromaDB 内部使用 HNSW 算法，比暴力遍历快几个数量级
        """
        if self.collection.count() == 0:
            return []

        q_embedding = self._embed(query)
        n_results = min(top_k, self.collection.count())  # 请求数不超过集合大小
        results = self.collection.query(
            query_embeddings=[q_embedding],
            n_results=n_results,
            include=["documents", "metadatas", "distances"],
        )

        scored = []
        for i, doc in enumerate(results["documents"][0]):
            # ChromaDB 返回 cosine distance = 1 - cosine_similarity
            score = round(1 - results["distances"][0][i], 4)
            if score >= threshold:
                scored.append({
                    "id": results["ids"][0][i],
                    "content": doc,
                    "score": score,
                    "metadata": results["metadatas"][0][i],
                })
        # ChromaDB 已按 distance 升序返回（即 similarity 降序），无需再排序
        return scored

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

    def build_prompt(self, query, top_k=3):
        """检索相关记忆并格式化为 LLM 上下文"""
        results = self.search(query, top_k=top_k)
        if not results:
            return ""
        lines = ["[相关记忆]"]
        for i, r in enumerate(results, 1):
            lines.append(f"{i}. {r['content']} (相关度: {r['score']})")
        return "\n".join(lines)


# 使用
from openai import OpenAI
client = OpenAI()

m = VectorMemory(openai_client=client, persist_dir="/tmp/chroma_memory")
m.add("用户常用 FastAPI 搭建 API，数据库用 PostgreSQL", {"source": "对话"})
m.add("部署环境是 Docker + K8s，CI 用 GitHub Actions", {"source": "对话"})
print(m.search("容器化部署方案", top_k=2))
print(m.build_prompt("怎么部署服务？", top_k=2))

适用场景：非结构化的对话记录、经验总结、知识片段。

优点：能理解"语义"，不用精准匹配关键词，灵活性强。

缺点：存在"幻觉检索"——语义相近但不相关的结果可能被检索出来。

比如用户问"我叫什么？“，可能检索出"你之前的客户叫张三”，然后 Agent 误以为你叫张三。

3. 混合存储

结合方式 1 和方式 2。结构化的明确信息用 KV 存，非结构化的经验和知识用向量存。

"""
长期记忆 - 混合存储（键值 + 向量，生产推荐）
SQLite 存结构化数据，ChromaDB 存向量数据，统一检索接口
"""

import tiktoken
from openai import OpenAI

# 复用上文的 KeyValueMemory 和 VectorMemory
# 实际项目: from .kv_memory import KeyValueMemory
#           from .vector_memory import VectorMemory


class HybridMemory:
    def __init__(self, openai_client, llm_model="gpt-4o-mini",
                 embedding_model="text-embedding-3-small",
                 kv_db_path="agent_kv.db",
                 chroma_persist_dir="./chroma_memory",
                 max_context_tokens=4000):
        """
        openai_client:     OpenAI 客户端实例
        llm_model:         用于 auto_save 智能路由的模型
        embedding_model:   用于向量存储的 Embedding 模型
        kv_db_path:        SQLite 数据库路径
        chroma_persist_dir: ChromaDB 持久化目录
        max_context_tokens: build_prompt 输出的最大 token 数
        """
        self.openai_client = openai_client
        self.llm_model = llm_model
        self.kv = KeyValueMemory(kv_db_path)
        self.vec = VectorMemory(openai_client, embedding_model, chroma_persist_dir)
        self.max_context_tokens = max_context_tokens

        try:
            self.encoder = tiktoken.encoding_for_model(llm_model)
        except KeyError:
            self.encoder = tiktoken.get_encoding("cl100k_base")

    def _count_tokens(self, text):
        return len(self.encoder.encode(text))

    # ==================== 显式写入 ====================

    def save_structured(self, user_id, category, key, value):
        """存储结构化记忆（偏好、画像、配置）→ SQLite"""
        self.kv.save(user_id, category, key, value)

    def save_unstructured(self, content, metadata=None):
        """存储非结构化记忆（经验、知识片段）→ ChromaDB"""
        return self.vec.add(content, metadata=metadata)

    # ==================== 智能路由 ====================

    def auto_save(self, user_id, content):
        """
        自动判断存储方式（核心生产逻辑）：
        - 调用 LLM 判断内容类型 + 提取结构化字段
        - LLM 失败时降级到规则路由
        """
        try:
            resp = self.openai_client.chat.completions.create(
                model=self.llm_model,
                messages=[{
                    "role": "user",
                    "content": f"""判断以下内容应该存为结构化键值还是非结构化向量。
内容: {content}

如果是明确的偏好/事实/配置（如"我喜欢Python"、"项目用FastAPI"），输出:
STRUCTURED|分类名|键名|值
例: STRUCTURED|preferences|language|Python

如果是经验/知识/对话片段，输出:
UNSTRUCTURED
只输出一行，不要解释。"""
                }],
                temperature=0,
            )
            result = resp.choices[0].message.content.strip()

            if result.startswith("STRUCTURED|"):
                parts = result.split("|")
                if len(parts) >= 4:
                    _, cat, key, val = parts[0], parts[1], parts[2], "|".join(parts[3:])
                    self.kv.save(user_id, cat, key, val)
                    print(f"  [路由→SQLite] {cat}.{key}={val}")
                    return

            # LLM 判断为非结构化，存入 ChromaDB
            self.vec.add(content, metadata={"user_id": user_id})
            print(f"  [路由→ChromaDB] {content[:40]}...")
            return

        except Exception as e:
            # LLM 路由失败，降级到规则路由
            print(f"  [路由] LLM异常({e})，降级到规则路由")

        # 规则路由：短文本 + 偏好关键词 → SQLite，否则 → ChromaDB
        keywords = ["喜欢", "偏好", "我是", "我用", "我的", "设置", "配置", "项目用"]
        if len(content) < 50 and any(kw in content for kw in keywords):
            self.kv.save(user_id, "auto_facts", f"fact_{len(self.kv.get_all(user_id))}", content)
            print(f"  [路由→SQLite] {content}")
        else:
            self.vec.add(content, metadata={"user_id": user_id})
            print(f"  [路由→ChromaDB] {content[:40]}...")

    # ==================== 统一检索 ====================

    def search(self, user_id, query, top_k=5):
        """
        统一检索：同时从 SQLite 和 ChromaDB 中查找
        返回 {"structured": {...}, "unstructured": [...]}
        """
        return {
            "structured": self.kv.get_all(user_id),
            "unstructured": self.vec.search(query, top_k=top_k),
        }

    # ==================== 上下文构建 ====================

    def build_prompt(self, user_id, query, top_k=3):
        """
        构建完整的 LLM 上下文：结构化画像（始终注入）+ 语义检索（按查询动态匹配）
        带 token 预算控制：如果拼接结果超限，优先裁剪向量检索结果
        """
        parts = []
        kv_ctx = self.kv.build_prompt(user_id)
        if kv_ctx:
            parts.append(kv_ctx)
        vec_ctx = self.vec.build_prompt(query, top_k=top_k)
        if vec_ctx:
            parts.append(vec_ctx)

        result = "\n\n".join(parts)
        # Token 预算控制：超出时逐步减少向量检索条数
        while self._count_tokens(result) > self.max_context_tokens and top_k > 0:
            top_k -= 1
            vec_ctx = self.vec.build_prompt(query, top_k=top_k) if top_k > 0 else ""
            parts = []
            if kv_ctx:
                parts.append(kv_ctx)
            if vec_ctx:
                parts.append(vec_ctx)
            result = "\n\n".join(parts)

        # 极端情况：连键值画像都超限，截断
        if self._count_tokens(result) > self.max_context_tokens:
            tokens = self.encoder.encode(result)
            safe = max(self.max_context_tokens - 3, 1)
            result = self.encoder.decode(tokens[:safe]) + "..."

        return result

    def close(self):
        self.kv.close()


# 使用
from openai import OpenAI
client = OpenAI()

m = HybridMemory(
    openai_client=client,
    llm_model="gpt-4o-mini",
    kv_db_path="/tmp/agent_kv.db",
    chroma_persist_dir="/tmp/chroma_memory",
)
m.save_structured("user_001", "profile", "name", "张三")
m.save_structured("user_001", "prefs", "framework", "FastAPI")
m.save_unstructured("PostgreSQL JSONB 比 JSON 更高效，支持索引查询")

# 自动路由
m.auto_save("user_001", "我喜欢用 Python")       # → SQLite
m.auto_save("user_001", "Docker 多阶段构建可大幅减小镜像体积，推荐 slim 基础镜像")  # → ChromaDB

# 统一检索 + 上下文
print(m.search("user_001", "数据库优化"))
print(m.build_prompt("user_001", "怎么优化查询性能？"))
m.close()

适用场景：生产环境、需要兼顾精度和灵活性的场景。

优点：两全其美，结构化信息精准，非结构化信息灵活。

缺点：实现复杂，需要维护两套存储。

---

选型建议

维度	全量保留	滑动窗口	Token 截断	总结压缩	键值存储	向量存储	混合存储
复杂度	极低	低	中	高	低	中	高
Token 效率	极差	中	好	最好	N/A	N/A	N/A
信息保留	完整	丢早期	丢早期	保留要点	完整	完整	完整
查询方式	-	-	-	-	精确	语义	精确+语义
存储引擎	内存	内存	内存	内存	SQLite/Redis	ChromaDB/Milvus	两者组合
生产可用	否	勉强	是	是	是	是	是

实践建议：

1. 短期记忆：生产环境推荐 Token 截断 + 总结压缩 组合。先按 token 裁剪保证不超限，再对裁剪掉的内容做摘要保留要点。

2. 长期记忆：生产环境推荐 混合存储。结构化走 SQLite/Redis（快而准），非结构化走 ChromaDB/Milvus（广而灵）。

3. 短期 + 长期联动：

┌──────────────────────────────────────────────────┐
│                  完整记忆架构                      │
│                                                   │
│  对话开始 → 从长期记忆检索相关上下文               │
│      │           (SQLite 精确查询                  │
│      │          + ChromaDB 语义检索)               │
│      ▼                                             │
│  注入 System Prompt → 开始对话                     │
│      │                                             │
│      ▼                                             │
│  对话进行中 → 短期记忆管理                         │
│      │           (Token截断 + 总结压缩)            │
│      ▼                                             │
│  对话结束 → 提取关键信息                           │
│      │           (LLM 提取事实/偏好/经验)          │
│      ▼                                             │
│  写入长期记忆 → SQLite（结构化）or ChromaDB（向量）│
└──────────────────────────────────────────────────┘

---

四类记忆各司其职

在完整的 Agent 系统中，四类记忆各司其职：

Working Memory（工作记忆）：是中枢，所有决策在这里做出。

LLM 每次生成回答，都是基于 Working Memory 中的上下文。它是推理的"工作台"。

Episodic Memory（情景记忆）：负责"回忆"。

当用户说"上次"“之前”"刚才提到"时，Episodic Memory 被激活，检索相关的历史对话。

Semantic Memory（语义记忆）：负责"理解"。

提供背景知识，让回复更专业。比如知道用户是技术负责人，回答时用技术术语；知道用户是产品经理，回答时关注业务价值。

Procedural Memory（程序记忆）：负责"执行"。

为什么需要分层记忆？

想象你是一个客服 Agent，正在处理用户咨询：

用户说：“我上周投诉的 COD 超标问题，你们查得怎么样了？”

要回答这个问题，你需要：

1. 现在对话的上下文（用户刚才说了什么）→ Working Memory
2. 上周的投诉记录（具体事件）→ Episodic Memory
3. COD 是什么、超标标准是什么（领域知识）→ Semantic Memory
4. 处理投诉的标准流程（先查上游→生成报告→人工审核）→ Procedural Memory

没有分层，所有信息混在一起，Agent 会"失忆"或"错乱"。

协作模式

场景：用户投诉 COD 超标

用户：我上周投诉的 COD 超标问题，你们查得怎么样了？

import time
from datetime import datetime, timedelta

# ========== 初始化四层记忆 ==========
working_mem = WorkingMemory(max_tokens=4000)
episodic_mem = EpisodicMemory(embedding_model=mock_embed)
semantic_mem = SemanticMemory(embedding_model=mock_embed)
procedural_mem = ProceduralMemory()

# 预置一些知识（Semantic Memory）
semantic_mem.set_user_profile("user_001", {
    "name": "张三",
    "region": "湘江流域",
    "role": "环保监测员",
    "preferred_channel": "email"
})
semantic_mem.add_knowledge(
    "COD（化学需氧量）标准值：地表水 IV 类 ≤ 30mg/L",
    source="GB 3838-2002"
)

# 预置流程（Procedural Memory）
procedural_mem.register_workflow("complaint_followup", {
    "trigger": "user asks about previous complaint",
    "steps": [
        {"name": "查询历史", "action": "query_episodic", "params": {"keyword": "{{complaint_type}}"}},
        {"name": "检查状态", "action": "check_status", "params": {"ticket_id": "{{ticket_id}}"}},
        {"name": "生成回复", "action": "generate_response", "params": {"template": "followup"}}
    ]
})

# ========== 模拟：上周的投诉记录（Episodic Memory） ==========
last_week = time.time() - 7 * 86400
episodic_mem.record_event(
    session_id="sess_lastweek_001",
    event_type="user_complaint",
    content="用户张三投诉湘江流域 COD 检测值 45mg/L，超过标准值 30mg/L，要求查明原因",
    metadata={"severity": "high", "region": "湘江流域", "pollutant": "COD"}
)

# ========== 开始新会话 ==========
session_id = "sess_today_001"

# Step 1: 用户输入进入 Working Memory
user_input = "我上周投诉的 COD 超标问题，你们查得怎么样了？"
working_mem.add_message("user", user_input, tokens=25)

# Step 2: Agent 理解意图（在 Working Memory 中推理）
working_mem.add_thought("用户询问上周投诉的跟进情况")
working_mem.add_thought("关键词：上周、COD 超标、投诉")
working_mem.add_thought("需要查询 Episodic Memory 获取历史事件")

# Step 3: 查询 Episodic Memory（时间 + 语义）
print("=== [Episodic Memory] 查询历史投诉 ===")
# 3.1 按时间范围查
start = time.time() - 14 * 86400  # 两周内
events = episodic_mem.recall_by_time(start, time.time())
print(f"找到 {len(events)} 个事件")

# 3.2 按语义查（更精确）
similar = episodic_mem.recall_by_similarity("COD 超标投诉", top_k=3)
print(f"语义相似事件: {[e['content'][:30] for e in similar]}")

# Step 4: 查询 Semantic Memory（用户画像 + 知识）
print("\n=== [Semantic Memory] 获取用户信息和领域知识 ===")
profile = semantic_mem.get_user_profile("user_001")
print(f"用户: {profile['name']}, 区域: {profile['region']}")

# RAG 检索相关知识
knowledge = semantic_mem.query_knowledge("COD 标准值")
print(f"相关知识: {knowledge[0]['text'] if knowledge else '无'}")

# Step 5: 加载 Procedural Memory（处理流程）
print("\n=== [Procedural Memory] 执行跟进流程 ===")
# 实际执行工作流
result = procedural_mem.execute_workflow(
    "complaint_followup",
    context={
        "user_id": "user_001",
        "complaint_type": "COD超标",
        "ticket_id": "TICKET_001"
    },
    tool_registry={
        "query_episodic": lambda **kw: {"found": True, "events": similar},
        "check_status": lambda **kw: {"status": "处理中", "progress": "已查上游，待生成报告"},
        "generate_response": lambda **kw: {
            "response": f"张先生，您上周投诉的湘江流域 COD 超标问题（检测值 45mg/L，超过 IV 类标准 30mg/L）目前正在处理中。我们已完成上游排查，正在生成详细报告，预计今日内提交人工审核。审核通过后将第一时间邮件通知您。"
        }
    }
)

# Step 6: 将结果和回复存入 Working Memory
response = result['variables']['result_生成回复']['response']
working_mem.add_message("assistant", response, tokens=80)

# Step 7: 记录本次交互到 Episodic Memory（为下次准备）
episodic_mem.record_event(
    session_id=session_id,
    event_type="followup_inquiry",
    content=f"用户询问上周 COD 投诉进展，Agent回复：{response[:50]}...",
    metadata={"resolved": False, "related_event": "evt_xxx"}
)

print("\n=== [Working Memory] 当前状态 ===")
print(working_mem.get_context())

---

分析维度

维度	含义	工程对应	例子
Form（形式）	记忆以什么数据结构存在	文本 / 向量 / 键值 / 图结构	用户画像是 JSON（形式），对话历史是文本数组（形式）
Function（功能）	记忆用来解决什么问题	检索 / 推理 / 决策支持	查用户偏好是为了个性化回答（功能）
Dynamics（动态）	记忆怎么变化、更新、遗忘	写入 / 压缩 / 过期清理 / 强化	7天前的记忆自动降权（动态）

Form（形式/结构维度）

是什么

Form 指的是记忆在系统中以什么数据结构存在，即记忆的物理/逻辑表示形式。

具体内容

类型	说明	示例
参数化记忆	存储在模型权重中，通过微调/训练获得	大模型预训练知识、LoRA 适配层
非参数化记忆	外挂存储，模型通过检索访问	向量数据库、知识图谱、文档库
隐式记忆	模型推理时的内部状态	Transformer 的 KV Cache、隐藏层状态
显式记忆	结构化/半结构化的外部记录	符号记忆、工作记忆缓冲区、日志

作用

• 解耦知识与推理：让模型参数专注推理能力，知识通过外挂记忆动态更新
• 可解释性：显式记忆可被人类审计、编辑、删除
• 持久化：对话结束后记忆不丢失，支持长期累积

Function（功能维度）

是什么

Function 指的是记忆在 Agent 运行中承担什么认知功能，即记忆被用来做什么。

具体内容

功能类型	作用	典型场景
工作记忆 (Working Memory)	临时保存当前任务上下文，支持多步推理	维护对话历史、待办事项列表
情景记忆 (Episodic Memory)	记录具体事件/经历，支持经验复用	“上次用户说喜欢简约风格”
语义记忆 (Semantic Memory)	存储事实性知识和概念关系	产品知识库、领域术语
程序记忆 (Procedural Memory)	存储技能和操作步骤	调用 API 的流程、工具使用规范
前瞻性记忆 (Prospective Memory)	未来需要执行的计划/提醒	“明天上午提醒用户开会”

作用

• 突破上下文长度限制：通过功能分区，只把最相关的记忆加载到上下文
• 支持复杂任务规划：程序记忆让 Agent 能执行多步骤工作流
• 个性化：情景记忆让 Agent 记住用户偏好，提供定制化服务
• 持续学习：新经验通过情景记忆积累，提炼为语义记忆

Dynamics（动态性维度）

是什么

Dynamics 指的是记忆如何随时间变化——如何写入、更新、遗忘、巩固，以及记忆之间的相互作用。

具体内容

动态过程	说明	机制
编码 (Encoding)	原始信息转换为记忆表示	文本→向量、对话→摘要、经验→知识图谱
巩固 (Consolidation)	短期记忆转化为长期记忆	睡眠机制（离线摘要）、知识蒸馏
检索 (Retrieval)	根据当前需求提取记忆	相似度搜索、注意力机制、图遍历
更新 (Update)	修正或扩展已有记忆	增量学习、冲突检测与合并
遗忘 (Forgetting)	清理过时/无关记忆	TTL 过期、重要性衰减、主动遗忘策略
泛化 (Generalization)	从具体经验抽象出规律	模式提取、规则归纳

作用

• 防止记忆膨胀：无限增长的记忆会导致检索噪音和成本激增
• 保持记忆准确性：旧记忆可能被新信息覆盖或修正
• 模拟人类认知节律：工作记忆→情景记忆→语义记忆的层级流转
• 自适应学习：Agent 越用越"懂"用户，但不会过时

三个维度的关系图

┌─────────────────────────────────────────────┐
│              Dynamics（动态性）               │
│   编码 → 巩固 → 检索 → 更新 → 遗忘 → 泛化     │
│         ↑ 驱动 Form 的流转 ↓                  │
├─────────────────────────────────────────────┤
│              Function（功能）                 │
│   工作记忆 → 情景记忆 → 语义记忆 → 程序记忆     │
│         ↑ 决定 Form 的组织方式 ↓               │
├─────────────────────────────────────────────┤
│              Form（形式）                     │
│   向量 / 图谱 / 键值对 / 模型参数 / 符号       │
└─────────────────────────────────────────────┘

核心逻辑：

• Form 是载体：决定记忆"长什么样"
• Function 是目的：决定记忆"用来干什么"
• Dynamics 是过程：决定记忆"如何演化"

---

说点实在的

Memory 这个东西，搞复杂了容易陷入"过度设计"。

简单对话，用滑动窗口就够了，别上向量库。

需要跨会话记忆的，先用 KV 存储，够用了别折腾向量。

真需要灵活检索的，再用向量存储。

混合存储是生产环境的首选，但前期开发成本高。

写代码的时候，记得多做测试。Memory 系统最容易出问题的是"检索准确率"——你明明存了"用户叫王五"，但检索时返回"用户叫王六"，这种 Bug 调起来非常痛苦。

下一篇文章，我会详细说说 Action。Action 是把"想法"变成"现实"的关键，包括 Action 的类型、执行流程、可靠性控制，每个环节都有不少细节。

---