From b984bb25137f47a7f7bfc70a0885a0240660885b Mon Sep 17 00:00:00 2001 From: Soulter <905617992@qq.com> Date: Thu, 20 Nov 2025 13:51:53 +0800 Subject: [PATCH] stage --- astrbot/core/memory/_README.md | 94 +++++++++++++++++++++++ astrbot/core/memory/entities.py | 108 +++++++++++++++++++++++++++ astrbot/core/memory/mem_db_sqlite.py | 0 astrbot/core/memory/tools.py | 0 4 files changed, 202 insertions(+) create mode 100644 astrbot/core/memory/_README.md create mode 100644 astrbot/core/memory/entities.py create mode 100644 astrbot/core/memory/mem_db_sqlite.py create mode 100644 astrbot/core/memory/tools.py diff --git a/astrbot/core/memory/_README.md b/astrbot/core/memory/_README.md new file mode 100644 index 000000000..2be1d5411 --- /dev/null +++ b/astrbot/core/memory/_README.md @@ -0,0 +1,94 @@ +# AstrBot Long-Term Memory + +这个模块实现了 AstrBot 的长期记忆功能,基于 Function Calling 机制,允许模型调用预定义的 Tools 来管理记忆,并实现了简单的记忆遗忘机制。使用 AstrBot 向量数据库 API 存储和检索记忆片段。 + +## 一些概念 + +### 相关理论 + +1. 赫布理论:记忆通过神经元之间的连接(突触)进行存储和强化,频繁使用的连接会变得更强。 +2. 遗忘曲线:记忆会随着时间的推移而减弱,除非通过复习或使用来强化。 + +#### 建模方案一 + +为了建模上面两个理论,我们需要引入以下的变量: + +1. 记忆强度(Memory Strength):表示记忆的持久性和易检索性。记忆强度越高,记忆越不容易遗忘。 +2. 使用频率(Usage Frequency):表示记忆被访问或使用的频率。使用频率越高,记忆强度越高。 +3. 时间衰减(Time Decay):表示记忆强度随时间的自然衰减。时间越长,记忆强度越低。 +4. 强化因子(Reinforcement Factor):表示每次使用记忆时对记忆强度的提升效果。强化因子越大,记忆强度提升越显著。 + +写入记忆时,我们可以初始化记忆强度和使用频率: + +- 记忆强度(S)初始化为一个基准值 S0。 +- 使用频率(F)初始化为 1。 +读取记忆时,我们可以根据以下公式计算记忆强度: + +S = S0 * e^(-λt) + kF + +其中: + +- S0 是初始记忆强度。 +- λ 是时间衰减常数,表示记忆强度随时间的衰减速度。 +- t 是自记忆创建以来经过的时间。 +- k 是强化因子,表示每次使用记忆时对记忆强度的提升效果。 +- F 是使用频率,表示记忆被访问或使用的次数 + +当记忆被访问时,我们更新使用频率和记忆强度: + +- 使用频率(F)增加 1。 +- 记忆强度(S)根据上述公式重新计算。 + +相似记忆的合并: + +对相似记忆我们有两种处理模式: + +- 过于相似的记忆,我们会执行合并成新的记忆。 +- 较为相似的记忆,比如某些实体相同,根据赫布理论,我们会提升相似记忆的记忆强度和使用频率。 + +具体算法如下: + +1. 计算新记忆与现有记忆的相似度。 +2. 根据相似度,执行以下操作: + - 如果相似度超过高阈值,合并记忆内容,更新记忆强度和使用频率。 + - 如果相似度在中等范围内,提升现有记忆的记忆强度和使用频率。 + - 如果不是高似记忆,都按正常流程存储新记忆。 + +#### 建模方案二 + +我们参考艾宾浩斯遗忘曲线,基于这两个变量设计了一个公式,其表示了每个对话总结的遗忘得分。 + +每个记忆节点带有 + +1. last_retrieval_at +2. retrieval_count + +$decayscore = \alpha * exp(-\lambda * \delta_t * \beta) + (1-\alpha) * (1-exp(-\gamma * c))$ + +其中: + +- $\delta_t$: 自上次检索以来经过的时间(以天为单位)。 +- $c$: 检索次数。 +- $\alpha$: 控制时间衰减和检索次数影响的权重 +- $\gamma$: 控制检索次数影响的速率 +- $\lambda$: 控制时间衰减影响的速率 +- $\beta$: 时间衰减的调节因子 + +$\beta = \frac{1}{1 + a * c}$ + +- $a$: 控制检索次数对时间衰减影响的权重 + +相似记忆的合并: + +对相似记忆我们有两种处理模式: + +- 过于相似的记忆,我们会执行合并成新的记忆。 +- 较为相似的记忆,比如某些实体相同,根据赫布理论,我们会提升相似记忆的记忆强度和使用频率。 + +具体算法如下: + +1. 计算新记忆与现有记忆的相似度。 +2. 根据相似度,执行以下操作: + - 如果相似度超过高阈值,合并记忆内容 + - 如果相似度在中等范围内 + - 如果不是高似记忆,都按正常流程存储新记忆。 diff --git a/astrbot/core/memory/entities.py b/astrbot/core/memory/entities.py new file mode 100644 index 000000000..ac97e680a --- /dev/null +++ b/astrbot/core/memory/entities.py @@ -0,0 +1,108 @@ +from datetime import datetime + +from pydantic import BaseModel + +""" +我们参考艾宾浩斯遗忘曲线,基于这两个变量设计了一个公式,其表示了每个对话总结的遗忘得分。 + +$decayscore = alpha * exp(-lambda * delta_t * \beta) + (1-alpha) * (1-exp(-gamma * c))$ + +其中: + +- $delta_t$: 自上次检索以来经过的时间(以天为单位)。 +- $c$: 检索次数。 +- $alpha$: 控制时间衰减和检索次数影响的权重 +- $gamma$: 控制检索次数影响的速率 +- $lambda$: 控制时间衰减影响的速率 +- $beta$: 时间衰减的调节因子 + +$beta = frac{1}{1 + a * c}$ + +- $a$: 控制检索次数对时间衰减影响的权重 + +相似记忆的合并: + +对相似记忆我们有两种处理模式: + +- 过于相似的记忆,我们会执行合并成新的记忆。 +- 较为相似的记忆,比如某些实体相同,根据赫布理论,我们会提升相似记忆的记忆强度和使用频率。 + +具体算法如下: + +1. 计算新记忆与现有记忆的相似度。 +2. 根据相似度,执行以下操作: + - 如果相似度超过高阈值,合并记忆内容 + - 如果相似度在中等范围内 + - 如果不是高似记忆,都按正常流程存储新记忆。 +""" + + +class MemoryChunk(BaseModel): + """A chunk of memory stored in the system.""" + + id: str + fact: str + """The factual content of the memory chunk.""" + created_at: datetime + """The timestamp when the memory chunk was created.""" + last_retrieval_at: datetime + """The timestamp when the memory chunk was last retrieved.""" + retrieval_count: int + """The number of times the memory chunk has been retrieved.""" + importance_bias: float + """A bias score indicating the importance of the memory chunk.""" + + +# from astrbot.core.db.vec_db.faiss_impl import FaissVecDB +# from astrbot.core.provider.provider import EmbeddingProvider + +# memdb = None + + +# async def test_mem(embed_provider: EmbeddingProvider): +# global memdb +# mem_doc_path = "data/astr_memory/doc.db" +# mem_index_path = "data/astr_memory/index.faiss" +# memdb = FaissVecDB( +# doc_store_path=mem_doc_path, +# index_store_path=mem_index_path, +# embedding_provider=embed_provider, +# ) +# await memdb.initialize() + + +# @dataclass +# class AddMemory(FunctionTool[AstrAgentContext]): +# name: str = "astr_add_memory" +# description: str = ( +# "Add a new memory to the user's long-term memory storage. " +# "Use this tool only when the user explicitly asks you to remember something, " +# "or when they share stable preferences, identity, or long-term goals that will be useful in future interactions." +# ) +# parameters: dict = Field( +# default_factory=lambda: { +# "type": "object", +# "properties": { +# "query": { +# "type": "string", +# "description": "A concise keyword query for the knowledge base.", +# }, +# }, +# "required": ["query"], +# } +# ) + +# async def call( +# self, context: ContextWrapper[AstrAgentContext], **kwargs +# ) -> ToolExecResult: +# query = kwargs.get("query", "") +# if not query: +# return "error: Query parameter is empty." +# result = await retrieve_knowledge_base( +# query=kwargs.get("query", ""), +# umo=context.context.event.unified_msg_origin, +# context=context.context.context, +# ) +# if not result: +# return "No relevant knowledge found." +# return result diff --git a/astrbot/core/memory/mem_db_sqlite.py b/astrbot/core/memory/mem_db_sqlite.py new file mode 100644 index 000000000..e69de29bb diff --git a/astrbot/core/memory/tools.py b/astrbot/core/memory/tools.py new file mode 100644 index 000000000..e69de29bb