Chainlit集成Mem0使用一个拥有个性化AI记忆的网页聊天应用

前言

Mem0 简介，可以看我上一篇文章《解决LLM的永久记忆的解决方案-Mem0实现个性化AI永久记忆功能》。本篇文章是对Mem0 实战使用的一个示例。通过Chainlit 快速实现ui界面和open ai的接入，通过使用Mem0 实现对聊天者的对话记录的记忆。

设计实现基本原理：

用户每次发送消息时，会通过ai的调用，总结用户发送的消息，并将总结后的消息进行向量化的存储
在总结消息后到向量化存储的过程中，会先用总结后的消息，进行向量化检索，检查出最相关的前5条数据，并把用户最新总结的消息和从数据库查询出来最相关的5条数据交给ai，让AI根据相关度来判断是否选择调用数据库增加方法、修改方法或者删除方法。
用户记忆搜索的逻辑，就是从更具用户ID和用户问题，从向量数据库查询相关的记录（可以设置具体返回几条最相关）以提示词的方式交给 AI，作为AI的记忆。
Mem0 还支持neo4j图数据库,通过提取三元组的数据，和对三元组数据的查询作为AI的记忆，提示AI记忆检索的广度。

快速上手

创建一个文件，例如“chainlit_chat”

mkdir chainlit_chat

进入 chainlit_chat文件夹下，执行命令创建python 虚拟环境空间(需要提前安装好python sdk。 Chainlit 需要python>=3.8。,具体操作，由于文章长度问题就不在叙述，自行百度)，命令如下：

python -m venv .venv

这一步是避免python第三方库冲突，省事版可以跳过
.venv是创建的虚拟空间文件夹可以自定义

接下来激活你创建虚拟空间，命令如下：

#linux or mac
source .venv/bin/activate
#windows
.venv\Scripts\activate

在项目根目录下创建`requirements.txt`，内容如下：

chainlit
mem0

在项目根目录下创建`.env`环境变量，配置如下：

OPENAI_BASE_URL="open地址或者代理地址"
OPENAI_API_KEY="your-api-key"

OPENAI_BASE_URL 是 openai的api 的base地址，如果你能直接使用openai,这可以不设置，也可以设置成openai兼容的代理地址，注意代理服务要能兼容openai向量化接口
OPENAI_API_KEY 替换成你自己的API 密匙

执行以下命令安装依赖：

pip install -r .\requirements.txt

安装后，项目根目录下会多出.chainlit 和.files文件夹和chainlit.md文件

在项目根目录下创建app.py文件，代码如下：

import chainlit as clfrom personalai_tutor import PersonalAITutor@cl.on_chat_start
async def on_chat_start():ai_tutor = PersonalAITutor()cl.user_session.set("ai_tutor", ai_tutor)@cl.on_message
async def on_message(message: cl.Message):ai_tutor = cl.user_session.get("ai_tutor")msg = cl.Message(content="")response = await ai_tutor.ask(question=message.content, user_id="tarzan")async for part in response:if token := part.choices[0].delta.content or "":await msg.stream_token(token)await msg.update()

在项目根目录下创建personalai_tutor.py文件，代码如下：

import threading
from typing import Optional, Listfrom mem0 import Memory
from mem0.configs.prompts import MEMORY_ANSWER_PROMPT
from openai import AsyncOpenAIMEMORY_DEDUCTION_PROMPT = """
从所提供的文本中推断出事实、偏好和记忆。
只需将事实、偏好和记忆以要点形式返回即可:
自然语言文本:{user_input}
User/Agent详细信息:{metadata}推断:推断事实、偏好和记忆的限制:
-事实、偏好和记忆应该简明扼要。
-不要以“这个人喜欢披萨”开头。相反，从“喜欢披萨”开始。
-不记得提供的用户/代理详细信息。只记住事实、偏好和回忆。推断出的事实、偏好和记忆:
"""def _format_query_with_memories(messages, relevant_memories):memories_text = "\n".join(memory["memory"] for memory in relevant_memories)return f"- Relevant Memories/Facts: {memories_text}\n\n- User Question: {messages[-1]['content']}"def _prepare_messages(messages: List[dict]) -> List[dict]:if not messages or messages[0]["role"] != "system":return [{"role": "system", "content": MEMORY_ANSWER_PROMPT}] + messagesmessages[0]["content"] = MEMORY_ANSWER_PROMPTreturn messagesclass PersonalAITutor:def __init__(self):"""Initialize the PersonalAITutor with memory configuration and OpenAI client."""config = {'llm': {'config': {'openai_base_url': 'https://dashscope.aliyuncs.com/compatible-mode/v1','api_key': 'api_key','model': 'qwen-plus'}},'vector_store': {'provider': 'chroma','config': {'path': 'chroma',}},'embedder': {'config': {'openai_base_url': 'https://open.bigmodel.cn/api/paas/v4','api_key': 'api_key','model': 'embedding-3'}}}self.memory = Memory.from_config(config)self.client = AsyncOpenAI(base_url='https://dashscope.aliyuncs.com/compatible-mode/v1',api_key='api_key')def ask(self, question: Optional[str] = None, agent_id: Optional[str] = None, run_id: Optional[str] = None,metadata: Optional[dict] = None,filters: Optional[dict] = None,user_id=None, limit: Optional[int] = 10):"""Ask a question to the AI and store the relevant facts in memory:param run_id::param limit::param filters::param metadata::param agent_id::param question: The question to ask the AI.:param user_id: Optional user ID to associate with the memory."""if not any([user_id]):raise ValueError("One of user_id must be provided")messages = [{"role": "system", "content": "你是一个私人的AI助理。"},{"role": "user", "content": question}]prepared_messages = _prepare_messages(messages)if prepared_messages[-1]["role"] == "user":self._async_add_to_memory(question, user_id, agent_id, run_id, metadata, filters)relevant_memories = self._fetch_relevant_memories(question, user_id, agent_id, run_id, filters, limit)print('relevant_memories', relevant_memories)prepared_messages[-1]["content"] = _format_query_with_memories(messages, relevant_memories)print('prepared_messages', prepared_messages)stream = self.client.chat.completions.create(model="qwen-plus",stream=True,messages=prepared_messages)return streamdef get_user_memories(self, user_id=None):"""Retrieve all memories associated with the given user ID.:param user_id: Optional user ID to filter memories.:return: List of memories."""return self.memory.get_all(user_id=user_id)def get_knowledge(self, app_id=None):"""Retrieve all memories associated with the given user ID.:param app_id: Optional user ID to filter memories.:return: List of memories."""return self.memory.get_all(app_id=app_id)def _async_add_to_memory(self, question, user_id, agent_id, run_id, metadata, filters):# 如果用户发送的消息以英文为主，可以不传prompt，会使用自带的英文提示词prompt = MEMORY_DEDUCTION_PROMPT.format(user_input=question, metadata=metadata)def add_task():self.memory.add(data=question,user_id=user_id,agent_id=agent_id,run_id=run_id,metadata=metadata,filters=filters,prompt=prompt)threading.Thread(target=add_task, daemon=True).start()def _fetch_relevant_memories(self, question, user_id, agent_id, run_id, filters, limit):# TODO: 过总结过去的谈话来做得更好return self.memory.search(query=question,user_id=user_id,agent_id=agent_id,run_id=run_id,filters=filters,limit=limit,)

代码实现使用时国内通义千文和智谱清言的兼容open ai的接口
llm的接口是用的通义千文、向量化接口是用的智谱清言的接口
因为通义千文没有提供和openai向量化处理兼容的接口，原本想只用智谱清言接口实现的，当时实践发现，智谱清言在tool函数识别选择调用的准确率太低，于是换成了阿里的qwen-plus，几乎准确率100%。
Mem0默认先从env环境变量里取base_url和api_key的值，再从 config配置的openai_base_url和api_key的值。所以使用config文件设置openai_base_url和api_key时，不要在设置环境变量了。