深入浅出分析最近火热的Mem0个性化AI记忆层

pip install mem0ai

import os
from mem0 import Memory

# 依赖LLM提取记忆，所以需要open ai
os.environ["OPENAI_API_KEY"] = "xxx"

# 吃石化 Mem0
m = Memory()

# 通过add方法，存储非结构化的记忆，metadata提供schema定义
result = m.add("I am working on improving my tennis skills. Suggest some online courses.", user_id="alice", metadata={"category": "hobbies"})
print(result)
# Created memory: Improving her tennis skills. Looking for online suggestions.

# Retrieve memories
all_memories = m.get_all()
print(all_memories)

# 搜索记忆  Search memories
related_memories = m.search(query="What are Alice's hobbies?", user_id="alice")
print(related_memories)

# 更新记忆 Update a memory
result = m.update(memory_id="m1", data="Likes to play tennis on weekends")
print(result)

# Get memory history
history = m.history(memory_id="m1")
print(history)

def add(
        self,
        data,
        user_id=None,
        agent_id=None,
        run_id=None,
        metadata=None,
        filters=None,
        prompt=None,
    ):
        """
        Create a new memory.

        Args:
            data (str): Data to store in the memory.
            user_id (str, optional): ID of the user creating the memory. Defaults to None.
            agent_id (str, optional): ID of the agent creating the memory. Defaults to None.
            run_id (str, optional): ID of the run creating the memory. Defaults to None.
            metadata (dict, optional): Metadata to store with the memory. Defaults to None.
            filters (dict, optional): Filters to apply to the search. Defaults to None.

        Returns:
            str: ID of the created memory.
        """
        if metadata is None:
            metadata = {}
        embeddings = self.embedding_model.embed(data)

        filters = filters or {}
        if user_id:
            filters["user_id"] = metadata["user_id"] = user_id
        if agent_id:
            filters["agent_id"] = metadata["agent_id"] = agent_id
        if run_id:
            filters["run_id"] = metadata["run_id"] = run_id

        if not prompt:
            prompt = MEMORY_DEDUCTION_PROMPT.format(user_input=data, metadata=metadata)
        extracted_memories = self.llm.generate_response(
            messages=[
                {
                    "role": "system",
                    "content": "You are an expert at deducing facts, preferences and memories from unstructured text.",
                },
                {"role": "user", "content": prompt},
            ]
        )
        existing_memories = self.vector_store.search(
            name=self.collection_name,
            query=embeddings,
            limit=5,
            filters=filters,
        )
        existing_memories = [
            MemoryItem(
                id=mem.id,
                score=mem.score,
                metadata=mem.payload,
                text=mem.payload["data"],
            )
            for mem in existing_memories
        ]
        serialized_existing_memories = [
            item.model_dump(include={"id", "text", "score"})
            for item in existing_memories
        ]
        logging.info(f"Total existing memories: {len(existing_memories)}")
        messages = get_update_memory_messages(
            serialized_existing_memories, extracted_memories
        )
        # Add tools for noop, add, update, delete memory.
        tools = [ADD_MEMORY_TOOL, UPDATE_MEMORY_TOOL, DELETE_MEMORY_TOOL]
        response = self.llm.generate_response(messages=messages, tools=tools)
        tool_calls = response["tool_calls"]

        response = []
        if tool_calls:
            # Create a new memory
            available_functions = {
                "add_memory": self._create_memory_tool,
                "update_memory": self._update_memory_tool,
                "delete_memory": self._delete_memory_tool,
            }
            for tool_call in tool_calls:
                function_name = tool_call["name"]
                function_to_call = available_functions[function_name]
                function_args = tool_call["arguments"]
                logging.info(
                    f"[openai_func] func: {function_name}, args: {function_args}"
                )

                # Pass metadata to the function if it requires it
                if function_name in ["add_memory", "update_memory"]:
                    function_args["metadata"] = metadata

                function_result = function_to_call(**function_args)
                # Fetch the memory_id from the response
                response.append(
                    {
                        "id": function_result,
                        "event": function_name.replace("_memory", ""),
                        "data": function_args.get("data"),
                    }
                )
                capture_event(
                    "mem0.add.function_call",
                    self,
                    {"memory_id": function_result, "function_name": function_name},
                )
        capture_event("mem0.add", self)
        return response

MEMORY_DEDUCTION_PROMPT = """
Deduce the facts, preferences, and memories from the provided text.
Just return the facts, preferences, and memories in bullet points:
Natural language text: {user_input}
User/Agent details: {metadata}

Constraint for deducing facts, preferences, and memories:
- The facts, preferences, and memories should be concise and informative.
- Don't start by "The person likes Pizza". Instead, start with "Likes Pizza".
- Don't remember the user/agent details provided. Only remember the facts, preferences, and memories.

Deduced facts, preferences, and memories:


从提供的文本中推断出事实、偏好和记忆。
仅以项目符号形式返回事实、偏好和记忆：
自然语言文本：{用户输入}
用户/代理详细信息：{元数据}

推断事实、偏好和记忆的约束：
- 事实、偏好和记忆应简洁且信息丰富。
- 不要以“此人喜欢披萨”开头。而是以“喜欢披萨”开头。
- 不要记住提供的用户/代理详细信息。只记住事实、偏好和记忆。

推断出的事实、偏好和记忆

You are an expert at merging, updating, and organizing memories. When provided with existing memories and new information, your task is to merge and update the memory list to reflect the most accurate and current information. You are also provided with the matching score for each existing memory to the new information. Make sure to leverage this information to make informed decisions about which memories to update or merge.

Guidelines:
- Eliminate duplicate memories and merge related memories to ensure a concise and updated list.
- If a memory is directly contradicted by new information, critically evaluate both pieces of information:
    - If the new memory provides a more recent or accurate update, replace the old memory with new one.
    - If the new memory seems inaccurate or less detailed, retain the original and discard the old one.
- Maintain a consistent and clear style throughout all memories, ensuring each entry is concise yet informative.
- If the new memory is a variation or extension of an existing memory, update the existing memory to reflect the new information.

Here are the details of the task:
- Existing Memories:
{existing_memories}

- New Memory: {memory}


您是合并、更新和组织记忆的专家。当您获得现有记忆和新信息时，您的任务是合并和更新记忆列表，以反映最准确和最新的信息。您还会获得每个现有记忆与新信息的匹配分数。请务必利用这些信息，就更新或合并哪些记忆做出明智的决策。

指南：
- 消除重复记忆并合并相关记忆，以确保列表简洁且是最新的。
- 如果新信息与某一记忆直接矛盾，请仔细评估这两部分信息：
    - 如果新记忆提供了更新或更准确的更新内容，用新记忆替换旧记忆。
    - 如果新记忆似乎不准确或不够详细，则保留原始记忆并丢弃新记忆。
- 在所有记忆中保持一致、清晰的风格，确保每个条目都简洁且内容丰富。
- 如果新记忆是现有记忆的变体或扩展，请更新现有记忆以反映新信息。

以下是任务的详细信息：
- 现有记忆：
{现有记忆}

- 新记忆：{记忆}