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。使用长期记忆的前置条件要使用长期记忆必须提供两样东西一个Embedding 模型负责将记忆文本转换为向量一个向量存储Vector Store负责向量化记忆的持久化保存与相似度检索。第一步准备 Embedding 模型Embedding 模型用于把文本转换为向量DB-GPT 通过DefaultEmbeddingFactory提供多种创建方式实现见 embedding_factory.pyDefaultEmbeddingFactory.openai(...)调用 OpenAI 兼容的 Embedding API默认模型text-embedding-3-smallDefaultEmbeddingFactory.default(...)加载本地模型如 text2vecDefaultEmbeddingFactory.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_urlapi_url, api_keyapi_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_errorsTrue) vector_store ChromaStore( vector_store_configChromaVectorConfig( persist_path/tmp/tmp_ltm_vector_store, ), nameltm_vector_store, embedding_fnembeddings, )参数说明ChromaVectorConfig.persist_path向量数据持久化目录默认读取环境变量CHROMA_PERSIST_PATH未设置时可为None对应 chroma_store.py 中的ChromaVectorConfig定义name集合collection名称即本次长期记忆在向量库中的命名空间这里使用ltm_vector_storeembedding_fn上一步创建好的 Embedding 对象Chroma 在写入与检索时使用它完成向量化。代码开头通过shutil.rmtree(..., ignore_errorsTrue)清理旧目录是为了保证示例可重复运行、避免历史数据干扰。创建并接入长期记忆准备好向量存储后即可创建LongTermMemory并包装为AgentMemory供 Agent 使用from concurrent.futures import ThreadPoolExecutor from dbgpt.agent import AgentMemory, LongTermMemory # 创建一个包含长期记忆的 Agent 记忆 memory LongTermMemory( executorThreadPoolExecutor(), vector_storevector_store, _default_importance0.5 ) agent_memory: AgentMemory AgentMemory(memorymemory)这里_default_importance表示一条记忆片段的默认重要程度。因为此处直接使用LongTermMemory没有经过感觉记忆/短期记忆的自动打分流程所以需要显式指定默认重要性。从 long_term.py 的write()实现可以看到当写入的记忆片段自身未携带importance时会回退使用_default_importance若两者都为空则抛出ValueError。LongTermMemory 构造函数参数解析对照源码long_term.pyLongTermMemory.__init__支持以下参数参数类型说明executorExecutor线程池执行器用于把阻塞的向量库写入/检索操作异步化vector_storeVectorStoreBase承载记忆的向量存储实例nowOptional[datetime]当前时间用于时间加权计算默认datetime.now()reflection_thresholdOptional[float]反思阈值聚合重要性达到该值后触发反思当前实现中预留_default_importanceOptional[float]默认重要性直接使用LongTermMemory时必须设置metadataOptional[Dict]附加元数据默认{memory_type: long_term}写入时一并存入向量库此外类属性importance_weight: float 0.15定义了重要性在记忆召回综合评分中的权重系数该权重在 base.py 的score_memory_importance中参与计算。记忆写入与元数据设计LongTermMemory.write()的写入流程见 long_term.py非常直观读取记忆片段的重要性缺失时回退到_default_importance将重要性、最后访问时间last_accessed_at、会话 IDsession_id等写入元数据把原始观察构造成Chunk内容格式为[{memory_idx}] {raw_observation}通过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_size5的ShortTermMemorygpts_memory管理计划与消息记忆默认创建GptsMemoryAgentMemory提供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_urlapi_url, api_keyapi_key) # 2. 准备向量存储ChromaStore shutil.rmtree(/tmp/tmp_ltm_vector_store, ignore_errorsTrue) vector_store ChromaStore( vector_store_configChromaVectorConfig( persist_path/tmp/tmp_ltm_vector_store, ), nameltm_vector_store, embedding_fnembeddings, ) # 3. 创建长期记忆并包装为 AgentMemory memory LongTermMemory( executorThreadPoolExecutor(), vector_storevector_store, _default_importance0.5 ) agent_memory: AgentMemory AgentMemory(memorymemory)小结DB-GPT 的长期记忆是一套“外部向量存储 时间加权检索”的实现通过 Embedding 模型将 Agent 的历史观察向量化并持久化到向量库再以LongTermMemory提供写入、批量写入与语义检索能力检索时综合语义相关度、时间新鲜度与重要性三者加权排序从而让 Agent 能够跨会话记住并召回过去的经验。实践要点回顾长期记忆必须搭配向量存储使用默认落在向量库中直接使用LongTermMemory时必须通过_default_importance指定记忆片段的默认重要性AgentMemory是接入 Agent 的统一入口长期记忆与短期记忆可通过HybridMemory组合联动除 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
