Tuana Çelik • 2025-05-13

改进的长期和短期记忆适用于 LlamaIndex 代理

代理

与 Logan Markewich 合著

对于任何需要保留基本信息的代理应用，无论是关于过去的对话、用户、他们过去的互动等等，记忆都是一个核心组件，这不足为奇。因此，作为我们对 LlamaIndex 最新一轮改进的一部分，我们引入了一个新的改进的 Memory 组件。

在本文中，我们将介绍新的 LlamaIndex 记忆组件的一些核心功能，以及如何在您自己的代理应用中开始使用它。首先，我们将探讨记忆的最基本实现，即我们只存储聊天消息历史。然后，我们还将介绍通过我们最新添加的长期记忆块实现的更高级用法。

何时使用记忆

并非所有 AI 应用都需要记忆实现。我们今天看到的许多代理应用不一定依赖于聊天历史，或关于用户的持久信息。例如，我们自己的 LlamaExtract 是一个代理式文档提取应用，它使用 LLM 从复杂文件中提取结构化信息。它在执行此操作时不需要保留任何形式的记忆。

但是，如果一个应用的核心依赖于您不想重复的持久信息（例如某人的姓名和年龄），和/或任何形式的对话流程（因此几乎任何带有聊天界面的应用），记忆很快就成为一个非常关键的组件。

考虑一下这种未实现任何形式记忆的简单（且令人恼火的）聊天流程

注意到当问到第二个问题时，代理无法提供任何答案。简单原因是，在句子“Can you tell me what the weather is like there?”中，它没有关于“there”（那里）是什么的上下文。

这可以通过一个非常简单的步骤来解决，即我们存储聊天消息历史，并将其作为上下文发送给 LLM。这正是 LlamaIndex 记忆的最基本形式的开始方式。

记忆的基本用法

在最简单的形式中，LlamaIndex Memory 将把符合给定 token 限制的任意数量的聊天消息存储到 SQL 数据库中，默认使用内存中的 SQLite 数据库。简而言之，这个记忆组件能够存储聊天消息，但有上限。一旦达到该上限（取决于记忆组件的配置方式），聊天历史中最旧的消息将被丢弃，或刷新到长期记忆中。

因此，在其最基本的形式中，LlamaIndex 记忆组件允许您构建能够保留过去对话的代理工作流。您可以通过初始化带有记忆的代理来将聊天历史存储到此记忆中

from llama_index.core.agent.workflow import FunctionAgent
from llama_index.core.memory import Memory

memory = Memory.from_defaults(session_id="my_session", token_limit=40000)

agent = FunctionAgent(llm=llm, tools=tools)

response = await agent.run("<question that invokes tool>", memory=memory)

或者，通过使用 memory.put_messages() 将特定的聊天消息添加到记忆中

memory = Memory.from_defaults(session_id="my_session",
                              token_limit=40000)
memory.put_messages(
    [
        ChatMessage(role="user", content="Hello, world!"),
        ChatMessage(role="assistant", content="Hello, world to you too!"),
    ]
)
chat_history = memory.get()

长期记忆块

对于某些用例，上述简单的聊天历史实现可能已经足够。但对于其他用例，实现更高级的功能更有意义。我们引入了 3 个新的“记忆块”，它们充当长期记忆。

静态记忆块：用于跟踪静态的、不变的信息
事实提取记忆块：它填充从对话中提取的事实列表
向量记忆块：它允许我们利用经典的嵌入生成向量搜索作为一种方式，既可以存储聊天历史，也可以获取旧对话的相关部分。

一个记忆组件可以用任意数量的这些长期记忆块初始化。然后，每当短期记忆达到 token 限制时，它会将相关信息写入这些长期记忆块。要添加长期记忆块，我们使用 memory_blocks 初始化记忆组件。

blocks = [<list of long-term memory blocks>]

memory = Memory.from_defaults(
    session_id="my_session",
    memory_blocks=blocks,
    insert_method="system")

让我们更详细地看看如何初始化每个块。

将静态信息作为长期记忆

想象一下可能与您的代理应用相关且随时间变化的可能性较小的信息。例如，您的姓名、居住地、工作地点等。StaticMemoryBlock 允许您将这些信息作为附加上下文提供给您的代理。

from llama_index.core.memory import StaticMemoryBlock

static_info_block = StaticMemoryBlock(
    name="core_info", 
	static_content="My name is Tuana, and I live in Amsterdam. I work at LlamaIndex.",
	priority=0
)

您可能认为这里的信息是事实，因此更适合事实提取块。主要区别在于事实提取块本身是一个由 LLM 驱动的提取系统，旨在在对话进行时提取事实。

将事实作为长期记忆

我个人最喜欢的一个。FactExtractionMemoryBlock 是一个独特的长期记忆块，它使用默认提示（您可以覆盖）初始化，该提示指示 LLM 从正在进行的对话中提取事实列表。要初始化此块，

from llama_index.core.memory import FactExtractionMemoryBlock
from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-4o-mini")
 
facts_block = FactExtractionMemoryBlock(
    name="extracted_info",
	llm=llm,
    max_facts=50,
    priority=1
)

然后，例如，在与代理进行长时间对话并提供一些关于我自己的信息后，我们可能会得到以下内容

memory.memory_blocks[1].facts

# ['User is 29 years old.', 'User has a sister.']

将向量搜索作为长期记忆

最后，我们还有 VectorMemoryBlock，它必须使用像 Weaviate、Qdrant 或类似的向量存储进行初始化。这里的想法很简单：一旦我们达到短期记忆的 token 限制，记忆组件将把发生过的所有聊天消息写入我们初始化此块时使用的向量存储中。完成此操作后，任何正在进行的对话，如果适用，都可以从历史中获取相关对话，并在回复用户之前将其用作上下文。

from llama_index.core.memory import VectorMemoryBlock

vector_block = VectorMemoryBlock(
    name="vector_memory",
    # required: pass in a vector store like qdrant, chroma, weaviate, milvus, etc.
    vector_store=vector_store,
    priority=2,
    embed_model=embed_model,
    similarity_top_k=2,
)

定制记忆

除此之外，我们还允许您通过继承 BaseMemoryBlock 类来引入您自己的记忆块。例如，下面我们定义了一个记忆块，用于计算对话中给定名字的提及次数。

from typing import Optional, List, Any
from llama_index.core.llms import ChatMessage
from llama_index.core.memory.memory import BaseMemoryBlock

class MentionCounter(BaseMemoryBlock[str]):
    """
    A memory block that counts the number of times a user mentions a specific name.
    """
    mention_name: str = "Logan"
    mention_count: int = 0

    async def _aget(self, messages: Optional[List[ChatMessage]] = None, **block_kwargs: Any) -> str:
        return f"Logan was mentioned {self.mention_count} times."

    async def _aput(self, messages: List[ChatMessage]) -> None:
        for message in messages:
            if self.mention_name in message.content:
                self.mention_count += 1

    async def atruncate(self, content: str, tokens_to_truncate: int) -> Optional[str]:
        return ""