1. 项目概述：当AI开始拥有“记忆”

最近在AI Agent的圈子里，一个叫“Gliding Horse”的记忆系统讨论度挺高。它被描述为“像CPU一样思考的AI记忆架构”，这个比喻一下子就抓住了我的注意力。我们平时聊AI的记忆，要么是RAG那种从外部知识库检索，要么是给大模型加个上下文窗口，本质上还是在处理“当下”的输入输出。但“像CPU一样思考”暗示了一种更根本的转变：它试图为AI Agent构建一个系统性的、可管理的、分层的记忆工作流，就像CPU有寄存器、高速缓存和内存一样，各司其职，协同工作。

简单来说，Gliding Horse不是一个单一的工具，而是一套架构理念。它属于一个更广泛的范畴——Agent Harness。你可以把Harness理解为“缰绳”或“马具”，它不是替代马（Agent）去奔跑，而是为马提供控制、支持和赋能的基础设施层。Agent Harness就是包裹在AI Agent核心推理逻辑之外的那一层，负责管理记忆、工具调用、状态持久化、任务调度等“脏活累活”，让Agent能更专注、更高效地进行决策和推理。Gliding Horse，就是这套Harness中专精于“记忆”问题的核心组件。

那么，它到底解决了什么痛点？我们都有体会，现在的AI对话常常是“金鱼记忆”，上下文一长就丢三落四，更别提在长期、复杂的多轮交互中保持连贯的认知状态了。对于想构建真正实用AI应用的开发者来说，一个可靠、可扩展的记忆系统是刚需。Gliding Horse提出的“三层记忆架构”，正是为了应对这个挑战，它想让AI的记忆变得可预测、可管理、可优化，而不仅仅是堆叠更多的Token。

2. 核心架构解析：三层记忆的CPU式分工

Gliding Horse记忆系统的核心创新，在于它借鉴了现代计算机CPU的存储层次结构，设计了一套清晰的三层记忆架构。这不仅仅是命名上的类比，其内在的数据流转逻辑和职责划分，确实与CPU的工作方式有神似之处。

2.1 工作记忆：CPU的寄存器

这是记忆系统的最顶层，速度最快，容量最小，但直接服务于Agent当前的“思考”过程。你可以把它想象成CPU的寄存器，存放着正在被直接操作和计算的数据。

• 功能定位 ：存储当前会话轮次中产生的临时信息、中间推理步骤、被频繁引用的关键事实，以及从下层记忆“加载”上来的相关上下文。它是Agent“意识”的直接体现。
• 数据特性 ：极短的保留时间（通常仅限当前任务或对话轮次），高访问频率，内容高度相关且结构化（可能是经过提炼的键值对或对象）。
• 类比解释 ：就像你在心算一个复杂公式时，会把中间结果（比如先算出的乘法积）暂时记在脑子里，用于下一步计算。这个“脑子里的临时存储区”就是工作记忆。
• 实操意义 ：在工程实现上，这一层通常存在于应用程序的内存中，甚至直接是Agent推理过程中的一个变量或数据结构。它的高效与否，直接决定了Agent单次响应的速度和连贯性。

2.2 短期记忆：CPU的高速缓存

这是承上启下的一层，容量和工作记忆相比有所扩大，访问速度稍慢，但远快于长期存储。它对应CPU的L1/L2/L3高速缓存。

• 功能定位 ：缓存近期高频使用的信息。例如，一个用户在过去几分钟内反复提及的偏好、当前复杂任务分解后的子任务状态、以及从长期记忆中检索出来准备用于当前会话的“热数据”。
• 数据特性 ：保留时间从几分钟到几小时（可配置），遵循一定的缓存淘汰策略（如LRU-最近最少使用）。内容是工作记忆的缓冲区和长期记忆的预加载区。
• 类比解释 ：好比你在写一篇文章，当前段落（工作记忆）正在写，而整篇文章的大纲、前几段的内容、以及你刚刚查过的参考文献摘要，都摊开在桌面上（短期记忆），方便你随时取用，而不需要每次都去书架上翻找。
• 实操意义 ：这一层通常由高性能的键值数据库（如Redis）或内存数据库实现。它的存在极大地减少了对底层慢速存储的访问次数，是提升系统整体吞吐量的关键。开发者需要在这里精心设计缓存键和过期策略。

2.3 长期记忆：CPU的主内存与硬盘

这是记忆系统的基石，容量最大，但访问速度最慢。它存储了Agent所有的“人生经验”和“知识积累”，对应计算机的主内存和硬盘。

• 功能定位 ：持久化存储所有历史交互记录、学到的知识、用户画像、任务完成日志等。当工作记忆和短期记忆中没有所需信息时，Agent会从这里进行检索。
• 数据特性 ：永久或长期存储，数据量可能非常庞大。访问通常需要经过索引和检索过程（例如通过向量数据库进行语义搜索）。数据可以是结构化的（如用户属性表），也可以是非结构化的（如对话历史文本）。
• 类比解释 ：这就是你的个人图书馆或硬盘里的所有文档。你不会把所有书都放在桌上，但你知道它们在哪，需要时可以按索引（书名、关键词）去找到并取出来阅读。
• 实操意义 ：这一层的实现最为多样，可能结合关系型数据库（存储结构化元数据）、向量数据库（用于基于语义的相似性检索）、以及对象存储（存储大型文件如生成的图片）。其设计直接决定了Agent的知识广度和历史回溯能力。

这三层之间并非孤立，而是通过一套精密的“加载/回写”机制进行数据交换，模拟了CPU与内存之间的数据调度，这才是“像CPU一样思考”的精髓所在。

3. 数据流转与协同机制

理解了静态的分层，动态的数据流转才是系统活起来的灵魂。Gliding Horse 定义了一套清晰的数据移动规则，确保信息在正确的时间出现在正确的位置。

3.1 自上而下的“缓存未命中”与检索

当 Agent 在工作记忆层进行推理，发现需要某个关键信息（例如“用户上次提到的项目截止日期”）但本地没有时，就发生了“缓存未命中”。

1. 查询短期记忆 ：系统首先检查短期记忆（缓存）。如果信息因为近期使用过而被缓存，则直接加载到工作记忆，过程快速完成。
2. 检索长期记忆 ：如果在短期记忆中仍未找到，则触发对长期记忆的检索。这通常是一个成本较高的操作：
   • 索引查询 ：可能通过用户ID、会话ID等键进行快速查询。
   • 语义检索 ：更常见的是，将当前的工作记忆上下文（或提炼出的查询意图）向量化，然后在向量数据库中进行相似性搜索，找出历史上最相关的对话片段或知识条目。
3. 数据加载与更新 ：检索到的结果，一方面被注入到当前的工作记忆中供Agent使用，另一方面，根据策略可能会被写入短期记忆，以备后续快速访问。这个过程，像极了CPU需要的数据不在高速缓存中，于是去主内存加载，并可能根据预取策略更新缓存。

3.2 自下而上的“写回”与持久化

并非所有在工作记忆中产生的数据都需要永久保存。系统需要智能地决定哪些信息值得沉淀。

1. 重要性评估 ：在工作记忆处理过程中或任务结束时，系统（或由Agent自身）会对产生的信息进行重要性打分。例如，最终确认的用户需求、完成的任务结果、新学到的关键知识会获得高分；而中间推理的临时变量、无关紧要的寒暄则分数很低。
2. 写入短期缓存 ：高重要性、且预期短期内会再次用到的信息，会被写入短期记忆。例如，用户刚刚设置的首选项，在本次会话中很可能被反复引用。
3. 沉淀至长期存储 ：具有长期价值的信息，则会经过结构化处理（例如，提取实体、关系，生成摘要，转换为向量嵌入）后，被持久化到长期记忆层。例如，将一次完整的客户服务对话，总结为“用户X于X月X日反馈了Y问题，已通过Z方案解决，用户表示满意”，并关联相关知识点，存入数据库。

3.3 协同策略与一致性保障

多层架构带来了性能优势，也引入了数据一致性的挑战。

• 更新传播 ：当长期记忆中的基础信息被修正（例如，知识库更新），系统需要有机制（如失效标记、定期刷新）来通知上层缓存，避免Agent使用过时的信息。
• 冲突解决 ：在分布式或高并发场景下，可能出现对同一记忆条目的并发更新。系统需要定义简单的冲突解决策略，例如“最后写入获胜”，或为关键数据引入版本管理。
• 经验之谈 ：在实际开发中，我们往往会对“写”操作采取保守策略。即，工作记忆的写入可以很频繁，但向短期和长期记忆的写入，最好有明确的触发条件和去重逻辑，避免产生大量垃圾数据，拖慢检索速度。一个常见的技巧是，为记忆条目设计“热度”指标，综合访问频率、新鲜度和重要性，动态决定其在各层间的去留。

4. 在Agent Harness中的定位与集成

Gliding Horse 并非一个孤立的系统，它必须嵌入到更完整的 Agent Harness 中才能发挥最大价值。理解它与Agent核心及其他组件的边界至关重要。

4.1 与核心Agent的边界：专注与赋能

这是最需要厘清的关系。根据社区讨论的共识，Harness（包括Gliding Horse）是 基础设施层 ，而Agent是 核心推理逻辑 。

• Harness (Gliding Horse) 负责 ：“记什么”、“记在哪”、“怎么找”。它提供标准化的API，例如 save_memory(key, value, ttl, priority) 和 recall_memory(query, context) 。它不关心记忆内容的具体含义，只负责安全、高效、可靠地存储和检索。
• 核心Agent 负责 ：“用什么”、“怎么想”。Agent在推理时，决定何时调用 recall_memory API，以及如何处理检索回来的信息。它理解语义，做出决策。
• 一个比喻 ：Gliding Horse 是一个极其专业、高效的图书馆管理员。你（Agent）只需要告诉管理员“帮我找一些关于文艺复兴时期绘画与科学关系的资料”，管理员会运用他的分类法、索引系统（三层记忆架构）快速找到相关书籍并放到你的阅览桌（工作记忆）上。但如何阅读这些书、从中提炼出观点并写成论文，是你自己的事。管理员不替你做研究。

4.2 与RAG的协同关系

RAG（检索增强生成）是当前增强大模型知识的主流技术，它和Gliding Horse的长期记忆层功能上有重叠，但定位不同。

• RAG ：通常被视为一个 功能模块 或 技术手段 ，专注于解决“如何从海量外部知识中快速找到与当前问题最相关的片段”这一特定问题。它的输入是查询和文档库，输出是相关文本片段。
• Gliding Horse长期记忆 ：是一个 存储与检索的架构层 ，RAG可以是实现该层检索能力的一种（甚至是主要）技术选型。长期记忆层除了存储可供RAG检索的非结构化文档，还可能存储结构化的用户数据、会话日志等。
• 集成模式 ：在Gliding Horse架构下，当需要从长期记忆检索时，可以调用集成的RAG引擎。RAG引擎内部可能先通过向量数据库做语义检索，再通过关键词索引做精确匹配，最后将综合结果返回给记忆系统，再由记忆系统向上层传递。这样，RAG的强大检索能力就被纳入了统一、分层的记忆管理框架中。

4.3 完整的Harness生态系统视图

一个完整的Agent Harness通常包含以下组件，Gliding Horse是其中关键一环：

1. 记忆管理 (Gliding Horse) ：负责状态和知识的持久化与回溯。
2. 工具调用框架 ：标准化Agent调用外部API、函数的方式，管理工具的描述、认证和执行。
3. 任务规划与分解 ：将复杂目标拆解为可执行的子任务序列，并管理任务状态。
4. 对话与状态管理 ：管理多轮对话的流程，维护会话状态。
5. 监控与可观测性 ：记录Agent的决策日志、性能指标，便于调试和优化。

Gliding Horse 为其他组件提供记忆服务。例如，任务规划器需要从记忆中读取历史任务执行记录来优化新计划；对话管理器需要记忆上下文来维持连贯性。

5. 实操：设计并实现一个简化版三层记忆系统

理论聊了很多，我们来动手设计一个简化版的系统，看看核心代码逻辑如何体现三层架构的思想。这里我们以Python为例，使用内存字典模拟工作记忆，Redis作为短期记忆，Chroma向量数据库+SQLite作为长期记忆。

5.1 系统组件与接口定义

首先，我们定义核心的抽象接口和数据结构。

from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
from pydantic import BaseModel
import numpy as np

class MemoryItem(BaseModel):
    """记忆条目的基础模型"""
    id: str
    content: Any  # 可以是文本、字典等
    embedding: Optional[np.ndarray] = None  # 向量嵌入
    metadata: Dict[str, Any] = {}  # 如来源、时间、重要性分数
    access_count: int = 0
    last_accessed: float = 0.0  # 时间戳

class MemoryLayer(ABC):
    """记忆层的抽象基类"""
    @abstractmethod
    def write(self, item: MemoryItem) -> bool:
        """写入一个记忆条目"""
        pass

    @abstractmethod
    def read(self, query: str, top_k: int = 5) -> List[MemoryItem]:
        """根据查询读取相关记忆条目"""
        pass

    @abstractmethod
    def update(self, item_id: str, updates: Dict[str, Any]) -> bool:
        """更新记忆条目"""
        pass

5.2 各层具体实现

接下来，我们实现三个具体的层。

import time
from collections import OrderedDict

class WorkingMemoryLayer(MemoryLayer):
    """工作记忆层：基于内存的LRU缓存"""
    def __init__(self, capacity: int = 10):
        self.capacity = capacity
        self.cache = OrderedDict()  # id -> MemoryItem

    def write(self, item: MemoryItem) -> bool:
        item.last_accessed = time.time()
        if item.id in self.cache:
            self.cache.move_to_end(item.id)
        else:
            if len(self.cache) >= self.capacity:
                self.cache.popitem(last=False)  # 移除最久未使用的
            self.cache[item.id] = item
        return True

    def read(self, query: str, top_k: int = 5) -> List[MemoryItem]:
        # 工作记忆的“读取”更简单：直接返回最近使用的条目
        items = list(self.cache.values())
        items.sort(key=lambda x: x.last_accessed, reverse=True)
        return items[:top_k]

    def update(self, item_id: str, updates: Dict[str, Any]) -> bool:
        if item_id in self.cache:
            for key, value in updates.items():
                setattr(self.cache[item_id], key, value)
            self.cache[item_id].last_accessed = time.time()
            return True
        return False

import redis
import json
import pickle

class ShortTermMemoryLayer(MemoryLayer):
    """短期记忆层：基于Redis，带TTL"""
    def __init__(self, redis_url: str = "redis://localhost:6379", default_ttl: int = 3600):
        self.client = redis.from_url(redis_url)
        self.default_ttl = default_ttl

    def write(self, item: MemoryItem) -> bool:
        # 将MemoryItem序列化存储
        serialized = pickle.dumps(item.dict())
        # 使用一个Sorted Set记录访问热度，分数=last_accessed
        self.client.setex(f"memory:{item.id}", self.default_ttl, serialized)
        self.client.zadd("memory_access_rank", {item.id: item.last_accessed})
        return True

    def read(self, query: str, top_k: int = 5) -> List[MemoryItem]:
        # 简化版：返回最近访问过的条目（实际可根据query做简单关键词匹配）
        item_ids = self.client.zrevrange("memory_access_rank", 0, top_k-1)
        items = []
        for item_id in item_ids:
            data = self.client.get(f"memory:{item_id}")
            if data:
                item_dict = pickle.loads(data)
                items.append(MemoryItem(**item_dict))
        return items

    def update(self, item_id: str, updates: Dict[str, Any]) -> bool:
        data = self.client.get(f"memory:{item_id}")
        if not data:
            return False
        item_dict = pickle.loads(data)
        item = MemoryItem(**item_dict)
        for key, value in updates.items():
            setattr(item, key, value)
        # 写回并更新访问排名
        self.client.setex(f"memory:{item_id}", self.default_ttl, pickle.dumps(item.dict()))
        self.client.zadd("memory_access_rank", {item_id: item.last_accessed})
        return True

import chromadb
from chromadb.config import Settings
import sqlite3

class LongTermMemoryLayer(MemoryLayer):
    """长期记忆层：Chroma（向量检索）+ SQLite（元数据）"""
    def __init__(self, persist_dir: str = "./chroma_db"):
        # 初始化Chroma客户端，用于语义检索
        self.chroma_client = chromadb.PersistentClient(path=persist_dir)
        self.collection = self.chroma_client.get_or_create_collection(name="long_term_memories")
        # 初始化SQLite，用于精确查询和元数据管理
        self.conn = sqlite3.connect(f"{persist_dir}/metadata.db")
        self._init_db()

    def _init_db(self):
        cursor = self.conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS memory_metadata (
                id TEXT PRIMARY KEY,
                content TEXT,
                importance REAL DEFAULT 0.5,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                last_accessed TIMESTAMP,
                access_count INTEGER DEFAULT 0
            )
        ''')
        self.conn.commit()

    def write(self, item: MemoryItem) -> bool:
        # 1. 存入Chroma（如果有关联的向量）
        if item.embedding is not None:
            self.collection.add(
                embeddings=[item.embedding.tolist()],
                documents=[str(item.content)],
                metadatas=[item.metadata],
                ids=[item.id]
            )
        # 2. 存入SQLite
        cursor = self.conn.cursor()
        cursor.execute('''
            INSERT OR REPLACE INTO memory_metadata 
            (id, content, importance, last_accessed, access_count)
            VALUES (?, ?, ?, ?, ?)
        ''', (item.id, str(item.content), item.metadata.get('importance', 0.5),
              item.last_accessed, item.access_count))
        self.conn.commit()
        return True

    def read(self, query: str, top_k: int = 5) -> List[MemoryItem]:
        items = []
        # 策略1：先尝试从SQLite中做关键词匹配（简化示例）
        cursor = self.conn.cursor()
        cursor.execute('''
            SELECT * FROM memory_metadata 
            WHERE content LIKE ? 
            ORDER BY importance DESC, last_accessed DESC 
            LIMIT ?
        ''', (f'%{query}%', top_k))
        rows = cursor.fetchall()
        for row in rows:
            item = MemoryItem(
                id=row[0],
                content=row[1],
                metadata={'importance': row[2]},
                last_accessed=row[4],
                access_count=row[5]
            )
            items.append(item)

        # 策略2：如果关键词匹配结果少，则使用Chroma做语义检索（此处需预先计算query的embedding，略）
        # 实际应用中，两种检索方式的结果可以融合去重后返回
        return items[:top_k]

    def update(self, item_id: str, updates: Dict[str, Any]) -> bool:
        # 更新SQLite中的元数据
        cursor = self.conn.cursor()
        set_clause = ', '.join([f"{k}=?" for k in updates.keys()])
        values = list(updates.values())
        values.append(item_id)
        cursor.execute(f"UPDATE memory_metadata SET {set_clause} WHERE id=?", values)
        self.conn.commit()
        return cursor.rowcount > 0

5.3 记忆管理器的协同调度

最后，我们需要一个顶层的记忆管理器来协调三层之间的数据流动。

class GlidingHorseMemoryManager:
    """记忆管理器：协调三层记忆的读写"""
    def __init__(self):
        self.working_memory = WorkingMemoryLayer(capacity=15)
        self.short_term_memory = ShortTermMemoryLayer()
        self.long_term_memory = LongTermMemoryLayer()

    def remember(self, query: str, context: Dict[str, Any]) -> List[MemoryItem]:
        """核心回忆流程：模拟CPU的缓存访问层次"""
        all_results = []

        # 1. 检查工作记忆（最快）
        wm_results = self.working_memory.read(query, top_k=3)
        if wm_results:
            print(f"[工作记忆命中] 找到 {len(wm_results)} 条相关记忆。")
            all_results.extend(wm_results)

        # 2. 如果工作记忆不足，检查短期记忆
        if len(all_results) < 5:
            stm_results = self.short_term_memory.read(query, top_k=5 - len(all_results))
            if stm_results:
                print(f"[短期记忆命中] 找到 {len(stm_results)} 条相关记忆。")
                # 将短期记忆中找到的热点数据，加载一份到工作记忆
                for item in stm_results[:2]:  # 加载前两条到工作记忆
                    self.working_memory.write(item)
                all_results.extend(stm_results)

        # 3. 如果仍不足，检索长期记忆（最慢）
        if len(all_results) < 5:
            ltm_results = self.long_term_memory.read(query, top_k=5 - len(all_results))
            if ltm_results:
                print(f"[长期记忆检索] 找到 {len(ltm_results)} 条相关记忆。")
                # 将长期记忆检索到的结果，写入短期记忆缓存
                for item in ltm_results:
                    self.short_term_memory.write(item)
                all_results.extend(ltm_results)

        # 更新所有命中条目的访问记录
        for item in all_results:
            item.access_count += 1
            item.last_accessed = time.time()
            # 更新回原存储层（这里简化处理，实际需根据ID更新对应层）
            self._update_item_in_layer(item)

        return all_results[:5]  # 返回最多5条

    def memorize(self, content: Any, importance: float = 0.5) -> str:
        """记忆流程：决定信息存储在哪一层"""
        item_id = f"mem_{int(time.time())}_{hash(str(content)) % 10000}"
        item = MemoryItem(
            id=item_id,
            content=content,
            metadata={"importance": importance, "source": "user_input"},
            last_accessed=time.time()
        )

        # 根据重要性决定存储策略
        if importance > 0.8:  # 高重要性，存长期
            self.long_term_memory.write(item)
            print(f"信息已存入长期记忆: {item_id}")
        elif importance > 0.3:  # 中等重要性，存短期
            self.short_term_memory.write(item)
            print(f"信息已存入短期记忆: {item_id}")
        # 所有信息都尝试放入工作记忆（如果空间允许）
        self.working_memory.write(item)

        return item_id

    def _update_item_in_layer(self, item: MemoryItem):
        """内部方法：更新条目访问信息到其所在层"""
        # 简化实现：尝试更新所有层（实际应记录条目所在层）
        self.working_memory.update(item.id, {"access_count": item.access_count, "last_accessed": item.last_accessed})
        self.short_term_memory.update(item.id, {"access_count": item.access_count, "last_accessed": item.last_accessed})
        # 长期记忆更新成本高，可设定阈值，例如每访问5次才更新一次
        if item.access_count % 5 == 0:
            self.long_term_memory.update(item.id, {"access_count": item.access_count, "last_accessed": item.last_accessed})

# 使用示例
if __name__ == "__main__":
    manager = GlidingHorseMemoryManager()

    # 1. 记忆一些信息
    manager.memorize("用户Alice喜欢喝美式咖啡，不加糖。", importance=0.9)  # 高重要性，入长期
    manager.memorize("当前会话主题：讨论周末计划。", importance=0.6)  # 中重要性，入短期
    manager.memorize("上一条消息是问候。", importance=0.2)  # 低重要性，可能只在工作记忆

    # 2. 回忆信息
    print("\n--- 尝试回忆与‘咖啡’相关的信息 ---")
    memories = manager.remember("咖啡", {})
    for mem in memories:
        print(f"  - {mem.content} (重要性: {mem.metadata.get('importance')})")

这个简化实现清晰地展示了几点：

1. 分层存储 ：不同重要性和访问模式的数据被自动分配到不同层级。
2. 缓存回填 ：从深层记忆检索到的数据，会自动填充到上层缓存（短期->工作，长期->短期），加速后续访问。
3. 访问模式记录 ：通过 access_count 和 last_accessed ，为未来的缓存淘汰和重要性重评估提供了数据基础。

6. 性能调优与生产级考量

将原型投入生产环境，会面临一系列新的挑战。以下是基于经验的一些关键调优点和考量。

6.1 各层容量与淘汰策略的权衡

三层架构的性能优势，高度依赖于各层容量和淘汰策略的精细配置。

• 工作记忆容量 ：太小会导致频繁的缓存失效，Agent思维容易“断片”；太大会占用过多运行时内存，且可能降低检索效率（因为要在更大的无序集合中查找）。通常建议设置为能容纳当前复杂任务分解后所有关键上下文的大小，例如对应一个中等长度对话的精华摘要，容量在5-20个条目较为常见。
• 短期记忆TTL与淘汰 ：Redis的过期时间（TTL）和内存淘汰策略需要结合业务场景。对于客服机器人，用户本次会话的上下文可能需要在会话结束后保留一段时间（例如30分钟），TTL可据此设置。淘汰策略推荐使用 volatile-lru ，在内存不足时优先淘汰最近最少使用的、且设置了过期时间的键。
• 长期记忆的索引优化 ：这是性能瓶颈最可能出现的环节。对于向量数据库，需要关注：
  • 索引类型 ：HNSW（近似最近邻）在精度和速度之间取得了较好平衡，适用于大多数场景。
  • 向量维度 ：选择与你的嵌入模型匹配的维度，不是越高越好。例如，使用 text-embedding-3-small 模型就是1536维。
  • 分区与过滤 ：为记忆条目添加丰富的元数据（如 user_id , session_id , topic ），在检索时先通过元数据过滤，可以大幅缩小向量搜索范围，提升效率。

6.2 检索质量与混合搜索策略

单纯依靠向量相似性搜索，可能会返回语义相关但事实错误的“幻觉”信息，或者漏掉关键词完全匹配的重要条目。

• 混合检索 ：结合 语义搜索（向量） 和 关键词搜索（倒排索引） 是业内的最佳实践。例如，可以先用关键词快速筛选出候选集，再用向量搜索在候选集内进行精排。或者将两种检索方式的得分进行加权融合（如 Hybrid Score = α * 向量相似度 + β * BM25分数 ）。
• 重排序 ：在初步检索出Top-K（例如20条）结果后，可以使用一个更精细但更耗资源的“重排序器”模型对结果进行二次排序，进一步提升最相关结果的位置。这对于将最关键的记忆提供给工作记忆至关重要。
• 查询理解与扩展 ：在将用户查询送入检索系统前，可以先让大模型对查询进行改写、扩展或提炼。例如，将“它怎么用？”根据上下文扩展为“Gliding Horse记忆系统如何使用？”，能显著提升检索准确性。

6.3 分布式环境下的挑战

当AI Agent服务需要水平扩展时，记忆系统面临状态同步问题。

• 无状态Agent与共享记忆 ：最佳实践是保持Agent推理实例的无状态化。所有记忆的读写都通过一个共享的、高可用的记忆服务（即Gliding Horse架构的实现）来完成。这要求记忆服务本身是分布式的。
• 记忆服务的分布式设计 ：
  • 短期记忆层（Redis） ：可以使用Redis Cluster模式实现高可用和分片存储。
  • 长期记忆层（向量数据库） ：选择支持分布式的向量数据库，如 Milvus、Weaviate Cluster，或者使用支持分区的PGVector。
  • 数据分片策略 ：最常见的分片键是 user_id 或 tenant_id ，确保同一个用户的所有记忆操作都被路由到同一个分片，避免跨分片事务。
• 最终一致性 ：在分布式系统中，强一致性往往代价高昂。对于记忆系统，通常可以接受最终一致性。例如，Agent A写入了一条记忆，Agent B可能在极短时间内读取不到，但这在大多数对话场景下是可接受的。需要明确业务对一致性的要求级别。

7. 典型应用场景与效果评估

Gliding Horse这类分层记忆架构，在特定的复杂应用场景下能展现出巨大价值。

7.1 场景一：深度个性化AI助手

想象一个陪伴你数月的AI个人助手，它需要记住你的饮食习惯、工作习惯、家庭成员的生日、你正在读的书、以及上周你们讨论过的某个晦涩哲学概念。

• 工作记忆 ：存放当前对话中你刚刚提到的“今晚想吃的菜系”和“冰箱里现有的食材”。
• 短期记忆 ：缓存你本周的饮食偏好、最近在追的剧，以及昨天你询问过的某个菜谱步骤。
• 长期记忆 ：存储你所有历史对话中提取的长期偏好、过敏史、重要日期，以及经过总结的深度话题讨论纪要。
• 效果 ：当你问“今晚吃什么？”时，它能结合工作记忆（刚才的对话）、短期记忆（本周偏好）和长期记忆（过敏史、口味）给出高度个性化的建议，而不是一个通用回答。对话连贯性从“轮次级”提升到“天级”甚至“月级”。

7.2 场景二：复杂任务驱动的自治Agent

一个需要自主研究某个技术课题并撰写报告的AI Agent。

• 工作记忆 ：存放当前正在执行的子任务（如“搜索关于XXX的最新论文”）、上一步的结果摘要、下一步的计划。
• 短期记忆 ：缓存本次研究任务中已经阅读过的论文摘要、访问过的有用网页链接、初步形成的观点大纲。
• 长期记忆 ：存储该Agent历史上所有研究任务的元数据、学到的通用研究方法论、以及从过往报告中提炼出的高质量写作模板。
• 效果 ：Agent在执行任务时不会“忘记”最初的目标，能有效引用自己在前几步中发现的信息，并能借鉴历史成功经验来优化当前的任务规划。任务完成的可靠性和质量显著提高。

7.3 评估指标与监控

如何衡量一个记忆系统的好坏？不能只看感觉，需要可量化的指标。

• 核心性能指标 ：
  • 回忆准确率 ：Agent需要某信息时，系统能正确提供该信息的比例。可通过人工评估或基于已知知识库的自动化测试来衡量。
  • 回忆延迟 ：从Agent发出查询到收到记忆结果的平均时间。分层架构的目标就是让高频访问的记忆延迟极低（<10ms），低频访问的记忆延迟在可接受范围内（<200ms）。
  • 缓存命中率 ：工作记忆和短期记忆的命中率。高命中率意味着系统有效减少了昂贵的长期记忆检索。
• 业务效果指标 ：
  • 任务完成率 ：对于任务型Agent，配备记忆系统后，复杂多步任务的完成率是否有提升。
  • 用户满意度 ：在对话型应用中，用户对AI“记忆力”的主观评分或相关负面反馈（如“你刚才已经问过这个了”）的减少比例。
• 系统健康度监控 ：
  • 各层存储的使用量（内存/磁盘）。
  • 长期记忆检索的QPS和平均响应时间。
  • 向量索引的构建状态和健康度。

部署后，需要持续监控这些指标，并根据数据反馈调整各层容量、淘汰策略和检索参数，使系统在实际负载下达到最佳平衡。

8. 常见陷阱与进阶优化方向

即使理解了架构，在实际落地时依然会踩很多坑。这里分享一些从实践中总结的教训和更深入的思考。

8.1 典型陷阱与规避方案

1. 记忆泛滥与污染

   • 问题 ：不加选择地将所有对话内容都存入长期记忆，导致记忆库迅速膨胀，检索速度下降，且大量低价值、重复或矛盾的信息污染了检索结果。
   • 规避 ：实施严格的记忆写入过滤器。除了基于重要性的打分，还可以结合规则：例如，仅存储包含特定实体（人名、项目名）或用户明确指令（“记住这个”）的对话；对内容进行去重和摘要化处理后再存储。

2. 上下文幻觉与记忆冲突

   • 问题 ：当长期记忆中关于同一事实存在多个版本时，检索系统可能返回过时或冲突的信息，导致Agent产生混淆。
   • 规避 ：为记忆条目引入 版本管理 或 置信度 。新记忆可以覆盖旧记忆，但保留历史版本以备审计。或者，在检索到冲突信息时，将冲突本身作为上下文提供给Agent，让Agent结合时间戳和来源可信度进行判断。

3. “缓存穿透”与“缓存雪崩”

   • 问题 ：这是从传统缓存系统继承来的经典问题。“穿透”指查询一个不存在的数据，请求直接打到昂贵的长期记忆层。“雪崩”指短期记忆（Redis）大量键同时过期，导致瞬间所有请求涌向长期记忆，造成宕机。
   • 规避 ：对于“穿透”，可以在短期记忆层为“空结果”也设置一个很短的缓存时间。对于“雪崩”，为Redis的TTL设置一个随机波动值（例如，基础TTL是1小时，实际TTL在55-65分钟之间随机），避免同时失效。

4. 向量检索的“语义漂移”

   • 问题 ：单纯依赖向量相似度，可能会检索到语义相关但主题偏离的条目。例如，查询“苹果公司”，可能检索到关于“水果苹果”的营养文章。
   • 规避 ：必须结合 元数据过滤 。在检索“苹果公司”时，强制添加元数据过滤器 topic: “technology” 或 entity_type: “company” 。这要求在设计记忆条目时，就要有良好的元数据标注体系。

8.2 进阶优化方向

1. 记忆的主动管理与遗忘 当前的系统主要是被动的“存”和“取”。更高级的系统应具备主动管理能力：

   • 记忆压缩与摘要 ：定期对长期记忆中关于同一主题的琐碎记忆进行自动摘要，生成一个更精炼、更高阶的记忆条目，并归档原始细节。
   • 记忆衰减与遗忘 ：引入类似人脑的遗忘曲线。长时间不被访问的记忆，其“重要性”分数应缓慢衰减。当分数低于某个阈值时，可以将其迁移到更廉价的归档存储，或进行摘要后删除原始内容。这能保持核心记忆库的“健康度”。

2. 基于记忆的预测与预热 系统可以学习用户的访问模式，进行预测性缓存。

   • 模式学习 ：分析历史日志，发现规律。例如，用户每次询问“本周日程”后，有很大概率会接着问“明天天气”。
   • 预加载 ：当检测到用户开始询问日程时，系统可以异步预加载与该用户相关的天气信息到短期记忆，当用户真的问到时，实现“零延迟”响应。

3. 记忆间的关联与图谱化 目前的记忆条目大多是孤立的。更强大的记忆系统应能建立记忆之间的联系。

   • 构建记忆图谱 ：利用知识图谱技术，自动或半自动地从记忆条目中抽取实体（人、事、物、概念）和关系，形成一张私有的记忆图谱。
   • 关联检索 ：当检索到一条关于“项目A”的记忆时，系统可以同时返回与之关联的“成员B”、“技术栈C”等相关记忆，为Agent提供更丰富的上下文，激发更深入的推理。

Gliding Horse提出的三层架构，为AI记忆系统提供了一个坚实而优雅的蓝图。它让我们意识到，赋予AI记忆不仅仅是扩大上下文窗口，而是需要一套像计算机科学一样严谨的体系结构来管理信息的生命周期、访问速度和存储成本。实现它固然有挑战，但沿着这个方向探索，无疑是让AI从“即时应答机”走向“持久智能体”的关键一步。在实际项目中，我建议从最迫切的痛点入手，先实现一个简化版本，再随着业务复杂度的提升，逐步迭代和完善记忆层的各项高级功能。