企业官网建设流程全解析

DB-GPT 长期记忆（Long-term Memory）深度实践指南：从向量存储到时间加权检索

【免费下载链接】DB-GPTopen-source agentic AI data assistant for the next generation of AI + Data products.项目地址: https://gitcode.com/GitHub_Trending/db/DB-GPT

本篇指南围绕 DB-GPT 的 Long-term Memory 模块展开，系统讲解长期记忆与短期记忆的分工关系、基于向量存储的落地方式、Embedding 模型与向量库的准备步骤，以及LongTermMemory的完整使用方式与底层时间加权检索原理。读完本文，你将掌握在 DB-GPT 中为 Agent 配置持久化长期记忆的完整实操链路，并能根据源码理解重要性（importance）、时间衰减（decay）等参数对记忆召回的影响。

长期记忆在 Agent 记忆体系中的定位

在 DB-GPT 的 Agent 记忆设计中，记忆被划分为三个层级：感觉记忆（Sensory Memory）、短期记忆（Short-term Memory）与长期记忆（Long-term Memory），其设计灵感来自人类记忆从感知输入、瞬时保持到长期巩固的递进过程（参见 base.py 模块注释）。

原文档给出了清晰的界定：

• 短期记忆包含 Agent 当前所处情境的上下文信息，在大多数场景下等同于 LLM 上下文窗口内的输入内容，通常会被写入 Agent 的 prompt 中；
• 长期记忆存储 Agent 过去的行为与思考，可根据当前事件被检索召回；
• 长期记忆在形态上类似于外部向量存储，Agent 可以按需快速查询与检索。

在 DB-GPT 中，长期记忆默认存储在向量存储（vector storage）中，这意味着记忆以文本片段经 Embedding 模型向量化后的形式持久化保存，具备跨会话、跨 Agent 复用的能力。这一点与仅保存在进程内存中的ShortTermMemory形成鲜明对比——后者的实现就是一个带buffer_size的 FIFO 缓冲区（见 short_term.py）。

使用长期记忆的前置条件

要使用长期记忆，必须提供两样东西：

1. 一个Embedding 模型：负责将记忆文本转换为向量；
2. 一个向量存储（Vector Store）：负责向量化记忆的持久化保存与相似度检索。

第一步：准备 Embedding 模型

Embedding 模型用于把文本转换为向量，DB-GPT 通过DefaultEmbeddingFactory提供多种创建方式（实现见 embedding_factory.py）：

• DefaultEmbeddingFactory.openai(...)：调用 OpenAI 兼容的 Embedding API（默认模型text-embedding-3-small）；
• DefaultEmbeddingFactory.default(...)：加载本地模型（如 text2vec）；
• DefaultEmbeddingFactory.remote(...)：连接已部署的 DB-GPT Embedding API Server。

原文档以 OpenAI Embedding API 为例：

import os from dbgpt.rag.embedding import DefaultEmbeddingFactory api_url = os.getenv("OPENAI_API_BASE", "https://api.openai.com/v1") + "/embeddings" api_key = os.getenv("OPENAI_API_KEY") embeddings = DefaultEmbeddingFactory.openai(api_url=api_url, api_key=api_key)

如果希望使用本地模型，也可以直接替换为DefaultEmbeddingFactory.default("/data/models/text2vec-large-chinese")，详细的 Embedding 模型准备方式可参考 短期记忆文档 中的多方案对比。

第二步：准备向量存储

DB-GPT 的向量存储层支持多种实现。在dbgpt-ext的 storage/vector_store 目录下，可以看到ChromaStore、MilvusStore、WeaviateStore、QdrantStore、PGVectorStore、ElasticStore等实现，可按实际部署环境选用。

原文档以ChromaStore为例，首先安装 chroma 依赖：

pip install chromadb

然后创建向量存储实例：

import shutil from dbgpt_ext.storage.vector_store.chroma_store import ChromaVectorConfig, ChromaStore # 删除旧的向量存储目录(/tmp/tmp_ltm_vector_store) shutil.rmtree("/tmp/tmp_ltm_vector_store", ignore_errors=True) vector_store = ChromaStore( vector_store_config=ChromaVectorConfig( persist_path="/tmp/tmp_ltm_vector_store", ), name="ltm_vector_store", embedding_fn=embeddings, )

参数说明：

• ChromaVectorConfig.persist_path：向量数据持久化目录，默认读取环境变量CHROMA_PERSIST_PATH，未设置时可为None（对应 chroma_store.py 中的ChromaVectorConfig定义）；
• name：集合（collection）名称，即本次长期记忆在向量库中的命名空间，这里使用"ltm_vector_store"；
• embedding_fn：上一步创建好的 Embedding 对象，Chroma 在写入与检索时使用它完成向量化。

代码开头通过shutil.rmtree(..., ignore_errors=True)清理旧目录，是为了保证示例可重复运行、避免历史数据干扰。

创建并接入长期记忆

准备好向量存储后，即可创建LongTermMemory并包装为AgentMemory供 Agent 使用：

from concurrent.futures import ThreadPoolExecutor from dbgpt.agent import AgentMemory, LongTermMemory # 创建一个包含长期记忆的 Agent 记忆 memory = LongTermMemory( executor=ThreadPoolExecutor(), vector_store=vector_store, _default_importance=0.5 ) agent_memory: AgentMemory = AgentMemory(memory=memory)

这里_default_importance表示一条记忆片段的默认重要程度。因为此处直接使用LongTermMemory（没有经过感觉记忆/短期记忆的自动打分流程），所以需要显式指定默认重要性。从 long_term.py 的write()实现可以看到：当写入的记忆片段自身未携带importance时，会回退使用_default_importance，若两者都为空则抛出ValueError。

LongTermMemory 构造函数参数解析

对照源码（long_term.py），LongTermMemory.__init__支持以下参数：

此外，类属性importance_weight: float = 0.15定义了重要性在记忆召回综合评分中的权重系数（该权重在 base.py 的score_memory_importance中参与计算）。

记忆写入与元数据设计

LongTermMemory.write()的写入流程（见 long_term.py）非常直观：

1. 读取记忆片段的重要性（缺失时回退到_default_importance）；
2. 将重要性、最后访问时间（last_accessed_at）、会话 ID（session_id）等写入元数据；
3. 把原始观察构造成Chunk，内容格式为"[{memory_idx}] {raw_observation}"；
4. 通过blocking_func_to_async在线程池中调用memory_retriever.load_document写入向量库。

因此，向量库中每条长期记忆都携带importance、last_accessed_at、buffer_idx、memory_type等元数据字段，这些字段是后续时间加权检索的重要依据。write_batch()支持批量写入，将短期记忆中溢出的记忆批量落盘到长期记忆。

长期记忆的检索原理：时间加权 + 语义相似度

LongTermMemory内部通过LongTermRetriever（继承自TimeWeightedEmbeddingRetriever，实现见 time_weighted.py）完成检索。fetch_memories()会先按当前memory_type、session_id等元数据构造MetadataFilters，再调用检索器召回相关片段。

检索的综合评分融合了两个维度：

• 语义相关性：向量库的similar_search_with_scores返回的相似度得分；
• 时间衰减：基于last_accessed_at计算经过的小时数hours_passed，时间分数为(1.0 - decay_rate) ** hours_passed，即访问越久远的记忆得分越低。

在向量库独立模式（_retrieve_vector_store_only）下，还会额外叠加重要性得分：combined_score = doc.score + time_score + importance_score（见 long_term.py）。这正是原文档强调“长期记忆可根据当前事件检索召回”的底层机制——它不只是纯粹的向量相似度，而是相关性 + 时间新鲜度 + 重要性的加权组合。

decay_rate默认值为0.01（时间加权检索器构造参数），决定记忆随时间的遗忘速度；top_k默认 100，最终返回的_k默认为 4 条。检索时还会跳过内容中带有[FORGET]、[MERGE]占位符的片段，为将来的遗忘与合并机制预留了接口。

在 Agent 中集成长期记忆

LongTermMemory通过AgentMemory暴露给 Agent。AgentMemory（见 agent_memory.py）是 Agent 记忆的统一门面，其构造函数接受memory、importance_scorer、insight_extractor、gpts_memory等参数：

• memory：实际承载记忆的存储器，传入的即是我们创建的LongTermMemory；未传入时默认使用buffer_size=5的ShortTermMemory；
• gpts_memory：管理计划与消息记忆，默认创建GptsMemory；
• AgentMemory提供write、write_batch、read、clear、initialize、structure_clone等统一方法，并暴露plans_memory、message_memory两个属性。

值得留意的是，在实际的 Agent 链路中，更常见的做法是使用HybridMemory（见 hybrid.py）将感觉记忆、短期记忆、长期记忆组合起来：短期记忆的transfer_to_long_term()策略会在缓冲区溢出时把最旧的记忆转移给长期记忆（见 base.py 中ShortTermMemory的溢出转移逻辑）。也就是说，长期记忆既是 Agent 跨会话记忆的“持久层”，也是短期记忆溢出的“归处”。

完整代码示例

将原文档中的步骤串联起来，即可得到一份可直接运行的完整示例：

import os import shutil from concurrent.futures import ThreadPoolExecutor from dbgpt.agent import AgentMemory, LongTermMemory from dbgpt.rag.embedding import DefaultEmbeddingFactory from dbgpt_ext.storage.vector_store.chroma_store import ChromaVectorConfig, ChromaStore # 1. 准备 Embedding 模型（OpenAI Embedding API） api_url = os.getenv("OPENAI_API_BASE", "https://api.openai.com/v1") + "/embeddings" api_key = os.getenv("OPENAI_API_KEY") embeddings = DefaultEmbeddingFactory.openai(api_url=api_url, api_key=api_key) # 2. 准备向量存储（ChromaStore） shutil.rmtree("/tmp/tmp_ltm_vector_store", ignore_errors=True) vector_store = ChromaStore( vector_store_config=ChromaVectorConfig( persist_path="/tmp/tmp_ltm_vector_store", ), name="ltm_vector_store", embedding_fn=embeddings, ) # 3. 创建长期记忆并包装为 AgentMemory memory = LongTermMemory( executor=ThreadPoolExecutor(), vector_store=vector_store, _default_importance=0.5 ) agent_memory: AgentMemory = AgentMemory(memory=memory)

小结

DB-GPT 的长期记忆是一套“外部向量存储 + 时间加权检索”的实现：通过 Embedding 模型将 Agent 的历史观察向量化并持久化到向量库，再以LongTermMemory提供写入、批量写入与语义检索能力，检索时综合语义相关度、时间新鲜度与重要性三者加权排序，从而让 Agent 能够跨会话记住并召回过去的经验。

实践要点回顾：

1. 长期记忆必须搭配向量存储使用，默认落在向量库中；
2. 直接使用LongTermMemory时必须通过_default_importance指定记忆片段的默认重要性；
3. AgentMemory是接入 Agent 的统一入口，长期记忆与短期记忆可通过HybridMemory组合联动；
4. 除 Chroma 外，DB-GPT 还支持 Milvus、Weaviate、Qdrant、PGVector、Elasticsearch 等多种向量存储后端，可按部署环境替换。

【免费下载链接】DB-GPTopen-source agentic AI data assistant for the next generation of AI + Data products.项目地址: https://gitcode.com/GitHub_Trending/db/DB-GPT