DB-GPT Agent 记忆模块深入指南从记忆结构到自定义读写实战【免费下载链接】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 Agent 框架中的 memory 模块为主线系统讲解 Agent 的三大记忆操作读取、写入、反思与四层记忆结构感觉记忆、短期记忆、长期记忆、混合记忆并结合仓库源码剖析Memory、MemoryFragment、AgentMemory、GptsMemory等核心类的设计原理。读完本文你将掌握在 DB-GPT 中创建与管理 Agent 记忆、通过 prompt 模板注入most_recent_memories、以及继承ConversableAgent自定义记忆读写策略的完整实战方法。一、记忆模块在 Agent 架构中的定位在 DB-GPT 的 Agent 架构设计中记忆模块扮演着至关重要的角色它存储从环境中感知到的信息并利用这些记录去辅助未来的行动决策。记忆模块能够帮助 Agent 持续积累经验、实现自我进化并以更一致、更合理、更高效的方式行事。从仓库源码结构看记忆模块的完整实现位于packages/dbgpt-core/src/dbgpt/agent/core/memory/目录核心抽象定义在 base.py具体实现分散在short_term.py、long_term.py、hybrid.py、agent_memory.py与gpts/子目录中。1.1 记忆的三大操作Memory Operations在 DB-GPT Agents 中记忆操作分为三类记忆读取Memory Reading从记忆中提取有意义的信息用以增强 Agent 的行动能力。记忆写入Memory Writing将感知到的环境信息存入记忆。有价值的信息被存储下来为未来检索有信息量的记忆提供基础使 Agent 能更高效、更理性地行动。记忆反思Memory Reflection模拟人类对自身认知、情感和行为过程的审视与评估能力。迁移到 Agent 上其目标是让 Agent 具备独立总结并推断更抽象、更复杂、更高层信息的能力。源码层面Memory抽象基类为这三个操作定义了统一接口read(observation, alpha, beta, gamma)、write(memory_fragment, now, op)与reflect(memory_fragments)见 base.py。1.2 记忆的四层结构Memory StructureDB-GPT Agents 中定义了四种主要记忆结构设计上对标人类记忆的渐进模型源码注释中对这一设计有明确说明见 base.py 文件头感觉记忆Sensory Memory如同人类感觉记忆用于登记感知输入接收来自环境的观察其中部分会被转移到短期记忆。其importance_weight默认为0.9threshold_to_short_term默认为0.1。短期记忆Short-term Memory临时缓冲最近的感知接收来自感觉记忆的部分内容并可通过其他观察或检索到的记忆得到增强进而进入长期记忆。长期记忆Long-term Memory存储 Agent 的经验与知识接收短期记忆的信息并随时间整合重要信息。混合记忆Hybrid Memory感觉记忆、短期记忆与长期记忆的组合体显式建模人类短时与长时记忆机制长期记忆负责随时间巩固重要信息。二、记忆核心概念解析在进入编码之前先理清 DB-GPT Agent 记忆体系中几个关键概念它们全部可以在源码中找到对应实现Memory存储所有记忆的类目前可以是SensorMemory、ShortTermMemory、EnhancedShortTermMemory、LongTermMemory和HybridMemory。它是一个泛型抽象基类Memory(ABC, Generic[T])声明了read、write、write_batch、clear、structure_clone等抽象方法并内置了score_memory_importance重要度打分与get_insights洞察提取两个能力钩子见 base.py。MemoryFragment存储记忆信息的抽象类是记忆的基本单元包含观察内容observation、嵌入向量embeddings、记忆 id、重要度importance、是否洞察is_insight、最后访问时间last_accessed_time等基础信息。其id通常由雪花算法生成因此可以从 id 反解出记忆片段的创建时间戳见 base.py。AgentMemoryFragment继承自MemoryFragment的默认实现在 agent_memory.py 中定义用于承载 Agent 的对话与行动记忆。仓库中还提供了StructuredAgentMemoryFragment结构化版本将观察序列化为 JSON 存储供需要结构化记忆的场景使用。GptsMemory用于存储对话conversation与计划plan信息从记忆结构的角度看它不属于记忆体系而是作为会话层的信息载体见 gpts_memory.py。AgentMemory同时包含Memory与GptsMemory的门面类是 Agent 与底层记忆交互的统一入口。其构造逻辑是未显式传入memory时默认创建ShortTermMemory(buffer_size5)未传入gpts_memory时默认创建GptsMemory()见 agent_memory.py。AgentMemory上还暴露了两个便捷属性plans_memory与message_memory分别指向gpts_memory内部的计划记忆与消息记忆方便直接读写会话/计划信息。三、创建 Agent 记忆从最小示例到完整配置3.1 创建带默认短期记忆的 AgentMemory正如前文所述记忆都包含在AgentMemory类中。最简单的创建方式如下from dbgpt.agent import AgentMemory, ShortTermMemory # Create an agent memory, default memory is ShortTermMemory memory ShortTermMemory(buffer_size5) agent_memory AgentMemory(memorymemory)这里buffer_size5表示短期记忆缓冲区最多保留 5 条记忆片段。从 short_term.py 的ShortTermMemory实现可以看到写入时会先做去重合并handle_duplicated当缓冲区溢出时通过transfer_to_long_term把最旧的记忆片段转存到长期记忆从而让短期记忆始终只保留最近的观察。3.2 附带 GptsMemory 的完整 AgentMemory常规理解下GptsMemory不属于记忆结构它专门用于存放对话与计划信息。你可以在AgentMemory中传入一个GptsMemoryfrom dbgpt.agent import AgentMemory, ShortTermMemory, GptsMemory # Create an agent memory, default memory is ShortTermMemory memory ShortTermMemory(buffer_size5) # Store the conversation and plan information gpts_memory GptsMemory() agent_memory AgentMemory(memorymemory, gpts_memorygpts_memory)GptsMemory在 gpts_memory.py 中维护plans_memory默认DefaultGptsPlansMemory与message_memory默认DefaultGptsMessageMemory两块存储并负责会话消息的缓存、消息队列推送以及可视化消息VIS的组装是 Agent 多轮对话与多步计划得以持久化的底层支撑。3.3 进阶创建混合记忆当需要同时具备短期记忆的实时性与长期记忆的持久性时可以使用HybridMemory。它支持通过from_chroma工厂方法一键创建默认使用 OpenAI Embedding API 与 ChromaStore或通过from_vstore基于自定义向量库创建相关用法可参考 hybrid_memory.md 以及实现文件 hybrid.py。混合记忆的读写采用流水线式策略写入时先进入感觉记忆溢出后转存短期记忆短期记忆中被多次增强enhance的片段会合并为高层洞察写入长期记忆读取时则先从长期记忆经TimeWeightedEmbeddingRetriever按时间加权召回再回写短期记忆以增强后续推理。四、在 Agent 中读写记忆完整实战示例Agent 会调用read_memories方法从记忆中读取记忆片段并调用write_memories方法把记忆片段写入记忆。当 Agent 调用 LLM 时记忆会被写入 LLM promptLLM 返回响应后Agent 又会把查询和响应写回记忆。正如 Profile To Prompt 中介绍的prompt 模板中存在一个名为most_recent_memories的模板变量它会被替换为最近读取到的记忆内容。4.1 读取记忆并构建 Prompt下面是一个完整的读取记忆并构建 prompt 的示例来自官方文档可直接运行。该示例创建了一个名为 Joy 的喜剧演员 Agent通过两轮对话验证记忆注入的效果import os import asyncio from dbgpt.agent import ( AgentContext, ShortTermMemory, AgentMemory, ConversableAgent, ProfileConfig, LLMConfig, BlankAction, UserProxyAgent, ) from dbgpt.model.proxy import OpenAILLMClient llm_client OpenAILLMClient( model_aliasgpt-4o, api_baseos.getenv(OPENAI_API_BASE), api_keyos.getenv(OPENAI_API_KEY), ) context: AgentContext AgentContext( conv_idtest123, languageen, temperature0.9, max_new_tokens2048, verboseTrue, # Add verboseTrue to print out the conversation history ) # Create an agent memory, which contains a short-term memory memory ShortTermMemory(buffer_size2) agent_memory: AgentMemory AgentMemory(memorymemory) # Custom user prompt template, which includes most recent memories and question user_prompt_template \ {% if most_recent_memories %}\ Most recent observations: {{ most_recent_memories }} {% endif %}\ {% if question %}\ Question: {{ question }} {% endif %} # Custom write memory template, which includes question and thought write_memory_template \ {% if question %}user: {{ question }} {% endif %} {% if thought %}assistant: {{ thought }} {% endif %}\ async def main(): # Create a profile with a custom user prompt template joy_profile ProfileConfig( nameJoy, roleComedians, user_prompt_templateuser_prompt_template, write_memory_templatewrite_memory_template, ) joy ( await ConversableAgent(profilejoy_profile) .bind(context) .bind(LLMConfig(llm_clientllm_client)) .bind(agent_memory) .bind(BlankAction) .build() ) user_proxy await UserProxyAgent().bind(agent_memory).bind(context).build() await user_proxy.initiate_chat( recipientjoy, revieweruser_proxy, messageMy name is bob, please tell me a joke, ) await user_proxy.initiate_chat( recipientjoy, revieweruser_proxy, messageWhats my name?, ) if __name__ __main__: asyncio.run(main())关键点说明AgentContext中的verboseTrue用于打印完整的对话历史便于观察记忆注入前后的 prompt 差异user_prompt_template自定义了用户侧 prompt其中most_recent_memories变量用于注入最近记忆write_memory_template自定义了写入记忆的内容格式将用户问题与助手回答组织为一条观察。上述示例的运行输出大致如下节选-------------------------------------------------------------------------------- User (to Joy)-[]: My name is bob, please tell me a joke -------------------------------------------------------------------------------- un_stream ai response: Sure thing, Bob! Heres one for you: Why dont scientists trust atoms? Because they make up everything! -------------------------------------------------------------------------------- String Prompt[verbose]: system: You are a Comedians, named Joy, your goal is None. Please think step by step to achieve the goal. You can use the resources given below. At the same time, please strictly abide by the constraints and specifications in IMPORTANT REMINDER. *** IMPORTANT REMINDER *** Please answer in English. human: Question: My name is bob, please tell me a joke LLM Output[verbose]: Sure thing, Bob! Heres one for you: Why dont scientists trust atoms? Because they make up everything! -------------------------------------------------------------------------------- -------------------------------------------------------------------------------- Joy (to User)-[gpt-4o]: Sure thing, Bob! Heres one for you:\n\nWhy dont scientists trust atoms?\n\nBecause they make up everything! Joy Review info: Pass(None) Joy Action report: execution succeeded, Sure thing, Bob! Heres one for you: Why dont scientists trust atoms? Because they make up everything! -------------------------------------------------------------------------------- -------------------------------------------------------------------------------- User (to Joy)-[]: Whats my name? -------------------------------------------------------------------------------- un_stream ai response: Your name is Bob! And heres another quick joke for you: Why dont skeletons fight each other? They dont have the guts! -------------------------------------------------------------------------------- String Prompt[verbose]: system: You are a Comedians, named Joy, your goal is None. Please think step by step to achieve the goal. You can use the resources given below. At the same time, please strictly abide by the constraints and specifications in IMPORTANT REMINDER. *** IMPORTANT REMINDER *** Please answer in English. human: Most recent observations: user: My name is bob, please tell me a joke assistant: Sure thing, Bob! Heres one for you: Why dont scientists trust atoms? Because they make up everything! Question: Whats my name? LLM Output[verbose]: Your name is Bob! And heres another quick joke for you: Why dont skeletons fight each other? They dont have the guts! -------------------------------------------------------------------------------- -------------------------------------------------------------------------------- Joy (to User)-[gpt-4o]: Your name is Bob! \n\nAnd heres another quick joke for you:\n\nWhy dont skeletons fight each other?\n\nThey dont have the guts! Joy Review info: Pass(None) Joy Action report: execution succeeded, Your name is Bob! And heres another quick joke for you: Why dont skeletons fight each other? They dont have the guts! --------------------------------------------------------------------------------输出解读在第二轮对话中可以看到用户 prompt 中出现了Most recent observations:字段其内容正是第一轮的用户提问与助手回答。正是这段被注入的记忆让 Joy 在第二轮被问 Whats my name? 时能够正确答出 Your name is Bob!。这验证了记忆模块的实际效果Agent 通过短期记忆记住了多轮对话中的关键信息。从源码视角看这一流程发生在 base_agent.py 的消息回复处理逻辑中Agent 先调用read_memories(observation)读取记忆将其组装为most_recent_memories字符串再通过build_system_prompt/build_prompt注入系统与用户模板。most_recent_memories被声明为 profile 的标准渲染变量默认的用户模板中同样预留了该字段见 profile/base.py。4.2 写入记忆当 Agent 收到 LLM 的响应后会把查询和响应写入记忆。由于记忆片段中的content是字符串你需要自行决定如何将信息组织进 content。在上述示例中write_memory_template为write_memory_template \ {% if question %}user: {{ question }} {% endif %} {% if thought %}assistant: {{ thought }} {% endif %}\ 其中question是用户查询thought是 LLM 响应下一节我们将深入讲解其写入过程。4.3 默认读写实现的底层逻辑如果不做任何定制ConversableAgent的默认读写逻辑定义在 role.pyread_memories调用self.memory.read(question)读取记忆片段将所有片段的raw_observation拼接成一个字符串返回write_memories要求必须存在action_output否则抛出ValueError(Action output is required to save to memory.)随后取出action_output.thoughts为空时回退到ai_message连同action、action_input、observation失败原因或行动观察等字段组成memory_map交给 profile 的write_memory_template渲染成memory_content最后构造AgentMemoryFragment并调用self.memory.write(fragment)。五、自定义记忆读写继承 ConversableAgent你可以通过继承ConversableAgent并重写read_memories与write_memories方法完全定制 Agent 的记忆读写行为。from typing import Optional from dbgpt.agent import ( ConversableAgent, AgentMemoryFragment, ProfileConfig, BlankAction, ActionOutput, ) write_memory_template \ {% if question %}user: {{ question }} {% endif %} {% if thought %}assistant: {{ thought }} {% endif %}\ class JoyAgent(ConversableAgent): profile: ProfileConfig ProfileConfig( nameJoy, roleComedians, write_memory_templatewrite_memory_template, ) def __init__(self, **kwargs): super().__init__(**kwargs) self._init_actions([BlankAction]) async def read_memories( self, question: str, ) - str: Read the memories from the memory. memories await self.memory.read(observationquestion) recent_messages [m.raw_observation for m in memories] # Merge the recent messages. return .join(recent_messages) async def write_memories( self, question: str, ai_message: str, action_output: Optional[ActionOutput] None, check_pass: bool True, check_fail_reason: Optional[str] None, ) - None: Write the memories to the memory. We suggest you to override this method to save the conversation to memory according to your needs. Args: question(str): The question received. ai_message(str): The AI message, LLM output. action_output(ActionOutput): The action output. check_pass(bool): Whether the check pass. check_fail_reason(str): The check fail reason. if not action_output: raise ValueError(Action output is required to save to memory.) mem_thoughts action_output.thoughts or ai_message memory_map { question: question, thought: mem_thoughts, } # This is the template to write the memory. # It configured in the agents profile. write_memory_template self.write_memory_template memory_content: str self._render_template(write_memory_template, **memory_map) fragment AgentMemoryFragment(memory_content) await self.memory.write(fragment)要点解读重写read_memories自定义读取逻辑。在 DB-GPT 中最近读取到的记忆会构成 prompt 模板中的most_recent_memories变量。此处使用m.raw_observation直接取原始观察内容并拼接返回。重写write_memories自定义写入逻辑。write_memories的五个参数语义如下question收到的用户问题ai_messageLLM 输出的 AI 消息action_output行动输出默认实现要求必填否则无法落库check_pass审核是否通过check_fail_reason审核失败原因。写入时通过self.write_memory_template来自 profile 配置渲染memory_map最终构造AgentMemoryFragment并调用self.memory.write(fragment)完成落库。因此你可以根据自己的业务需求灵活定制记忆的读写方式——例如只在审核通过时写入、把记忆按结构化 JSON 存储、或对读取结果做重排过滤等。六、记忆模块的源码级工作原理6.1 短期记忆的增强与转存机制普通ShortTermMemory直接以内存列表保存片段缓冲区满时把最旧片段转给长期记忆见 short_term.py。而EnhancedShortTermMemory见 short_term.py引入了记忆增强策略每条新记忆先计算 embedding 向量与已有记忆做余弦相似度经 sigmoid 函数映射到 [0,1] 区间当相似度概率达到enhance_similarity_threshold默认0.7时对命中记忆的enhance_cnt计数加一当某条记忆被增强次数达到enhance_threshold默认3时将其与相关增强记忆合并reduce为一条高层记忆并尝试提取洞察insights后转存长期记忆缓冲区溢出时则按重要度 增强次数排序淘汰最不重要的片段。6.2 长期记忆的时间加权检索LongTermMemory见 long_term.py基于向量库与LongTermRetriever继承自TimeWeightedEmbeddingRetriever实现写入时为每个记忆片段计算重要度无重要度时使用_default_importance并将importance、last_accessed_at、session_id等写入元数据后存入向量库读取时通过时间加权打分(1.0 - decay_rate) ** hours_passed结合向量相似度与重要度召回最相关且较新的记忆被标记为[FORGET]或[MERGE]的占位内容会被检索器过滤以支持未来的遗忘与合并机制。6.3 记忆在 prompt 中的注入链路most_recent_memories的注入链路为read_memories→generate_reply_message组装most_recent_memories字符串→build_system_prompt/build_prompt→ProfileConfig.format_system_prompt/format_user_prompt渲染 Jinja2 模板。你可以在 profile/base.py 中看到该变量的完整渲染逻辑也可以在中 base_agent.py 中追踪记忆在回复生成中的注入时机。七、总结本文围绕 DB-GPT Agent 的 memory 模块完成了从概念到实战的完整梳理操作层面掌握记忆读取、记忆写入、记忆反思三大操作以及read_memories/write_memories的默认实现与自定义方式结构层面理解感觉记忆、短期记忆、长期记忆、混合记忆四层结构以及EnhancedShortTermMemory的增强转存、LongTermMemory的时间加权检索、HybridMemory的流水线读写机制工程层面能够通过AgentMemoryShortTermMemory/GptsMemory/HybridMemory组装记忆并通过most_recent_memories模板变量把记忆注入 prompt最终让 Agent 在多轮对话中保持上下文一致性与行为合理性。后续文档还将分别深入介绍每一种记忆结构的具体用法可继续阅读 short_term_memory.md、long_term_memory.md 与 hybrid_memory.md并结合 examples/agents 下的各类 Agent 示例如single_agent_dialogue_example.py、react_agent_example.py进行动手验证。【免费下载链接】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），仅供参考
DB-GPT Agent 记忆模块深入指南从记忆结构到自定义读写实战【免费下载链接】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 Agent 框架中的 memory 模块为主线系统讲解 Agent 的三大记忆操作读取、写入、反思与四层记忆结构感觉记忆、短期记忆、长期记忆、混合记忆并结合仓库源码剖析Memory、MemoryFragment、AgentMemory、GptsMemory等核心类的设计原理。读完本文你将掌握在 DB-GPT 中创建与管理 Agent 记忆、通过 prompt 模板注入most_recent_memories、以及继承ConversableAgent自定义记忆读写策略的完整实战方法。一、记忆模块在 Agent 架构中的定位在 DB-GPT 的 Agent 架构设计中记忆模块扮演着至关重要的角色它存储从环境中感知到的信息并利用这些记录去辅助未来的行动决策。记忆模块能够帮助 Agent 持续积累经验、实现自我进化并以更一致、更合理、更高效的方式行事。从仓库源码结构看记忆模块的完整实现位于packages/dbgpt-core/src/dbgpt/agent/core/memory/目录核心抽象定义在 base.py具体实现分散在short_term.py、long_term.py、hybrid.py、agent_memory.py与gpts/子目录中。1.1 记忆的三大操作Memory Operations在 DB-GPT Agents 中记忆操作分为三类记忆读取Memory Reading从记忆中提取有意义的信息用以增强 Agent 的行动能力。记忆写入Memory Writing将感知到的环境信息存入记忆。有价值的信息被存储下来为未来检索有信息量的记忆提供基础使 Agent 能更高效、更理性地行动。记忆反思Memory Reflection模拟人类对自身认知、情感和行为过程的审视与评估能力。迁移到 Agent 上其目标是让 Agent 具备独立总结并推断更抽象、更复杂、更高层信息的能力。源码层面Memory抽象基类为这三个操作定义了统一接口read(observation, alpha, beta, gamma)、write(memory_fragment, now, op)与reflect(memory_fragments)见 base.py。1.2 记忆的四层结构Memory StructureDB-GPT Agents 中定义了四种主要记忆结构设计上对标人类记忆的渐进模型源码注释中对这一设计有明确说明见 base.py 文件头感觉记忆Sensory Memory如同人类感觉记忆用于登记感知输入接收来自环境的观察其中部分会被转移到短期记忆。其importance_weight默认为0.9threshold_to_short_term默认为0.1。短期记忆Short-term Memory临时缓冲最近的感知接收来自感觉记忆的部分内容并可通过其他观察或检索到的记忆得到增强进而进入长期记忆。长期记忆Long-term Memory存储 Agent 的经验与知识接收短期记忆的信息并随时间整合重要信息。混合记忆Hybrid Memory感觉记忆、短期记忆与长期记忆的组合体显式建模人类短时与长时记忆机制长期记忆负责随时间巩固重要信息。二、记忆核心概念解析在进入编码之前先理清 DB-GPT Agent 记忆体系中几个关键概念它们全部可以在源码中找到对应实现Memory存储所有记忆的类目前可以是SensorMemory、ShortTermMemory、EnhancedShortTermMemory、LongTermMemory和HybridMemory。它是一个泛型抽象基类Memory(ABC, Generic[T])声明了read、write、write_batch、clear、structure_clone等抽象方法并内置了score_memory_importance重要度打分与get_insights洞察提取两个能力钩子见 base.py。MemoryFragment存储记忆信息的抽象类是记忆的基本单元包含观察内容observation、嵌入向量embeddings、记忆 id、重要度importance、是否洞察is_insight、最后访问时间last_accessed_time等基础信息。其id通常由雪花算法生成因此可以从 id 反解出记忆片段的创建时间戳见 base.py。AgentMemoryFragment继承自MemoryFragment的默认实现在 agent_memory.py 中定义用于承载 Agent 的对话与行动记忆。仓库中还提供了StructuredAgentMemoryFragment结构化版本将观察序列化为 JSON 存储供需要结构化记忆的场景使用。GptsMemory用于存储对话conversation与计划plan信息从记忆结构的角度看它不属于记忆体系而是作为会话层的信息载体见 gpts_memory.py。AgentMemory同时包含Memory与GptsMemory的门面类是 Agent 与底层记忆交互的统一入口。其构造逻辑是未显式传入memory时默认创建ShortTermMemory(buffer_size5)未传入gpts_memory时默认创建GptsMemory()见 agent_memory.py。AgentMemory上还暴露了两个便捷属性plans_memory与message_memory分别指向gpts_memory内部的计划记忆与消息记忆方便直接读写会话/计划信息。三、创建 Agent 记忆从最小示例到完整配置3.1 创建带默认短期记忆的 AgentMemory正如前文所述记忆都包含在AgentMemory类中。最简单的创建方式如下from dbgpt.agent import AgentMemory, ShortTermMemory # Create an agent memory, default memory is ShortTermMemory memory ShortTermMemory(buffer_size5) agent_memory AgentMemory(memorymemory)这里buffer_size5表示短期记忆缓冲区最多保留 5 条记忆片段。从 short_term.py 的ShortTermMemory实现可以看到写入时会先做去重合并handle_duplicated当缓冲区溢出时通过transfer_to_long_term把最旧的记忆片段转存到长期记忆从而让短期记忆始终只保留最近的观察。3.2 附带 GptsMemory 的完整 AgentMemory常规理解下GptsMemory不属于记忆结构它专门用于存放对话与计划信息。你可以在AgentMemory中传入一个GptsMemoryfrom dbgpt.agent import AgentMemory, ShortTermMemory, GptsMemory # Create an agent memory, default memory is ShortTermMemory memory ShortTermMemory(buffer_size5) # Store the conversation and plan information gpts_memory GptsMemory() agent_memory AgentMemory(memorymemory, gpts_memorygpts_memory)GptsMemory在 gpts_memory.py 中维护plans_memory默认DefaultGptsPlansMemory与message_memory默认DefaultGptsMessageMemory两块存储并负责会话消息的缓存、消息队列推送以及可视化消息VIS的组装是 Agent 多轮对话与多步计划得以持久化的底层支撑。3.3 进阶创建混合记忆当需要同时具备短期记忆的实时性与长期记忆的持久性时可以使用HybridMemory。它支持通过from_chroma工厂方法一键创建默认使用 OpenAI Embedding API 与 ChromaStore或通过from_vstore基于自定义向量库创建相关用法可参考 hybrid_memory.md 以及实现文件 hybrid.py。混合记忆的读写采用流水线式策略写入时先进入感觉记忆溢出后转存短期记忆短期记忆中被多次增强enhance的片段会合并为高层洞察写入长期记忆读取时则先从长期记忆经TimeWeightedEmbeddingRetriever按时间加权召回再回写短期记忆以增强后续推理。四、在 Agent 中读写记忆完整实战示例Agent 会调用read_memories方法从记忆中读取记忆片段并调用write_memories方法把记忆片段写入记忆。当 Agent 调用 LLM 时记忆会被写入 LLM promptLLM 返回响应后Agent 又会把查询和响应写回记忆。正如 Profile To Prompt 中介绍的prompt 模板中存在一个名为most_recent_memories的模板变量它会被替换为最近读取到的记忆内容。4.1 读取记忆并构建 Prompt下面是一个完整的读取记忆并构建 prompt 的示例来自官方文档可直接运行。该示例创建了一个名为 Joy 的喜剧演员 Agent通过两轮对话验证记忆注入的效果import os import asyncio from dbgpt.agent import ( AgentContext, ShortTermMemory, AgentMemory, ConversableAgent, ProfileConfig, LLMConfig, BlankAction, UserProxyAgent, ) from dbgpt.model.proxy import OpenAILLMClient llm_client OpenAILLMClient( model_aliasgpt-4o, api_baseos.getenv(OPENAI_API_BASE), api_keyos.getenv(OPENAI_API_KEY), ) context: AgentContext AgentContext( conv_idtest123, languageen, temperature0.9, max_new_tokens2048, verboseTrue, # Add verboseTrue to print out the conversation history ) # Create an agent memory, which contains a short-term memory memory ShortTermMemory(buffer_size2) agent_memory: AgentMemory AgentMemory(memorymemory) # Custom user prompt template, which includes most recent memories and question user_prompt_template \ {% if most_recent_memories %}\ Most recent observations: {{ most_recent_memories }} {% endif %}\ {% if question %}\ Question: {{ question }} {% endif %} # Custom write memory template, which includes question and thought write_memory_template \ {% if question %}user: {{ question }} {% endif %} {% if thought %}assistant: {{ thought }} {% endif %}\ async def main(): # Create a profile with a custom user prompt template joy_profile ProfileConfig( nameJoy, roleComedians, user_prompt_templateuser_prompt_template, write_memory_templatewrite_memory_template, ) joy ( await ConversableAgent(profilejoy_profile) .bind(context) .bind(LLMConfig(llm_clientllm_client)) .bind(agent_memory) .bind(BlankAction) .build() ) user_proxy await UserProxyAgent().bind(agent_memory).bind(context).build() await user_proxy.initiate_chat( recipientjoy, revieweruser_proxy, messageMy name is bob, please tell me a joke, ) await user_proxy.initiate_chat( recipientjoy, revieweruser_proxy, messageWhats my name?, ) if __name__ __main__: asyncio.run(main())关键点说明AgentContext中的verboseTrue用于打印完整的对话历史便于观察记忆注入前后的 prompt 差异user_prompt_template自定义了用户侧 prompt其中most_recent_memories变量用于注入最近记忆write_memory_template自定义了写入记忆的内容格式将用户问题与助手回答组织为一条观察。上述示例的运行输出大致如下节选-------------------------------------------------------------------------------- User (to Joy)-[]: My name is bob, please tell me a joke -------------------------------------------------------------------------------- un_stream ai response: Sure thing, Bob! Heres one for you: Why dont scientists trust atoms? Because they make up everything! -------------------------------------------------------------------------------- String Prompt[verbose]: system: You are a Comedians, named Joy, your goal is None. Please think step by step to achieve the goal. You can use the resources given below. At the same time, please strictly abide by the constraints and specifications in IMPORTANT REMINDER. *** IMPORTANT REMINDER *** Please answer in English. human: Question: My name is bob, please tell me a joke LLM Output[verbose]: Sure thing, Bob! Heres one for you: Why dont scientists trust atoms? Because they make up everything! -------------------------------------------------------------------------------- -------------------------------------------------------------------------------- Joy (to User)-[gpt-4o]: Sure thing, Bob! Heres one for you:\n\nWhy dont scientists trust atoms?\n\nBecause they make up everything! Joy Review info: Pass(None) Joy Action report: execution succeeded, Sure thing, Bob! Heres one for you: Why dont scientists trust atoms? Because they make up everything! -------------------------------------------------------------------------------- -------------------------------------------------------------------------------- User (to Joy)-[]: Whats my name? -------------------------------------------------------------------------------- un_stream ai response: Your name is Bob! And heres another quick joke for you: Why dont skeletons fight each other? They dont have the guts! -------------------------------------------------------------------------------- String Prompt[verbose]: system: You are a Comedians, named Joy, your goal is None. Please think step by step to achieve the goal. You can use the resources given below. At the same time, please strictly abide by the constraints and specifications in IMPORTANT REMINDER. *** IMPORTANT REMINDER *** Please answer in English. human: Most recent observations: user: My name is bob, please tell me a joke assistant: Sure thing, Bob! Heres one for you: Why dont scientists trust atoms? Because they make up everything! Question: Whats my name? LLM Output[verbose]: Your name is Bob! And heres another quick joke for you: Why dont skeletons fight each other? They dont have the guts! -------------------------------------------------------------------------------- -------------------------------------------------------------------------------- Joy (to User)-[gpt-4o]: Your name is Bob! \n\nAnd heres another quick joke for you:\n\nWhy dont skeletons fight each other?\n\nThey dont have the guts! Joy Review info: Pass(None) Joy Action report: execution succeeded, Your name is Bob! And heres another quick joke for you: Why dont skeletons fight each other? They dont have the guts! --------------------------------------------------------------------------------输出解读在第二轮对话中可以看到用户 prompt 中出现了Most recent observations:字段其内容正是第一轮的用户提问与助手回答。正是这段被注入的记忆让 Joy 在第二轮被问 Whats my name? 时能够正确答出 Your name is Bob!。这验证了记忆模块的实际效果Agent 通过短期记忆记住了多轮对话中的关键信息。从源码视角看这一流程发生在 base_agent.py 的消息回复处理逻辑中Agent 先调用read_memories(observation)读取记忆将其组装为most_recent_memories字符串再通过build_system_prompt/build_prompt注入系统与用户模板。most_recent_memories被声明为 profile 的标准渲染变量默认的用户模板中同样预留了该字段见 profile/base.py。4.2 写入记忆当 Agent 收到 LLM 的响应后会把查询和响应写入记忆。由于记忆片段中的content是字符串你需要自行决定如何将信息组织进 content。在上述示例中write_memory_template为write_memory_template \ {% if question %}user: {{ question }} {% endif %} {% if thought %}assistant: {{ thought }} {% endif %}\ 其中question是用户查询thought是 LLM 响应下一节我们将深入讲解其写入过程。4.3 默认读写实现的底层逻辑如果不做任何定制ConversableAgent的默认读写逻辑定义在 role.pyread_memories调用self.memory.read(question)读取记忆片段将所有片段的raw_observation拼接成一个字符串返回write_memories要求必须存在action_output否则抛出ValueError(Action output is required to save to memory.)随后取出action_output.thoughts为空时回退到ai_message连同action、action_input、observation失败原因或行动观察等字段组成memory_map交给 profile 的write_memory_template渲染成memory_content最后构造AgentMemoryFragment并调用self.memory.write(fragment)。五、自定义记忆读写继承 ConversableAgent你可以通过继承ConversableAgent并重写read_memories与write_memories方法完全定制 Agent 的记忆读写行为。from typing import Optional from dbgpt.agent import ( ConversableAgent, AgentMemoryFragment, ProfileConfig, BlankAction, ActionOutput, ) write_memory_template \ {% if question %}user: {{ question }} {% endif %} {% if thought %}assistant: {{ thought }} {% endif %}\ class JoyAgent(ConversableAgent): profile: ProfileConfig ProfileConfig( nameJoy, roleComedians, write_memory_templatewrite_memory_template, ) def __init__(self, **kwargs): super().__init__(**kwargs) self._init_actions([BlankAction]) async def read_memories( self, question: str, ) - str: Read the memories from the memory. memories await self.memory.read(observationquestion) recent_messages [m.raw_observation for m in memories] # Merge the recent messages. return .join(recent_messages) async def write_memories( self, question: str, ai_message: str, action_output: Optional[ActionOutput] None, check_pass: bool True, check_fail_reason: Optional[str] None, ) - None: Write the memories to the memory. We suggest you to override this method to save the conversation to memory according to your needs. Args: question(str): The question received. ai_message(str): The AI message, LLM output. action_output(ActionOutput): The action output. check_pass(bool): Whether the check pass. check_fail_reason(str): The check fail reason. if not action_output: raise ValueError(Action output is required to save to memory.) mem_thoughts action_output.thoughts or ai_message memory_map { question: question, thought: mem_thoughts, } # This is the template to write the memory. # It configured in the agents profile. write_memory_template self.write_memory_template memory_content: str self._render_template(write_memory_template, **memory_map) fragment AgentMemoryFragment(memory_content) await self.memory.write(fragment)要点解读重写read_memories自定义读取逻辑。在 DB-GPT 中最近读取到的记忆会构成 prompt 模板中的most_recent_memories变量。此处使用m.raw_observation直接取原始观察内容并拼接返回。重写write_memories自定义写入逻辑。write_memories的五个参数语义如下question收到的用户问题ai_messageLLM 输出的 AI 消息action_output行动输出默认实现要求必填否则无法落库check_pass审核是否通过check_fail_reason审核失败原因。写入时通过self.write_memory_template来自 profile 配置渲染memory_map最终构造AgentMemoryFragment并调用self.memory.write(fragment)完成落库。因此你可以根据自己的业务需求灵活定制记忆的读写方式——例如只在审核通过时写入、把记忆按结构化 JSON 存储、或对读取结果做重排过滤等。六、记忆模块的源码级工作原理6.1 短期记忆的增强与转存机制普通ShortTermMemory直接以内存列表保存片段缓冲区满时把最旧片段转给长期记忆见 short_term.py。而EnhancedShortTermMemory见 short_term.py引入了记忆增强策略每条新记忆先计算 embedding 向量与已有记忆做余弦相似度经 sigmoid 函数映射到 [0,1] 区间当相似度概率达到enhance_similarity_threshold默认0.7时对命中记忆的enhance_cnt计数加一当某条记忆被增强次数达到enhance_threshold默认3时将其与相关增强记忆合并reduce为一条高层记忆并尝试提取洞察insights后转存长期记忆缓冲区溢出时则按重要度 增强次数排序淘汰最不重要的片段。6.2 长期记忆的时间加权检索LongTermMemory见 long_term.py基于向量库与LongTermRetriever继承自TimeWeightedEmbeddingRetriever实现写入时为每个记忆片段计算重要度无重要度时使用_default_importance并将importance、last_accessed_at、session_id等写入元数据后存入向量库读取时通过时间加权打分(1.0 - decay_rate) ** hours_passed结合向量相似度与重要度召回最相关且较新的记忆被标记为[FORGET]或[MERGE]的占位内容会被检索器过滤以支持未来的遗忘与合并机制。6.3 记忆在 prompt 中的注入链路most_recent_memories的注入链路为read_memories→generate_reply_message组装most_recent_memories字符串→build_system_prompt/build_prompt→ProfileConfig.format_system_prompt/format_user_prompt渲染 Jinja2 模板。你可以在 profile/base.py 中看到该变量的完整渲染逻辑也可以在中 base_agent.py 中追踪记忆在回复生成中的注入时机。七、总结本文围绕 DB-GPT Agent 的 memory 模块完成了从概念到实战的完整梳理操作层面掌握记忆读取、记忆写入、记忆反思三大操作以及read_memories/write_memories的默认实现与自定义方式结构层面理解感觉记忆、短期记忆、长期记忆、混合记忆四层结构以及EnhancedShortTermMemory的增强转存、LongTermMemory的时间加权检索、HybridMemory的流水线读写机制工程层面能够通过AgentMemoryShortTermMemory/GptsMemory/HybridMemory组装记忆并通过most_recent_memories模板变量把记忆注入 prompt最终让 Agent 在多轮对话中保持上下文一致性与行为合理性。后续文档还将分别深入介绍每一种记忆结构的具体用法可继续阅读 short_term_memory.md、long_term_memory.md 与 hybrid_memory.md并结合 examples/agents 下的各类 Agent 示例如single_agent_dialogue_example.py、react_agent_example.py进行动手验证。【免费下载链接】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），仅供参考
