管理您的智能体记忆

智能体能够促进类人推理，是迈向通用人工智能（AGI）以及理解人类自身的重要一步。记忆是人类处理任务的关键组成部分，在构建人工智能智能体时也应给予同等重视。memary 模拟人类记忆，以推动这些智能体的发展。

快速入门 🏁

安装 memary

1. 使用 pip：

请确保您使用的 Python 版本不超过 3.11.9，然后运行以下命令：

pip install memary

2. 本地安装：

i. 创建一个使用上述指定版本 Python 的虚拟环境。

ii. 安装 Python 依赖项：

pip install -r requirements.txt

指定所用模型

截至撰写本文时，memary 假设已安装本地模型，并且目前支持通过 Ollama 提供的所有模型：

• 使用 Ollama 在本地运行的 LLM（建议默认为 Llama 3 8B/40B）或 gpt-3.5-turbo
• 使用 Ollama 在本地运行的视觉模型（建议默认为 LLaVA）或 gpt-4-vision-preview

除非明确指定，否则 memary 将默认使用本地运行的模型。此外，memary 还允许开发者 轻松切换已下载的模型。

运行 memary

步骤

1. [可选] 如果您使用 Ollama 在本地运行模型，请按照此 仓库 中的说明操作。

2. 确保存在一个包含必要凭据的 .env 文件。

   .env

   OPENAI_API_KEY="YOUR_API_KEY"
   PERPLEXITY_API_KEY="YOUR_API_KEY"
   GOOGLEMAPS_API_KEY="YOUR_API_KEY"
   ALPHA_VANTAGE_API_KEY="YOUR_API_KEY"
   
   数据库使用（参见 API 信息）：
   FALKORDB_URL="falkor://[[username]:[password]]@[falkor_host_url]:port"
   或
   NEO4J_PW="YOUR_NEO4J_PW"
   NEO4J_URL="YOUR_NEO4J_URL"
   

3. 获取 API 凭据：

   API 信息

   • OpenAI 密钥
   • FalkorDB
     • 登录 → 点击“订阅” → 在仪表板上创建一个免费实例 → 使用凭据（用户名、密码、falkor_host_url 和端口）。
   • Neo4j
     • 点击“免费开始” → 创建一个免费实例 → 打开自动下载的文本文件并使用凭据。
   • Perplexity 密钥
   • Google 地图
     • 密钥在 Google Cloud 控制台的“API 和服务”选项卡中的“凭据”页面生成。
   • Alpha Vantage
     • 建议使用 https://10minutemail.com/ 生成临时邮箱使用。

4. 更新用户角色，可在 streamlit_app/data/user_persona.txt 中找到，使用位于 streamlit_app/data/user_persona_template.txt 中的用户角色模板。已提供说明——将大括号替换为相关信息。

5. [可选] 如有需要，更新系统角色，可在 streamlit_app/data/system_persona.txt 中找到。

6. [可选] 多图谱——使用 FalkorDB 的用户可以生成多个图谱，并在其 ID 之间切换，这些 ID 对应不同的智能体。这使得不同智能体的记忆和知识上下文能够无缝切换和管理。

7. 运行：

cd streamlit_app
streamlit run app.py

基本用法

from memary.agent.chat_agent import ChatAgent

system_persona_txt = "data/system_persona.txt"
user_persona_txt = "data/user_persona.txt"
past_chat_json = "data/past_chat.json"
memory_stream_json = "data/memory_stream.json"
entity_knowledge_store_json = "data/entity_knowledge_store.json"
chat_agent = ChatAgent(
    "个人智能体",
    memory_stream_json,
    entity_knowledge_store_json,
    system_persona_txt,
    user_persona_txt,
    past_chat_json,
)

在初始化时，可根据需要传入 ['search', 'vision', 'locate', 'stocks'] 子集作为 include_from_defaults，以使用不同的默认工具集。

多图谱

当使用 FalkorDB 数据库时，您可以创建多个智能体。以下是为不同用户设置个人智能体的示例：

# 用户 A 的个人智能体
chat_agent_user_a = ChatAgent(
    "个人智能体",
    memory_stream_json_user_a，
    entity_knowledge_store_json_user_a，
    system_persona_txt_user_a，
    user_persona_txt_user_a，
    past_chat_json_user_a，
    user_id='user_a_id'
)

# 用户 B 的个人智能体
chat_agent_user_b = ChatAgent(
    "个人智能体",
    memory_stream_json_user_b，
    entity_knowledge_store_json_user_b，
    system_persona_txt_user_b，
    user_persona_txt_user_b，
    past_chat_json_user_b，
    user_id='user_b_id'
)

添加自定义工具

def multiply(a: int, b: int) -> int:
    """将两个整数相乘并返回结果"""
    return a * b

chat_agent.add_tool({"multiply": multiply})

有关为 LlamaIndex ReAct Agent 创建自定义工具的更多信息，请参阅 此处。

移除自定义工具

chat_agent.remove_tool("multiply")

核心概念 🧪

memary 当前的结构如下面的示意图所示。

截至撰写本文时，上述系统设计中，路由智能体、知识图谱和记忆模块均已集成到位于 src/agent 目录下的 ChatAgent 类中。

这些组件的原始源代码也可在其各自的目录中找到，包括基准测试、笔记本和更新内容。

原则

memary 以尽可能少的开发人员实现工作，无缝集成到您现有的代理中。我们通过坚持几项原则来实现这一点。

• 自动生成的记忆

  • 初始化 memary 后，随着代理的交互，代理记忆会自动更新。这种自动生成方式使我们能够捕获所有记忆，并在您的仪表板上轻松显示。此外，我们还支持无需或仅需少量代码即可将数据库组合在一起！

• 记忆模块

  • 根据当前数据库的状态，memary 会跟踪用户的偏好，并在您的仪表板上显示这些信息以便分析。

• 系统改进

  • memary 模仿人类记忆随时间演变和学习的方式。我们将在仪表板上展示您的代理改进的速度。

• 回溯记忆

  • memary 会自动记录所有聊天内容，以便您可以回溯代理的执行过程，并访问特定时间点的代理记忆（即将推出）。

代理

为了使没有现有代理的开发者也能使用 memary，我们设置了一个简单的代理实现。我们使用 ReAct 代理，根据提供的工具来规划并执行查询。

虽然我们并未强调为代理配备大量工具，但搜索工具对于从知识图谱中检索信息至关重要。该工具会基于现有节点查询知识图谱以获取响应；如果不存在相关实体，则会执行外部搜索。其他默认代理工具包括由 LLaVa 提供支持的计算机视觉工具，以及使用 geocodder 和 Google 地图的位置工具。

注意：在未来的版本发布中，目前用于演示目的的 ReAct 代理将被从软件包中移除，以便 memary 能够支持来自任何提供商的任何类型的代理。

def external_query(self, query: str):
    messages_dict = [
        {"role": "system", "content": "务必精确简洁。"},
        {"role": "user", "content": query},
    ]
    messages = [ChatMessage(**msg) for msg in messages_dict]
    external_response = self.query_llm.chat(messages)

    return str(external_response)

def search(self, query: str) -> str:
    response = self.query_engine.query(query)

    if response.metadata is None:
        return self.external_query(query)
    else:
        return response

知识图谱

知识图谱 ↔ 大型语言模型

• memary 使用图数据库来存储知识。
• 我们使用 Llama Index 根据文档向图存储中添加节点。
• 对于外部查询，则使用 Perplexity（mistral-7b-instruct 模型）。

知识图谱用例

• 将最终的代理响应注入到现有的知识图谱中。
• memary 采用一种递归式检索方法来搜索知识图谱：首先确定查询中的关键实体，然后构建一个包含这些实体且最大深度为 2 的子图，最后利用该子图来构建上下文。
• 当查询中存在多个关键实体时，memary 会使用多跳推理，将多个子图合并成一个更大的子图进行搜索。
• 这些技术相比一次性搜索整个知识图谱，能够显著降低延迟。

def query(self, query: str) -> str:
        # 获取 React 代理的响应
        response = self.routing_agent.chat(query)
        self.routing_agent.reset()
        # 将响应写入文件，以便后续写回知识图谱
        with open("data/external_response.txt", "w") as f:
            print(response, file=f)
        # 写回知识图谱
        self.write_back()
        return response

def check_KG(self, query: str) -> bool:
        """检查查询是否存在于知识图谱中。

        Args:
            query (str): 需要在知识图谱中检查的查询

        Returns:
            bool: 如果查询存在于知识图谱中则返回 True，否则返回 False
        """
        response = self.query_engine.query(query)

        if response.metadata is None:
            return False
        return generate_string(
            list(list(response.metadata.values())[0]["kg_rel_map"].keys())
        )

内存模块

内存模块由记忆流和实体知识库组成。该模块的设计受到微软研究院提出的K-LaMP的启发。

内存流

内存流会捕获所有被插入到知识图谱中的实体及其相关时间戳。这一流反映了用户的知识广度，即用户接触过但未深入理解的概念。

• 时间线分析：绘制交互的时间线，突出显示高互动时刻或话题焦点的变化。这有助于理解用户兴趣随时间的演变。

def add_memory(self, entities):
        self.memory.extend([
            MemoryItem(str(entity),
                       datetime.now().replace(microsecond=0))
            for entity in entities
        ])

• 提取主题：在交互中寻找反复出现的主题或话题。这种主题分析可以帮助我们在用户明确表达之前就预测其兴趣或问题。

def get_memory(self) -> list[MemoryItem]:
        return self.memory

实体知识库

实体知识库会跟踪存储在内存流中每个实体的引用频率和最近一次引用时间。该知识库反映了用户的知识深度，即用户对某些概念比其他概念更为熟悉。

• 按相关性对实体进行排序：结合频率和近期性对实体进行排序。如果某个实体被频繁提及（计数高）且最近被引用，则说明该实体可能非常重要，用户对该概念也非常了解。

def _select_top_entities(self):
        entity_knowledge_store = self.message.llm_message['knowledge_entity_store']
        entities = [entity.to_dict() for entity in entity_knowledge_store]
        entity_counts = [entity['count'] for entity in entities]
        top_indexes = np.argsort(entity_counts)[:TOP_ENTITIES]
        return [entities[index] for index in top_indexes]

• 对实体进行分类：根据实体的性质或其出现的上下文（例如技术术语、个人兴趣）将实体分组。这种分类有助于快速获取针对用户查询的相关信息。

def _convert_memory_to_knowledge_memory(
            self, memory_stream: list) -> list[KnowledgeMemoryItem]:
        """将内存流中的记忆转换为实体知识库，通过分组实体实现

        返回：
            knowledge_memory (list): KnowledgeMemoryItem列表
        """
        knowledge_memory = []

        entities = set([item.entity for item in memory_stream])
        for entity in entities:
            memory_dates = [
                item.date for item in memory_stream if item.entity == entity
            ]
            knowledge_memory.append(
                KnowledgeMemoryItem(entity, len(memory_dates),
                                    max(memory_dates)))
        return knowledge_memory

• 突出显示随时间的变化：识别实体排名或分类随时间发生的显著变化。最常提及实体的变化可能表明用户兴趣或知识结构发生了转变。
• 关于内存模块的更多信息请参见这里。

新上下文窗口

注意：我们利用与用户相关的关键分类实体和主题，以便使智能体的回答更贴合用户的当前兴趣/偏好以及知识水平/专业程度。新的上下文窗口由以下内容组成：

• 智能体响应

def get_routing_agent_response(self, query, return_entity=False):
        """从ReAct获取响应"""
        response = ""
        if self.debug:
            # 将ReAct智能体的步骤写入单独文件，并修改格式以便在.txt文件中可读
            with open("data/routing_response.txt", "w") as f:
                orig_stdout = sys.stdout
                sys.stdout = f
                response = str(self.query(query))
                sys.stdout.flush()
                sys.stdout = orig_stdout
            text = ""
            with open("data/routing_response.txt", "r") as f:
                text = f.read()

            plain = ansi_strip(text)
            with open("data/routing_response.txt", "w") as f:
                f.write(plain)
        else:
            response = str(self.query(query))

        if return_entity:
            # 上述查询已将最终响应添加到知识图谱中，因此实体会存在于知识图谱中
            return response, self.get_entity(self.query_engine.retrieve(query))
        return response

• 最相关实体

def get_entity(self, retrieve) -> list[str]:
        """retrieve 是一个 QueryBundle 对象列表。
        一个检索到的 QueryBundle 对象具有“node”属性，
        而该属性又包含“metadata”属性。

        例如对于“kg_rel_map”：
        kg_rel_map = {
            'Harry': [['DREAMED_OF', '未知关系'], ['FELL_HARD_ON', '水泥地面']],
            'Potter': [['WORE', '圆形眼镜'], ['HAD', '梦境']]
        }

        参数：
            retrieve (list[NodeWithScore]): NodeWithScore 对象列表
        返回：
            list[str]: 字符串实体列表
        """

        entities = []
        kg_rel_map = retrieve[0].node.metadata["kg_rel_map"]
        for key, items in kg_rel_map.items():
            # key 是问题中的实体
            entities.append(key)
            # items 是 [关系, 实体] 的列表
            entities.extend(item[1] for item in items)
            if len(entities) > MAX_ENTITIES_FROM_KG:
                break
        entities = list(set(entities))
        for exceptions in ENTITY_EXCEPTIONS:
            if exceptions in entities:
                entities.remove(exceptions)
        return entities

• 聊天历史（已摘要以避免令牌溢出）

def _summarize_contexts(self, total_tokens: int):
        """总结上下文。

        参数：
            total_tokens (int): 响应中的总令牌数
        """
        messages = self.message.llm_message["messages"]

        # 前两条消息分别是系统和用户角色
        if len(messages) > 2 + NONEVICTION_LENGTH:
            messages = messages[2:-NONEVICTION_LENGTH]
            del self.message.llm_message["messages"][2:-NONEVICTION_LENGTH]
        else:
            messages = messages[2:]
            del self.message.llm_message["messages"][2:]

        message_contents = [message.to_dict()["content"] for message in messages]

        llm_message_chatgpt = {
            "model": self.model,
            "messages": [
                {
                    "role": "user",
                    "content": "请将之前的对话总结成50字左右："
                    + str(message_contents),
                }
            ],
        }
        response, _ = self._get_gpt_response(llm_message_chatgpt)
        content = "总结的过往对话：" + response
        self._add_contexts_to_llm_message("assistant", content，index=2)
        logging.info(f"上下文已成功总结。 \n 总结内容：{response}")
        logging.info(f"驱逐后总令牌数：{total_tokens*EVICTION_RATE}")

许可证

memary 采用 MIT 许可证发布。

Memary 快速上手指南

Memary 是一个旨在模拟人类记忆机制的 AI 代理框架，通过知识图谱和记忆流模块，帮助开发者构建具备长期记忆和自我进化能力的智能代理。

环境准备

在开始之前，请确保满足以下系统要求和前置依赖：

• Python 版本：必须使用 Python <= 3.11.9（更高版本可能导致兼容性问题）。
• 本地模型服务（可选但推荐）：
  • 若使用本地模型，需安装并运行 Ollama。
  • 推荐拉取的模型：
    • LLM: Llama 3 8B 或 40B
    • 视觉模型：LLaVA
• API 密钥：根据需求准备以下服务的 API Key（需在 .env 文件中配置）：
  • OpenAI (必需，若不使用纯本地模型)
  • FalkorDB 或 Neo4j (用于知识图谱存储)
  • Perplexity, Google Maps, Alpha Vantage (可选，用于增强工具能力)

安装步骤

方法一：通过 PyPI 安装（推荐）

确保 Python 版本符合要求后，直接运行：

pip install memary

提示：国内用户若下载缓慢，可使用清华源加速：

pip install memary -i https://pypi.tuna.tsinghua.edu.cn/simple

方法二：本地源码安装

1. 创建并激活虚拟环境（确保 Python 版本 <= 3.11.9）：

   python -m venv venv
   # Windows
   venv\Scripts\activate
   # Mac/Linux
   source venv/bin/activate
   

2. 克隆项目并安装依赖：

   git clone https://github.com/kingjulio8238/memary.git
   cd memary
   pip install -r requirements.txt
   

基本使用

1. 配置文件设置

在项目根目录下创建 .env 文件，填入必要的凭证。示例如下：

OPENAI_API_KEY="YOUR_API_KEY"
PERPLEXITY_API_KEY="YOUR_API_KEY"
GOOGLEMAPS_API_KEY="YOUR_API_KEY"
ALPHA_VANTAGE_API_KEY="YOUR_API_KEY"

# 数据库配置 (二选一)
# 方案 A: FalkorDB
FALKORDB_URL="falkor://[username]:[password]@[host]:[port]"

# 方案 B: Neo4j
NEO4J_PW="YOUR_NEO4J_PW"
NEO4J_URL="YOUR_NEO4J_URL"

2. 设置用户人设

编辑 streamlit_app/data/user_persona.txt 文件。你可以参考同目录下的 user_persona_template.txt 模板，将大括号 {} 中的内容替换为实际的用户信息。如有需要，也可修改 system_persona.txt 来调整系统行为。

3. 启动演示应用

完成配置后，运行 Streamlit 应用以体验完整功能：

cd streamlit_app
streamlit run app.py

4. 代码集成（Python SDK）

你可以在自己的 Python 项目中直接调用 Memary 的核心代理类：

from memary.agent.chat_agent import ChatAgent

# 定义文件路径
system_persona_txt = "data/system_persona.txt"
user_persona_txt = "data/user_persona.txt"
past_chat_json = "data/past_chat.json"
memory_stream_json = "data/memory_stream.json"
entity_knowledge_store_json = "data/entity_knowledge_store.json"

# 初始化代理
chat_agent = ChatAgent(
    "Personal Agent",
    memory_stream_json,
    entity_knowledge_store_json,
    system_persona_txt,
    user_persona_txt,
    past_chat_json,
)

# (可选) 指定初始化工具子集，如搜索、视觉、定位等
# chat_agent = ChatAgent(..., include_from_defaults=['search', 'vision'])

高级功能示例

多用户/多图谱支持 (配合 FalkorDB):

# 用户 A 的代理
chat_agent_user_a = ChatAgent(
    "Personal Agent",
    memory_stream_json_user_a,
    entity_knowledge_store_json_user_a,
    system_persona_txt_user_a,
    user_persona_txt_user_a,
    past_chat_json_user_a,
    user_id='user_a_id'
)

添加自定义工具:

def multiply(a: int, b: int) -> int:
    """Multiply two integers and returns the result integer"""
    return a * b

chat_agent.add_tool({"multiply": multiply})

移除自定义工具:

chat_agent.remove_tool("multiply")