管理你的代理记忆
代理促进了类人的推理,是迈向构建 AGI 和理解我们自身的一大进步。记忆是人类处理任务的关键组件,在构建 AI 代理时应同样重视。memary 模仿人类记忆来推进这些代理的能力。
快速入门 🏁
安装 memary
- 使用 pip:
确保你正在运行 Python 版本 <= 3.11.9,然后运行
pip install memary
- 本地安装:
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
步骤
-
【可选】如果使用 Ollama 本地运行模型,请遵循此仓库中的说明。
-
确保存在
.env
文件,并包含所有必要的凭证。.env
OPENAI_API_KEY="你的API密钥" NEO4J_PW="你的NEO4J密码" NEO4J_URL="你的NEO4J网址" PERPLEXITY_API_KEY="你的API密钥" GOOGLEMAPS_API_KEY="你的API密钥" ALPHA_VANTAGE_API_KEY="你的API密钥"
-
获取API凭证:
API 信息
- OpenAI 密钥
- Neo4j
- 点击 'Start for free` → 创建一个免费实例 → 打开自动下载的 txt 文件并使用凭证
- Perplexity 密钥
- Google Maps
- 密钥在 Google Cloud Console 的 'APIs & Services' 选项卡的 'Credentials' 页面生成
- Alpha Vantage
- 推荐使用 https://10minutemail.com/ 生成一个临时电子邮件进行注册
-
更新用户角色信息,可在
streamlit_app/data/user_persona.txt
中找到用户角色模板,可在streamlit_app/data/user_persona_template.txt
中找到说明 - 用相关信息替换大括号。 -
【可选】如有需要,更新系统角色信息,可在
streamlit_app/data/system_persona.txt
中找到。 -
运行:
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,
)
在初始化时传入 ['search', 'vision', 'locate', 'stocks']
的子集作为 include_from_defaults
可使用不同的默认工具集合。
添加自定义工具
def multiply(a: int, b: int) -> int:
"""乘以两个整数并返回结果"""
return a * b
chat_agent.add_tool({"multiply": multiply})
关于创建 LlamaIndex ReAct Agent 自定义工具的更多信息可以在这里找到。
移除自定义工具
chat_agent.remove_tool("multiply")
记忆仪表盘 🧠
(即将推出)
功能 | 益处 |
---|---|
🗣️ 代理记忆对话 | 访问某些记忆 |
🧠 分析代理进展 | 跟踪代理随时间发展其记忆的情况 |
⏮️ 回顾执行 | 回顾代理记忆以了解特定的响应 |
🧑🧑🧒🧒 受众偏好 | 了解受众的最佳和最新偏好 |
✍🏻 memaryParse | 将专有数据注入代理记忆并结合解析器进行高级数据摄取 |
🗂️ memaryRetrieval | 访问记忆并组合检索器以进行高级记忆检索 |
🧪 配置代理记忆 | 搜索和结合记忆数据库 |
🛝 游乐场 | 指定使用的模型和工具并对不同的记忆技术进行基准测试 |
🔍 保持最新 | 当代理记忆被添加、更新或移除时接收通知 |
核心概念 🧪
memary 的当前结构在下图中详细说明。
在撰写本文时,上述系统设计包括路由代理、知识图谱和记忆模块,所有这些都集成到 ChatAgent
类中,位于 src/agent
目录中。
这些组件的原始源代码也可以在其各自的目录中找到,包括基准测试、笔记本和更新。
原则
memary 信守尽可能少的开发者实施即可集成到你现有的代理中。我们通过坚持几个原则来实现这一点。
-
自动生成记忆
- 在初始化 memary 后,代理记忆会随着代理的互动自动更新。这种生成方式允许我们捕捉所有记忆,轻松显示在你的仪表盘中。此外,我们允许通过少量或无需编码来组合数据库!
-
记忆模块
- 鉴于当前的数据库状态,memary 跟踪用户的偏好并显示在仪表盘中供分析。
-
系统改进
- memary 模仿人类记忆随时间演变和学习。我们将在你的仪表盘中提供你代理改进的速度。
-
回顾记忆
- memary 负责跟踪所有对话,以便你可以回顾代理的执行并访问特定时间段的代理记忆(即将推出)。
代理
为了向没有现有代理的开发者提供 memary 集成,我们设置了一个简单的代理实现。我们使用 ReAct 代理进行规划并执行工具提供的查询。
虽然我们没有强调为代理装备许多工具,但搜索工具对于从知识图谱中检索信息至关重要。该工具基于现有节点查询知识图谱的响应,在不存在相关实体时执行外部搜索。其他默认代理工具包括 LLaVa 提供的计算机视觉和使用地理编码器和谷歌地图的定位工具。
注意:在未来版本更新中,当前用于演示目的的 ReAct 代理将从程序包中移除,以便memary 可以支持从任何提供者来的任何类型的代理。
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)
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
知识图谱
知识图谱 ↔ LLMs
- memary 使用 Neo4j 图数据库存储知识。
- 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())
)
记忆模块
记忆模块包括记忆流和实体知识存储。 记忆模块受Microsoft研究院提出的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]:
"""将记忆从记忆流转换到实体知识存储,通过分组实体
Returns:
knowledge_memory (list): 知识记忆项列表
"""
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:
# 上面的查询已将最终响应添加到KG,因此实体将出现在KG中
return response, self.get_entity(self.query_engine.retrieve(query))
return response
- 最相关的实体
def get_entity(self, retrieve) -> list[str]:
"""retrieve是QueryBundle对象的列表。
取回的QueryBundle对象有一个“node”属性,
该“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在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"上下文成功总结。总结:{response}")
logging.info(f"清除后总令牌数: {total_tokens*EVICTION_RATE}")
未来功能 🔜
与您的代理记忆对话 🗣️
memary的聊天界面提供了访问代理记忆的门户,整合了如搜索、删除、查看指定时期的代理记忆等功能,一应俱全,均可在您的仪表盘中访问。
分析代理进度 🧠
跟踪您的代理记忆的发展。我们将提供访问相关指标以表示代理记忆随时间增长的数据,这些数据将在您的仪表盘中提供。
跟踪记忆 ⏮️
memary 分解每个生成的响应的代理记忆。响应列表及其各自的记忆将在您的仪表盘中提供。人为输入(好/坏响应)可以帮助系统改进。
受众偏好 🧑🧑🧒🧒
通过我们的专有记忆模块,我们能够推断出在某些时间段内的受众偏好。受众的最佳和最新偏好会不断更新,并将在您的仪表盘中提供。
memaryParse ✍🏻
在插入到代理记忆之前解析和清理您的专有数据。memary 支持各种文件类型,包括表格和图像提取。组合不同的解析器以形成具有高级功能的父解析器。还可访问预定义数据库模式和节点关系集的模板或自定义定义您自己的解析器!这一切都可在您的仪表盘中实现。
memaryRetrieval 🗂️
使用不同的技术检索代理记忆。还可组合各种检索器以形成具有高级功能的父检索器。所有这些功能均可在您的仪表盘中实现。
可定制记忆 🧪
memary部署知识图以跟踪代理操作。查看、搜索和配置记忆以符合您的需求。将不同的记忆结合在一起以改进检索,并在不同图提供商之间切换。所有这些功能均可在您的仪表盘中实现。
操作场地 🛝
- 工具操作场地:简单定义Python函数并将其添加为您的代理工具之一。查看所有可用工具并在必要时移除任何工具。所有这些操作均可在您的仪表盘中实现!
- 模型操作场地:为memary中的任务选择特定模型以降低系统LLM成本。部署在HF上的所有模型将在您的仪表盘中提供。
- 基准测试操作场地:轻松运行不同的memary配置以评估哪些记忆选项更适合特定任务。
许可证
memary根据MIT许可证发布。