Memary — 为 LLM 应用提供长期记忆与上下文管理

管理您的智能体记忆

智能体促进了类人推理，是迈向构建通用人工智能（AGI）和理解人类自身的重大进步。记忆是人类处理任务的关键组成部分，在构建AI智能体时也应被赋予同等权重。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 文件。

3. 获取 API 凭证：
   
   

   API 信息

   - [**OpenAI 密钥**](https://openai.com/index/openai-api) - [**FalkorDB**](https://app.falkordb.cloud/) - 登录 → 点击 'Subscribe` → 在仪表板上创建一个免费实例 → 使用凭证（用户名、密码、falkor_host_url 和端口）。 - [**Neo4j**](https://neo4j.com/cloud/platform/aura-graph-database/?ref=nav-get-started-cta) - 点击 'Start for free` → 创建一个免费实例 → 打开自动下载的 txt 文件并使用其中的凭证。 - [**Perplexity 密钥**](https://www.perplexity.ai/settings/api) - [**Google Maps**](https://console.cloud.google.com/apis/credentials) - 密钥在 Google Cloud Console 的 "APIs & Services" 标签页的 "Credentials" 页面生成。 - [**Alpha Vantage**](https://www.alphavantage.co/support/#api-key) - 建议使用 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(
    "Personal Agent",
    memory_stream_json,
    entity_knowledge_store_json,
    system_persona_txt,
    user_persona_txt,
    past_chat_json,
)

初始化时，可以通过 include_from_defaults 参数传入 ['search', 'vision', 'locate', 'stocks'] 的子集，以选择不同的默认工具集。

多图（多智能体）

当使用 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'
)

# 用户 B 的个人智能体
chat_agent_user_b = ChatAgent(
    "Personal Agent",
    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 智能体创建自定义工具的更多信息，请参阅此处。

移除自定义工具

chat_agent.remove_tool("multiply")

核心概念 🧪

memary 的当前结构在下图中详细说明。

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

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

设计原则

memary 旨在以最少的开发者实现工作量集成到您现有的智能体上。我们通过遵循几个原则来实现这一点。

• 自动生成记忆

  • 初始化 memary 后，随着智能体的交互，其记忆会自动更新。这种生成方式使我们能够捕获所有记忆，以便轻松地在仪表板中展示。此外，我们允许以很少或无需代码的方式组合数据库！

• 记忆模块

  • 根据数据库的当前状态，memary 会跟踪用户的偏好，并在您的仪表板中显示以供分析。

• 系统改进

  • memary 模拟人类记忆如何随时间演变和学习。我们将在您的仪表板中提供智能体的改进速率。

• 回放记忆

  • memary 负责记录所有聊天记录，因此您可以回放智能体的执行过程，并访问特定时期的智能体记忆（即将推出）。

智能体

为了让没有现成智能体的开发者也能使用 memary，我们设置了一个简单的智能体实现。我们使用 ReAct 智能体来规划和执行给定工具的查询。

虽然我们没有强调为智能体配备大量工具，但搜索工具对于从知识图谱中检索信息至关重要。该工具基于现有节点查询知识图谱以获取响应，如果不存在相关实体，则执行外部搜索。其他默认的智能体工具包括由 LLaVA 驱动的计算机视觉和使用地理编码器和谷歌地图的位置工具。

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

``` py title="external_query" hl_lines="1"
def external_query(self, query: str):
messages_dict = [
{"role": "system", "content": "Be precise and concise."},
{"role": "user", "content": query},
]
messages = [ChatMessage(**msg) for msg in messages_dict]
external_response = self.query_llm.chat(messages)

return str(external_response)

``` py title="search" hl_lines="1"
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 使用多跳推理将多个子图连接成一个更大的子图进行搜索。
• 这些技术相比一次性搜索整个知识图谱，降低了延迟。

``` py title="store in KG" hl_lines="1"
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

``` py title="recursive retrieval" hl_lines="1"
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 的影响。

记忆流

记忆流捕获所有插入知识图谱的实体及其关联的时间戳。这个流反映了用户知识的广度，即用户接触过的概念，但不推断接触的深度。
- 时间线分析：绘制交互时间线，突出显示高参与度时刻或主题焦点的转变。这有助于理解用户兴趣随时间的演变。

``` py title="add to memory stream" hl_lines="1"
def add_memory(self, entities):
self.memory.extend([
MemoryItem(str(entity),
datetime.now().replace(microsecond=0))
for entity in entities
])

- **提取主题**：在交互中寻找反复出现的主题或话题。这种主题分析有助于在用户明确表达之前预测其兴趣或问题。

``` py title="retrieve from memory stream" hl_lines="1"
def get_memory(self) -> list[MemoryItem]:
        return self.memory

实体知识存储

实体知识存储跟踪记忆流中存储的每个实体的引用频率和最近引用时间。这个知识存储反映了用户知识的深度，即他们比其他概念更熟悉的概念。
- 按相关性对实体排序：同时使用频率和最近性对实体进行排序。一个被频繁提及（高计数）且最近被引用的实体很可能具有高重要性，并且用户对这个概念非常了解。

``` py title="select most relevant entities" hl_lines="1"
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]

- **对实体进行分类**：根据实体的性质或提及的上下文（例如，技术术语、个人兴趣）将实体分组。这种分类有助于快速访问与用户查询相关的信息。

``` py title="group entities" hl_lines="1"
def _convert_memory_to_knowledge_memory(
            self, memory_stream: list) -> list[KnowledgeMemoryItem]:
        """通过分组实体，将记忆流中的记忆转换为实体知识存储

        Returns:
            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

• 突出显示随时间的变化：识别实体排名或分类随时间的任何显著变化。最常提及实体的变化可能表明用户的兴趣或知识发生了变化。
• 关于记忆模块的更多信息可以在这里找到。

新的上下文窗口

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

• 智能体响应
  ``` py title="retrieve agent response" hl_lines="1"
  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
  

- **最相关的实体**
``` py title="retrieve important entities" hl_lines="1"
def get_entity(self, retrieve) -> list[str]:
        """retrieve 是一个 QueryBundle 对象列表。
        一个检索到的 QueryBundle 对象有一个 "node" 属性，
        该属性有一个 "metadata" 属性。

        "kg_rel_map" 示例：
        kg_rel_map = {
            'Harry': [['DREAMED_OF', 'Unknown relation'], ['FELL_HARD_ON', 'Concrete floor']],
            'Potter': [['WORE', 'Round glasses'], ['HAD', 'Dream']]
        }

        Args:
            retrieve (list[NodeWithScore]): NodeWithScore 对象列表
        return:
            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

• 聊天历史记录（为避免令牌溢出而进行了总结）
  ``` py title="summarize chat history" hl_lines="1"
  def _summarize_contexts(self, total_tokens: int):
  """总结上下文。

  Args:
      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": "Summarize these previous conversations into 50 words:"
              + str(message_contents),
          }
      ],
  }
  response, _ = self._get_gpt_response(llm_message_chatgpt)
  content = "Summarized past conversation:" + response
  self._add_contexts_to_llm_message("assistant", content, index=2)
  logging.info(f"Contexts summarized successfully. \n summary: {response}")
  logging.info(f"Total tokens after eviction: {total_tokens*EVICTION_RATE}")
  

  ```

许可证

memary 基于 MIT 许可证发布。