记忆力并不是智慧；但没有记忆力还成什么智慧呢？

——美国 哈珀·李（Harper Lee，1926–2016）作品《杀死一只知更鸟》的深层主题

“记忆力并不是智慧；但没有记忆力还成什么智慧呢？”这句话虽非美国知名作家哈珀·李（Harper Lee，1926–2016）作品杀死一只知更鸟》小说原文，但它精准地概括了《杀死一只知更鸟》的一个深层主题：知识（记忆）是基础，但真正的智慧在于运用这些知识去理解复杂的人性、坚守正义，并保持精神的体面，这和AI Agent的记忆系统理念不谋而合。

要构建千人千面个性化且持续连贯的AI Agent，保持交流中上下文的一致，其核心便是为AI Agent装备上类似人类记忆的系统机制。

1. AI Memory简介

1.1 什么是Agent Memory

图1.1 和豆包聊过《基督山伯爵》但和千问没聊过

如图1.1，笔者最近在读《基督山伯爵》，碰到针对法国的一些历史、风土人情不懂的时候就习惯性的问豆包，豆包便有了关于此书内容的记忆，当我随便问豆包”一个路易大概等于多少钱？“时，豆包会自动联想到《基督山伯爵》的时代背景；同比笔者从未跟千问聊过类似的问题，当同样单纯问一句”一个路易大概等于多少钱？“时，千问就要从头开始对笔者这个问题的理解和分析了。

因此，所谓的Agent Memory，在理论上就是让Agent具有像人一样能记住之前互动过，积累到的知识经验从而更好的执行未来的规划与执行；实操层面而言，便是指实现Agent记忆能力的架构和相关技术了。

1.2 记忆的分类

虽然关于记忆的分类有一些不同的派系，但是比较普遍的分类还是通常分为短期记忆（short-term memory）和长期记忆(long-term memory)，如图1.2；

图1.2 AI Agent记忆的分类

• 短期记忆：为Agent在一个对话窗口（这个对话窗口的上限理论等于该模型实现的上下文窗口的最大长度，如GPT-4o的128K，Claude 3.5的200K，塞满了就塞不下，就需要有记忆被截断，或者被总结概括）提供交互的上下文内容，专注于当前对话情景，维持对话的连贯性；
• 长期记忆：构建用户的唯一身份特征，建立持久且个性化的知识库。

最适合描绘长期记忆和短期记忆理念的句子，就不得不提到中国有句古话叫作”识时务者……“，抱歉，差点串台了，叫作”台上一分钟，台下十年功“，其中台上一分钟就是所谓的短期记忆，台下十年功就是长期记忆，因此，短期记忆和长期记忆不是相互取代、非此即彼的关系，而是有机结合，才能让Agent实现持续进化，正如台上一分钟的精彩离不开台下十年功的苦练一般。

举个🌰类比，律师小刘在打一场官司；

• Agent的上下文窗口：这场官司的持续时间；

• 短期记忆：律师小刘在这场官司上使用到的法律知识与智慧；

• 长期记忆：律师小刘这辈子学过的所有关于法律的知识与结晶。

1.3 长短期记忆的组成部分及特点

图1.3 长短期记忆的特点

• 短期记忆重要组成部分：

  • 对话上下文/窗口：这是最核心的部分。即当前会话中最近几轮（或几千个tokens）的用户输入和AI的回应。它直接决定了AI下一句会说什么；
  • 当前任务的状态和中间结果：例如，在多步任务中（如“写一份报告，先列大纲，再写第一章”），Agent需要记住当前进行到哪一步、已经生成了哪些中间内容、用户的临时指令等；
  • 工作缓冲区中的临时信息：在工具调用（如联网搜索、代码执行）过程中，临时存储的API返回结果、计算出的数值等。

• 短期记忆特点：

  • 容量有限：受限于模型的上下文窗口长度（如128K tokens）。超出部分会被“遗忘”；

  • 易失性：会话结束后，如果没有特别设计，这部分记忆会消失；

  • 高速访问：直接存在于请求的上下文中，模型可以即时、无损地利用。

• 长期记忆重要组成部分：

  • 向量数据库：这是目前最主要的形式。将对话中的关键信息（如用户偏好、重要事实、个人经历）转换为嵌入向量并存储。当新对话发生时，通过语义相似度检索出相关记忆，并动态注入到短期记忆（上下文）中。这是一种按需、选择性回忆的机制。
  • 外部知识库/数据库：存储结构化或非结构化的专属信息，如产品文档、公司规章、私有代码库等。同样通过检索增强生成技术接入。
  • 元提示/系统指令：写入系统层面的、不会被常规对话覆盖的持久化指令。例如“你是一个喜欢用比喻的助手”，“用户叫小明，喜欢咖啡”。这构成了Agent的“底层性格”和“基础事实”。
  • 技能/工具集：Agent通过学习或配置掌握的可调用工具（函数）的集合，这也是一种长期记忆。
  • 反思与总结摘要：高级Agent会在会话结束后，自动对对话进行摘要，提炼出核心用户偏好、决策逻辑或新学到的知识，并存入长期记忆库，供未来使用。

• 长期记忆特点：

  • 容量大（理论上无限）：存储在外部数据库中，可以持续扩展。
  • 持久性：独立于任何会话而存在。
  • 选择性访问：并非所有记忆都同时加载，而是根据当前对话的语义相关性进行检索，只有最相关的部分会被激活并放入短期记忆。

2. 基于LangChain生态的记忆的存储实践

2.1 记忆存储的基操

在明确了记忆的概念和关键作用以后，接下来一起聊一聊记忆存储的基操，通常可以分为3大核心功能：

1. 记忆存储的实例化；
2. 保存记忆；
3. 检索和获取记忆；

为了快速说明，先以LangChain自带的InMemoryStore快速演绎记忆存储的以上3大核心功能。

• 准备工作，假设需要Agent记住程序员小刘的一些个性内容如下；

1.  "用户喜欢简短、直接的语言风格","用户使用中文和 Python","用户喜欢深色主题";

2. "用户是AI行业","特别关注记忆系统与Agent","使用的工具是LangGraph和LangChain";

3. "用户编码风格是逻辑清晰，代码整洁","用户文档风格喜欢markdown"。

• 基于InMemoryStore记忆存储和实例化、保存、获取与检索的实现对程序员小刘的记忆给到Agent实操demo，创建memory_store.py脚本，内容如下；

"""
InMemoryStore 实战示例：基于 LangGraph 的记忆存储与检索。

提供 put 保存、get 按 key 检索、search 语义检索的完整用法。
"""

from __future__ import annotations

from typing import Any


def _simple_embed(texts: list[str]) -> list[list[float]]:
    """简易嵌入函数，用于演示语义搜索。生产环境请使用真实嵌入模型。"""
    # 使用简单的数值模拟向量，实际应使用 OpenAI / HuggingFace 等
    dims = 4
    return [[float((hash(t) % 100) / 100) for _ in range(dims)] for t in texts]


def create_store() -> Any:
    """创建并返回配置了语义搜索的 InMemoryStore 实例。"""
    from langgraph.store.memory import InMemoryStore

    store = InMemoryStore(
        index={
            "embed": _simple_embed,
            "dims": 4,
        }
    )
    return store


def save_memory(
    store: Any,
    namespace: tuple[str, ...],
    key: str,
    value: dict[str, Any],
    *,
    index: list[str] | bool | None = None,
) -> None:
    """
    使用 put 保存记忆。

    Args:
        store: InMemoryStore 实例
        namespace: 命名空间，如 ("user_123", "preferences")
        key: 记忆唯一键
        value: 记忆内容（需可序列化）
        index: 指定要嵌入的字段；None 表示默认；False 表示不建立向量索引
    """
    kwargs: dict[str, Any] = {}
    if index is not None:
        kwargs["index"] = index
    store.put(namespace, key, value, **kwargs)


def get_memory(store: Any, namespace: tuple[str, ...], key: str) -> Any | None:
    """
    使用 get 按 key 检索记忆。

    Returns:
        对应记忆的值，不存在时返回 None
    """
    result = store.get(namespace, key)
    if result is None:
        return None
    # store.get 返回 Item，取 value 属性
    return getattr(result, "value", result)


def search_memories(
    store: Any,
    namespace: tuple[str, ...],
    query: str,
    *,
    filter_: dict[str, Any] | None = None,
    limit: int = 5,
) -> list[tuple[Any, float]]:
    """
    使用 search 做语义检索。

    Args:
        store: InMemoryStore 实例
        namespace: 命名空间
        query: 自然语言查询
        filter_: 可选内容过滤条件
        limit: 返回结果数量上限

    Returns:
        [(value, score), ...]，按相似度排序
    """
    kwargs: dict[str, Any] = {"query": query, "limit": limit}
    if filter_ is not None:
        kwargs["filter"] = filter_

    results = store.search(namespace, **kwargs)
    output: list[tuple[Any, float]] = []
    for r in results:
        val = getattr(r, "value", r)
        score = getattr(r, "score", 1.0)
        output.append((val, score))
    return output


def demo() -> None:
    """演示 InMemoryStore 的 put、get、search 用法。"""
    store = create_store()

    # 命名空间：用户 ID + 应用场景
    user_id = "user_001"
    namespace = (user_id, "preferences")

    # ---------- put：保存多条记忆 ----------
    save_memory(
        store,
        namespace,
        "language_prefs",
        {
            "rules": [
                "用户喜欢简短、直接的语言风格",
                "用户使用中文和 Python",
            ],
            "theme": "dark",
        },
    )

    save_memory(
        store,
        namespace,
        "work_context",
        {
            "industry": "AI",
            "focus": "记忆系统与 Agent",
            "tools": ["LangGraph", "LangChain"],
        },
    )

    save_memory(
        store,
        namespace,
        "communication_style",
        {
            "preference": "逻辑清晰，代码整洁",
            "format": "markdown",
        },
    )

    # ---------- get：按 key 精确检索 ----------
    lang_prefs = get_memory(store, namespace, "language_prefs")
    print("【get 检索】language_prefs:")
    print(lang_prefs)
    print()

    missing = get_memory(store, namespace, "not_exist")
    print("【get 检索】不存在的 key:", missing)
    print()

    # ---------- search：语义检索 ----------
    results = search_memories(store, namespace, "编程语言偏好", limit=3)
    print("【search 检索】query='编程语言偏好':")
    for val, score in results:
        print(f"  score={score:.3f} -> {val}")
    print()

    results = search_memories(
        store, namespace, "开发工具", filter_={"theme": "dark"}, limit=5
    )
    print("【search 检索】query='开发工具' + filter theme=dark:")
    for val, score in results:
        print(f"  score={score:.3f} -> {val}")
    print()

    print("InMemoryStore 实战演示完成。")


if __name__ == "__main__":
    demo()
    

• 代码demo的效果输出如下。

【get 检索】language_prefs:
{'rules': ['用户喜欢简短、直接的语言风格', '用户使用中文和 Python'], 'theme': 'dark'}

【get 检索】不存在的 key: None

【search 检索】query='编程语言偏好':
  score=1.000 -> {'rules': ['用户喜欢简短、直接的语言风格', '用户使用中文和 Python'], 'theme': 'dark'}
  score=1.000 -> {'industry': 'AI', 'focus': '记忆系统与 Agent', 'tools': ['LangGraph', 'LangChain']}
  score=1.000 -> {'preference': '逻辑清晰，代码整洁', 'format': 'markdown'}

【search 检索】query='开发工具' + filter theme=dark:
  score=1.000 -> {'rules': ['用户喜欢简短、直接的语言风格', '用户使用中文和 Python'], 'theme': 'dark'}

InMemoryStore 实战演示完成。

Process finished with exit code 0

2.2 语义搜索增强记忆检索

记忆检索的解决方案最简单的实现肯定是基于关键字的搜索，但是基于LLM的Agent，本质还是处理的自然语言，对于自然语言的检索，传统的关键字搜索就显得力不从心了，目前较为普遍的做法都是基于语义的相似性来检索信息的，而语义相似性检索的关键技术，就是基于向量嵌入技术实现的，2.1的demo实现了简单的4维向量嵌入技术的模拟，接下来使用BGE-M3模型高维向量向量化进行语义检索的实践，如果是对Agent开发的新人，还是建议先看看AI Agent开发实战QuickStart，熟悉了解LangChain/LangGraph的脚手架文件。

1. 先在.env文件配置你的Embedding模型，url和key；

# 恭喜你不用一分钱先白嫖好内部模型和外部模型
LOCAL_MODEL="ollama:qwen3:4b"          # 自己本地的Ollama模型
INTERNAL_EMBEDDING_MODEL="openai/BAAI/bge-m3"

OPENAI_API_KEY="sk-XXX"                # 模型引用的url的key，配置的前缀“OPEN_”会被初始化模型用init_chat_model()根据模型的“openai: ”自动引用
OPENAI_BASE_URL="https://api.XXXX"     # 模型引用的url，配置的前缀“OPEN_”会被初始化模型用init_chat_model()根据模型的“openai: ”自动引用

2. 配置文件config.py进行引用Embedding模型；

import os

from dotenv import load_dotenv

load_dotenv() # 加载环境变量文件.env

# ============================================================================
# MODEL CONFIGURATION
# ============================================================================



# DEFAULT_MODEL = os.getenv("LOCAL_MODEL", "anthropic:claude-haiku-4-5")   # 获取默认模型为环境变量中的本地Ollama模型模型

DEFAULT_MODEL = os.getenv("LOCAL_MODEL", "anthropic:claude-haiku-4-5")  # 获取默认模型为环境变量中的外部硅基流动免费模型
INTERNAL_EMBEDDING_MODEL = os.getenv("INTERNAL_EMBEDDING_MODEL","")

#print(INTERNAL_EMBEDDING_MODEL)

3. 准备工作做好了，创建一个memory_store_bge_m3.py脚本，这里还是基于实现程序员小刘的个性内容为例，就可以开始实践了；

"""
基于 BGE-M3 向量模型的 InMemoryStore 实战示例。

使用 BGE-M3（1024 维）配置 InMemoryStore，实现语义检索。
提供 put 保存、get 按 key 检索、search 语义检索的完整用法。
"""

from __future__ import annotations

from langchain_openai import OpenAIEmbeddings

from src.config import INTERNAL_EMBEDDING_MODEL

from typing import Any

#from langchain.embeddings import OpenAIEmbeddings

# BGE-M3 dense 向量维度
BGE_M3_DIMS = 1024

_bge_model: Any | None = None


def _get_bge_model() -> Any:
    """延迟加载并缓存 BGE-M3 模型。"""
    global _bge_model
    if _bge_model is None:

        _bge_model = INTERNAL_EMBEDDING_MODEL
    return _bge_model


def _bge_embed():
    """
    BGE-M3 嵌入函数，供 InMemoryStore 语义索引使用。
    """
    embeddings=OpenAIEmbeddings(model=INTERNAL_EMBEDDING_MODEL)
    return embeddings

def create_store() -> Any:
    """创建并返回配置了 BGE-M3 语义搜索的 InMemoryStore 实例。"""
    from langgraph.store.memory import InMemoryStore

    return InMemoryStore(
        index={
            "embed": _bge_embed(),
            "dims": BGE_M3_DIMS,
        }
    )


def save_memory(
    store: Any,
    namespace: tuple[str, ...],
    key: str,
    value: dict[str, Any],
    *,
    index: list[str] | bool | None = None,
) -> None:
    """
    使用 put 保存记忆。

    Args:
        store: InMemoryStore 实例
        namespace: 命名空间，如 ("user_123", "preferences")
        key: 记忆唯一键
        value: 记忆内容（需可序列化）
        index: 指定要嵌入的字段；None 表示默认；False 表示不建立向量索引
    """
    kwargs: dict[str, Any] = {}
    if index is not None:
        kwargs["index"] = index
    store.put(namespace, key, value, **kwargs)


def get_memory(store: Any, namespace: tuple[str, ...], key: str) -> Any | None:
    """
    使用 get 按 key 检索记忆。

    Returns:
        对应记忆的值，不存在时返回 None
    """
    result = store.get(namespace, key)
    if result is None:
        return None
    return getattr(result, "value", result)


def search_memories(
    store: Any,
    namespace: tuple[str, ...],
    query: str,
    *,
    filter_: dict[str, Any] | None = None,
    limit: int = 5,
) -> list[tuple[Any, float]]:
    """
    使用 search 做 BGE-M3 语义检索。

    Args:
        store: InMemoryStore 实例
        namespace: 命名空间
        query: 自然语言查询
        filter_: 可选内容过滤条件
        limit: 返回结果数量上限

    Returns:
        [(value, score), ...]，按语义相似度排序
    """
    kwargs: dict[str, Any] = {"query": query, "limit": limit}
    if filter_ is not None:
        kwargs["filter"] = filter_

    results = store.search(namespace, **kwargs)
    output: list[tuple[Any, float]] = []
    for r in results:
        val = getattr(r, "value", r)
        score = getattr(r, "score", 1.0)
        output.append((val, score))
    return output


def demo() -> None:
    """演示基于 BGE-M3 的 InMemoryStore 的 put、get、search 用法。"""
    print("初始化 InMemoryStore（BGE-M3）...")
    store = create_store()

    user_id = "user_001"
    namespace = (user_id, "preferences")

    # ---------- put：保存多条记忆 ----------
    save_memory(
        store,
        namespace,
        "language_prefs",
        {
            "rules": [
                "用户喜欢简短、直接的语言风格",
                "用户使用中文和 Python",
            ],
            "theme": "dark",
        },
    )

    save_memory(
        store,
        namespace,
        "work_context",
        {
            "industry": "AI",
            "focus": "记忆系统与 Agent",
            "tools": ["LangGraph", "LangChain"],
        },
    )

    save_memory(
        store,
        namespace,
        "communication_style",
        {
            "preference": "逻辑清晰，代码整洁",
            "format": "markdown",
        },
    )

    # ---------- get：按 key 精确检索 ----------
    lang_prefs = get_memory(store, namespace, "language_prefs")
    print("\n【get 检索】language_prefs:")
    print(lang_prefs)

    missing = get_memory(store, namespace, "not_exist")
    print("\n【get 检索】不存在的 key:", missing)

    # ---------- search：BGE-M3 语义检索 ----------
    results = search_memories(store, namespace, "编程语言偏好", limit=3)
    print("\n【search 检索】query='编程语言偏好':")
    for val, score in results:
        print(f"  score={score:.4f} -> {val}")

    results = search_memories(
        store, namespace, "开发工具", filter_={"theme": "dark"}, limit=5
    )
    print("\n【search 检索】query='开发工具' + filter theme=dark:")
    for val, score in results:
        print(f"  score={score:.4f} -> {val}")

    print("\nBGE-M3 InMemoryStore 实战演示完成。")


if __name__ == "__main__":
    demo()

4. 基于BGE-M3Embedding模型的运行结果如下。

初始化 InMemoryStore（BGE-M3）...

【get 检索】language_prefs:
{'rules': ['用户喜欢简短、直接的语言风格', '用户使用中文和 Python'], 'theme': 'dark'}

【get 检索】不存在的 key: None

【search 检索】query='编程语言偏好':
  score=0.6952 -> {'rules': ['用户喜欢简短、直接的语言风格', '用户使用中文和 Python'], 'theme': 'dark'}
  score=0.6418 -> {'preference': '逻辑清晰，代码整洁', 'format': 'markdown'}
  score=0.6142 -> {'industry': 'AI', 'focus': '记忆系统与 Agent', 'tools': ['LangGraph', 'LangChain']}

【search 检索】query='开发工具' + filter theme=dark:
  score=0.5598 -> {'rules': ['用户喜欢简短、直接的语言风格', '用户使用中文和 Python'], 'theme': 'dark'}

BGE-M3 InMemoryStore 实战演示完成。

Process finished with exit code 0

通过Embedding模型实现向量化嵌入技术实现语义增强记忆检索的实践后，语义搜索的核心优势就显现出来了，具体概括如下：

1. 核心原理是利用向量嵌入技术将文本转换为数值表示，通过计算向量间的相似度（如余弦相似度）确定语义相关性，使用Embedding模型的向量维度越高，一般效果越好；
2. 突破关键字检索的局限，语义检索可以实现同义词，近义词和概念类似但是表述方式不同的内容，实现语义概念层面的信息关联度分析，如小刘，附近有什么好吃的馆子和刘总，附近美食推荐一些呗，虽然句子表达不同，但是语义是相近的。

2.3 自定义记忆存储

LangChain/LangGraph的InMemoryStore虽然可以快速实现Agent记忆体的原型验证和单元测试，但还是有不少缺点的，如：

1. 数据不持久化：程序重启后所有数据丢失，无法用于生产环境中需要长期记忆的场景。
2. 容量受限：受限于可用内存大小，无法存储海量数据；若存储过多记忆可能导致内存溢出（OOM）。
3. 无法跨进程/分布式共享：每个进程拥有独立的内存空间，无法在多实例部署时共享记忆，不适合分布式系统。
4. 缺乏数据管理机制：无自动过期、淘汰策略（如 LRU），需要开发者自行维护内存释放，否则可能造成内存泄漏。
5. 不适合生产环境：由于上述限制，通常仅用于开发、测试或单机小规模应用，正式部署需替换为持久化存储（如 Redis、数据库）。

因此，针对生产环境的Agent记忆存储，还是需要基础特色的存储文件或数据库来存储，LangChain/LangGraph也提供了抽象类BaseStore专门用于实现自定义记忆存储，精准满足各类Agent的独特需求。

实现BaseStore时，正确处理记忆内容的IndexConfig并与向量数据库或索引库集成是关键所在，要点有以下几点：

1. 索引配置处理：初始化存储实例时保存IndexConfig参数（如嵌入模型或函数，向量维度，索引字段等），嵌入模型与维度参数的校验机制等；
2. 记忆内容数据存储时的索引处理：put（优先）/aput方法将生成的嵌入向量存入专用索引结构（向量数据库或内存索引）；
3. 语义检索实现：使用search/asearch方法处理检索相关参数时自动触发语义检索流程，执行向量相似度计算查询文件与历史记忆的相似度结果，最后返回带相似度评分的检索结果对象列表。
4. 关于存储数据的存储媒介选型：可以是向量数据库（如Milvus/Pinecone），也可以是各类关系或非关系型数据库（如PostgreSQL/MongoDB），当然也可以是最简单廉价的文件存储，至于选什么合适，笔者的建议是，基于业务或产品实现的前提下，哪个便宜选哪个就好，以后记忆内容膨胀了再优化选型更强更贵的数据库即可。

以下是关于自定义记忆存储的demo代码，还是以小刘的编程习惯为例子，仅在说明问题，此处就暂时先用最廉价的文件存储记忆为例实现记忆的索引配置，存储，检索等功能，记忆存储文件也直接选在了代码的根目录下设置文件夹为memory_store_data；

"""
基于 LangChain BaseStore 的文本文件记忆存储实现。

将记忆以 JSON 文件形式持久化到本地目录，实现 put、get、search 等基本功能。
"""

from __future__ import annotations

import json
import re
from collections.abc import Sequence
from pathlib import Path
from typing import Any, Iterator

from langchain_core.stores import BaseStore


def _namespace_to_key(namespace: tuple[str, ...], key: str) -> str:
    """将 namespace 与 key 编码为内部存储键。"""
    parts = (*namespace, key)
    return "/".join(parts)


def _key_to_path(root: Path, store_key: str) -> Path:
    """将存储键转换为文件路径。"""
    # 避免路径穿越，仅允许字母数字、下划线、连字符
    safe_parts = [
        re.sub(r"[^\w\-]", "_", p) or "_" for p in store_key.split("/")
    ]
    return root.joinpath(*safe_parts).with_suffix(".json")


def _path_to_key(root: Path, path: Path) -> str | None:
    """将文件路径反向解析为存储键。"""
    try:
        rel = path.relative_to(root)
    except ValueError:
        return None
    if rel.suffix != ".json":
        return None
    return str(rel.with_suffix("")).replace("\\", "/")


class TextFileStore(BaseStore[str, bytes]):
    """
    基于本地文本文件的 BaseStore 实现，用于持久化记忆。

    存储目录结构示例：
        root_dir/
            user_001/
                preferences/
                    language_prefs.json
                    work_context.json
    """

    def __init__(self, root_dir: str | Path) -> None:
        """
        初始化存储。

        Args:
            root_dir: 存储根目录，所有记忆文件将保存在该目录下。
        """
        self._root = Path(root_dir).resolve()
        self._root.mkdir(parents=True, exist_ok=True)

    def mget(self, keys: Sequence[str]) -> list[bytes | None]:
        """批量获取键对应的值。"""
        result: list[bytes | None] = []
        for key in keys:
            path = _key_to_path(self._root, key)
            if not path.exists() or not path.is_file():
                result.append(None)
                continue
            try:
                result.append(path.read_bytes())
            except OSError:
                result.append(None)
        return result

    def mset(self, key_value_pairs: Sequence[tuple[str, bytes]]) -> None:
        """批量写入键值对。"""
        for key, value in key_value_pairs:
            path = _key_to_path(self._root, key)
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_bytes(value)

    def mdelete(self, keys: Sequence[str]) -> None:
        """批量删除键。"""
        for key in keys:
            path = _key_to_path(self._root, key)
            if path.exists() and path.is_file():
                try:
                    path.unlink()
                except OSError:
                    pass

    def yield_keys(self, *, prefix: str | None = None) -> Iterator[str]:
        """按前缀迭代键。"""
        if not self._root.exists():
            return
        for path in self._root.rglob("*.json"):
            if not path.is_file():
                continue
            store_key = _path_to_key(self._root, path)
            if store_key is None:
                continue
            if prefix is None or store_key.startswith(prefix):
                yield store_key


# ---------- 高层 API：put / get / search ----------


def _serialize(value: dict[str, Any]) -> bytes:
    return json.dumps(value, ensure_ascii=False, indent=2).encode("utf-8")


def _deserialize(raw: bytes) -> dict[str, Any]:
    return json.loads(raw.decode("utf-8"))


def create_store(root_dir: str | Path) -> TextFileStore:
    """
    创建文本文件存储实例。

    Args:
        root_dir: 存储根目录，如 "./data/memories" 或 Path("memories")

    Returns:
        配置好的 TextFileStore 实例
    """
    return TextFileStore(root_dir)


def put(
    store: TextFileStore,
    namespace: tuple[str, ...],
    key: str,
    value: dict[str, Any],
) -> None:
    """
    存储记忆。

    Args:
        store: TextFileStore 实例
        namespace: 命名空间，如 ("user_001", "preferences")
        key: 记忆唯一键
        value: 记忆内容，需为可 JSON 序列化的字典
    """
    store_key = _namespace_to_key(namespace, key)
    store.mset([(store_key, _serialize(value))])


def get(
    store: TextFileStore,
    namespace: tuple[str, ...],
    key: str,
) -> dict[str, Any] | None:
    """
    按 key 获取记忆。

    Returns:
        记忆内容，不存在时返回 None
    """
    store_key = _namespace_to_key(namespace, key)
    results = store.mget([store_key])
    raw = results[0]
    if raw is None:
        return None
    return _deserialize(raw)


def search(
    store: TextFileStore,
    namespace: tuple[str, ...],
    *,
    filter_: dict[str, Any] | None = None,
    limit: int = 10,
) -> list[dict[str, Any]]:
    """
    检索命名空间下的记忆，支持按内容过滤。

    Args:
        store: TextFileStore 实例
        namespace: 命名空间
        filter_: 内容过滤条件，要求记忆的键值对完全匹配
        limit: 返回数量上限

    Returns:
        匹配的记忆列表
    """
    prefix = "/".join(namespace) + "/" if namespace else ""
    keys = list(store.yield_keys(prefix=prefix))[: limit * 2]
    if not keys:
        return []

    results = store.mget(keys)
    items: list[dict[str, Any]] = []
    for key, raw in zip(keys, results):
        if raw is None:
            continue
        try:
            val = _deserialize(raw)
        except (json.JSONDecodeError, UnicodeDecodeError):
            continue
        if filter_ is not None:
            if not all(val.get(k) == v for k, v in filter_.items()):
                continue
        items.append(val)
        if len(items) >= limit:
            break
    return items


def demo() -> None:
    """演示 TextFileStore 的 put、get、search 用法。"""
    root = Path(__file__).resolve().parent.parent / "memory_store_data"
    print(f"存储目录: {root}\n")

    store = create_store(root)
    namespace = ("user_001", "preferences")

    # ---------- put：存储记忆 ----------
    put(
        store,
        namespace,
        "language_prefs",
        {
            "rules": ["用户喜欢简短、直接的语言", "用户使用中文和 Python"],
            "theme": "dark",
        },
    )
    put(
        store,
        namespace,
        "work_context",
        {
            "industry": "AI",
            "focus": "记忆系统",
            "tools": ["LangGraph", "LangChain"],
        },
    )
    put(
        store,
        namespace,
        "communication_style",
        {
            "preference": "逻辑清晰，代码整洁",
            "theme": "dark",
        },
    )
    print("已存储 3 条记忆\n")

    # ---------- get：按 key 获取 ----------
    lang_prefs = get(store, namespace, "language_prefs")
    print("【get】language_prefs:")
    print(lang_prefs)
    print()

    missing = get(store, namespace, "not_exist")
    print("【get】不存在的 key:", missing)
    print()

    # ---------- search：检索 ----------
    all_items = search(store, namespace, limit=5)
    print("【search】命名空间下全部记忆 (limit=5):")
    for i, item in enumerate(all_items, 1):
        print(f"  {i}. {item}")
    print()

    filtered = search(store, namespace, filter_={"theme": "dark"}, limit=5)
    print("【search】filter theme=dark:")
    for i, item in enumerate(filtered, 1):
        print(f"  {i}. {item}")
    print()

    print("TextFileStore 示例演示完成。")


if __name__ == "__main__":
    demo()

运行结果如下；

存储目录: 你的项目名/src/memory_store_data

已存储 3 条记忆

【get】language_prefs:
{'rules': ['用户喜欢简短、直接的语言', '用户使用中文和 Python'], 'theme': 'dark'}

【get】不存在的 key: None

【search】命名空间下全部记忆 (limit=5):
  1. {'industry': 'AI', 'focus': '记忆系统', 'tools': ['LangGraph', 'LangChain']}
  2. {'preference': '逻辑清晰，代码整洁', 'theme': 'dark'}
  3. {'rules': ['用户喜欢简短、直接的语言', '用户使用中文和 Python'], 'theme': 'dark'}

【search】filter theme=dark:
  1. {'preference': '逻辑清晰，代码整洁', 'theme': 'dark'}
  2. {'rules': ['用户喜欢简短、直接的语言', '用户使用中文和 Python'], 'theme': 'dark'}

TextFileStore 示例演示完成。

Process finished with exit code 0

Agent的记忆也被持久化的保存下来了，记忆存储的文件效果如下图；

# 记忆存储文件树结构
./src/memory_store_data
└── user_001
    └── preferences
        ├── communication_style.json
        ├── language_prefs.json
        └── work_context.json

# 记忆存储文件communication_style.json内容
╰─$ cat src/memory_store_data/user_001/preferences/communication_style.json 
{
  "preference": "逻辑清晰，代码整洁",
  "theme": "dark"
}%   

# 记忆存储文件language_prefs.json内容
╰─$ cat src/memory_store_data/user_001/preferences/language_prefs.json     
{
  "rules": [
    "用户喜欢简短、直接的语言",
    "用户使用中文和 Python"
  ],
  "theme": "dark"
}%  

# 记忆存储文件work_context.json内容
╰─$ cat src/memory_store_data/user_001/preferences/work_context.json    
{
  "industry": "AI",
  "focus": "记忆系统",
  "tools": [
    "LangGraph",
    "LangChain"
  ]
}%  

实现记忆的存储和检索是基本的操作，当用户量高并发，用户历史使用记录数据存储空间越膨胀等情况出现后，Agent的记忆机制还是面临不少挑战的，此处不同的应用开发者也是八仙过海——各显神通，截至目前的技术和经费的平衡，记忆机制面临极端的情况仍然是个调优的问题，没有统一的最佳方案，只有相对的最优解。

经典案例：用户的记忆是”我梦中爬过长城……“；

Agent记忆检索的结果可能是：”我爬过长城……“

3. 记忆系统的实际应用

3.1 越用越聪明的个性化推荐

长期记忆最具价值的应用场景之一就是实现个性化推荐服务，通过持续记录用户偏好（访问，点击，搜藏，转发等）历史交互交易数据，AI Agent通过商品或内容（Content）的MCP工具结合用户的兴趣点提供精准定制的推荐内容，从而显著提升用户参与度与满意度，进一步提升电商的CTR和CVR，核心实现逻辑如下：

1. 记忆更新时机控制：确保生成推荐的内容后立即更新用户资料，使当前的用户对话线程以及后续的交互线程都能够基于最新偏好进行决策；
2. 智能体信息提取实现：输入（当前对话信息状态），处理（LLM解析最新用户消息），输出（结构化提取产品类别偏好），容错机制等
3. 资料更新工作流：获取最新的用户信息资料与老的记忆资料融合，更新记忆存储。

考虑到篇幅问题，核心代码就先略过了，下次有机会笔者会专门出一期关于Agent结合记忆实现个性化推荐项目的博文。

图3.1 基于记忆的Agent个性化推荐

3.2 多步骤、跨对话的情景化任务

对于需要让AI Agent协助用户完成多步骤任务或复杂工作流的需求，长短期记忆尤其至关重要了，短期记忆跟踪任务的当前阶段、记住当前步骤的用户输入，当前语境等状态；长期记忆则可以用户存储模版，用户个性化偏好等，从而总体上指导AI Agent的总体Workflow。

3.3 记忆存储应用的其他优化

• LangGraph结合TrustCall的使用，让记忆存储和检索更加结构化；
• 结合一些优秀的开源、闭源记忆体工具库或SDK，让记忆存储开发事半功倍，如LangChain团队的LangMem，将先进的记忆管理功能更好的集成到Agent开发中。

4. 火出地球的OpenClaw记忆系统原理

OpenClaw的记忆存储原理可概括为“文件优先、本地优先”的分层架构，核心由文件层与索引层组成，并通过工具接口为Agent提供记忆能力。

4.1 文件层：基于Markdown的持久化存储

• 长期记忆（MEMORY.md）：存放用户偏好、重要决策等稳定事实，手动或由Agent维护。
• 每日日志（memory/YYYY-MM-DD.md）：记录当天操作、临时决策，作为短期工作记忆。
• 会话日志（sessions/*.jsonl）：完整对话历史，供索引层选择性纳入检索。

所有记忆以纯文本形式存在，易于版本管理和人工干预，确保文件即真相。

4.2 索引层：混合检索实现精准召回

• 文本检索：利用SQLite的FTS5对Markdown分块建立全文索引，支持精确关键词匹配（如错误码、路径）。
• 向量检索：通过sqlite-vec扩展在SQLite内建向量索引，计算语义相似度，可配置本地或云端嵌入模型。
• 混合排序：融合BM25与向量分数，平衡语义理解与关键词敏感度。

4.3 与Agent的集成

• 通过工具（如memory_search）暴露记忆查询能力，Agent在回答前主动检索相关片段。
• 引入“预压缩记忆冲刷”机制：在上下文接近上限时，自动将重要信息写入当日日志，减少信息损失。

此外，记忆后端支持插件化扩展（如LanceDB），满足不同场景需求。整体设计兼顾数据主权、检索效率与灵活集成，为个人或单机Agent提供轻量级企业级记忆方案。

以上就是AI Agent的记忆存储与检索阐述和实践，然后在真正的AI Agent开发中，记忆系统的设计仍然面临不少的挑战，为AI Agent选择相对较优的记忆系统解决方案仍需要仔细权衡不同的因素，但还是以发展的眼光看待问题吧，把记忆系统的解决方案当成迭代来做，当过持续优化记忆系统，才能真正实现具备持续学习能力，个性化交互体验和安全可靠的新一代智能体，从而真正推动AI Agent技术的创新与最佳实践。