使用牛顿热冷却公式模拟记忆

基于 Chroma 向量数据库的内存管理辅助工具

特点和优势：
- 结合了向量搜索和牛顿冷却模型，可以模拟记忆衰退。
- 支持基于有效计数的过滤和排序。
- 提供了灵活的搜索和更新方法。
- 实现了 "遗忘" 机制，可以删除低于阈值的记录。
潜在的应用场景：
- 智能对话系统的长期记忆管理。
- 基于时间衰减的推荐系统。
- 知识库管理系统，可以自动淘汰不常用的信息。
可能的改进点：
- 添加更多的错误处理和输入验证。
- 考虑添加批量操作方法，以提高大规模操作的效率。
- 可以添加更多的统计和分析方法，如获取集合的统计信息。
- 考虑添加缓存机制，以提高频繁访问的数据的性能。

具体的代码实现

创建一个具体的实现，使用牛顿热冷却公式模拟记忆的过程。

add_memory: 添加新的记忆。
search_memories: 搜索相似的记忆。
update_memory_score: 更新记忆的分数。
forget_old_memories: 根据时间和分数遗忘旧的记忆。
get_all_memories: 获取所有存储的记忆。

下面是一个简单的 Python 实现：

import chromadb
from datetime import datetime
import math

class SimpleMemoryHelper:
    def __init__(self, collection_name: str):
        self.client = chromadb.Client()
        self.collection = self.client.get_or_create_collection(collection_name)

    def add_memory(self, id: str, content: str, embedding: list):
        self.collection.add(
            ids=[id],
            documents=[content],
            embeddings=[embedding],
            metadatas=[{
                "timestamp": datetime.now().isoformat(),
                "score": 1.0
            }]
        )

    def search_memories(self, query_embedding: list, n_results: int = 5):
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=n_results
        )
        return results

    def update_memory_score(self, id: str, new_score: float):
        self.collection.update(
            ids=[id],
            metadatas=[{"score": new_score}]
        )

    def forget_old_memories(self, threshold: float = 0.5):
        all_ids = self.collection.get()["ids"]
        for id in all_ids:
            metadata = self.collection.get(ids=[id])["metadatas"][0]
            timestamp = datetime.fromisoformat(metadata["timestamp"])
            score = metadata["score"]
            
            time_passed = (datetime.now() - timestamp).total_seconds()
            new_score = score * math.exp(-0.1 * time_passed / (24 * 3600))  # 简化的冷却公式
            
            if new_score < threshold:
                self.collection.delete(ids=[id])
            else:
                self.update_memory_score(id, new_score)

    def get_all_memories(self):
        return self.collection.get()

# 使用示例
memory_helper = SimpleMemoryHelper("my_memories")

# 添加记忆
memory_helper.add_memory("1", "This is a memory", [0.1, 0.2, 0.3])

# 搜索记忆
results = memory_helper.search_memories([0.1, 0.2, 0.3])
print(results)

# 更新记忆分数
memory_helper.update_memory_score("1", 0.8)

# 遗忘旧记忆
memory_helper.forget_old_memories()

# 获取所有记忆
all_memories = memory_helper.get_all_memories()
print(all_memories)

简单的改进

针对上面的评分机制进行更新，上面的评分是需要依赖于外部去调用 update_memory_score 方法，这样得依赖外部的评分算法，目前看了只能通过 LLM 去实现，但是这样的话就会增加很多的复杂度，所以可以考虑根据用户的反馈去更新评分，这样就可以减少很多的复杂度，同时也可以提高用户的体验。

我们可以设计一个基于用户反馈的评分机制，并将其整合到我们的记忆系统中。以下是一个改进的实现方案：

import chromadb
from datetime import datetime
import math

class UserFeedbackMemoryHelper:
    def __init__(self, collection_name: str):
        self.client = chromadb.Client()
        self.collection = self.client.get_or_create_collection(collection_name)
        self.base_score = 1.0
        self.cooling_rate = 0.1
        self.positive_feedback_boost = 0.2
        self.max_score = 2.0

    def add_memory(self, id: str, content: str, embedding: list):
        self.collection.add(
            ids=[id],
            documents=[content],
            embeddings=[embedding],
            metadatas=[{
                "timestamp": datetime.now().isoformat(),
                "score": self.base_score,
                "feedback_count": 0
            }]
        )

    def search_memories(self, query_embedding: list, n_results: int = 5):
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=n_results
        )
        return results

    def update_memory_score(self, id: str):
        metadata = self.collection.get(ids=[id])["metadatas"][0]
        timestamp = datetime.fromisoformat(metadata["timestamp"])
        current_score = metadata["score"]
        feedback_count = metadata["feedback_count"]

        # 应用冷却效果
        time_passed = (datetime.now() - timestamp).total_seconds()
        cooled_score = current_score * math.exp(-self.cooling_rate * time_passed / (24 * 3600))

        # 应用正面反馈boost
        new_score = min(cooled_score + self.positive_feedback_boost, self.max_score)

        self.collection.update(
            ids=[id],
            metadatas=[{
                "score": new_score,
                "feedback_count": feedback_count + 1,
                "timestamp": datetime.now().isoformat()
            }]
        )

    def provide_feedback(self, id: str, is_positive: bool):
        if is_positive:
            self.update_memory_score(id)
        # 如果是负面反馈，我们可以选择不做任何事，或者稍微降低分数

    def forget_old_memories(self, threshold: float = 0.5):
        all_ids = self.collection.get()["ids"]
        for id in all_ids:
            metadata = self.collection.get(ids=[id])["metadatas"][0]
            timestamp = datetime.fromisoformat(metadata["timestamp"])
            score = metadata["score"]
            
            time_passed = (datetime.now() - timestamp).total_seconds()
            new_score = score * math.exp(-self.cooling_rate * time_passed / (24 * 3600))
            
            if new_score < threshold:
                self.collection.delete(ids=[id])
            else:
                self.collection.update(
                    ids=[id],
                    metadatas=[{"score": new_score, "timestamp": datetime.now().isoformat()}]
                )

    def get_all_memories(self):
        return self.collection.get()

# 使用示例
memory_helper = UserFeedbackMemoryHelper("my_memories")

# 添加记忆
memory_helper.add_memory("1", "This is a memory", [0.1, 0.2, 0.3])

# 搜索记忆
results = memory_helper.search_memories([0.1, 0.2, 0.3])
print(results)

# 提供正面反馈
memory_helper.provide_feedback("1", True)

# 遗忘旧记忆
memory_helper.forget_old_memories()

# 获取所有记忆
all_memories = memory_helper.get_all_memories()
print(all_memories)

这个改进版本的 UserFeedbackMemoryHelper 类添加了以下新特性：

provide_feedback 方法：允许用户为特定的记忆提供反馈。如果反馈是正面的，它会调用 update_memory_score 方法来提高记忆的分数。
改进的 update_memory_score 方法：
- 应用冷却效果，基于时间衰减分数。
- 增加正面反馈的 boost。
- 限制最大分数，防止某些记忆的分数无限增长。
- 更新反馈计数和时间戳。
新的元数据字段：
- feedback_count：记录收到的反馈次数。
- timestamp：每次更新时都会刷新，用于更准确地计算冷却效果。
可调整的参数：
- base_score：新记忆的初始分数。
- cooling_rate：控制分数随时间衰减的速率。
- positive_feedback_boost：正面反馈时分数的增加量。
- max_score：分数的上限。

这个系统现在结合了时间衰减和用户反馈两个因素来评估记忆的重要性。频繁获得正面反馈的记忆会保持较高的分数，而长期未被使用或反馈的记忆会逐渐"遗忘"。

您可以根据具体需求进一步调整这个系统，例如：

添加负面反馈处理逻辑。
根据反馈的频率动态调整 boost 的大小。
实现更复杂的衰减函数或评分算法。

这种方法既保留了原有的时间衰减机制，又引入了用户反馈作为重要的评分因素，能够更好地反映记忆的实际价值和相关性。

具体的代码实现​

简单的改进​

具体的代码实现

简单的改进