stage

2025-11-20 13:51:53 +08:00
parent 8488c9aeab
commit b984bb2513
4 changed files with 202 additions and 0 deletions
@@ -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. 根据相似度，执行以下操作：
+   - 如果相似度超过高阈值，合并记忆内容
+   - 如果相似度在中等范围内
+   - 如果不是高似记忆，都按正常流程存储新记忆。
@@ -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