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 cl

from 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, List

from mem0 import Memory
from mem0.configs.prompts import MEMORY_ANSWER_PROMPT
from openai import AsyncOpenAI


MEMORY_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}] + messages
    messages[0]["content"] = MEMORY_ANSWER_PROMPT
    return messages


class 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 stream

    def 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时，不要在设置环境变量了。