Software 2.0

用RedisVL构建长期记忆

admin -

在本周末笔记中，我们继续讨论如何使用 RedisVL 为智能体构建长期记忆。

当我们为智能体构建长期记忆模块时，最需要关心两点：

  • 长期运行后，保存的记忆是否会变得太大导致上下文爆炸？
  • 我们如何回忆与当前上下文最相关的记忆？

今天我们将解决这两个问题。

TLDR：在这个实践教程中，我们首先使用 LLM 从用户消息中提取对后续聊天有价值的信息。然后将其作为长期记忆存储在 RedisVL 中。需要时，我们通过语义搜索查找相关记忆。通过这种方式，智能体理解用户的过去上下文并给出更准确的回答。

使用这种长期记忆，我们不必担心长期运行后的记忆爆炸。我们也不必担心不相关的记忆会损害 LLM 的回复。

1、人类如何处理记忆

1.1 什么值得记忆

首先，我们需要知道一件事。只有围绕我并与我紧密相关的信息才值得写在便利贴上。

那么关于我的什么信息我想写下来呢？

  • 偏好设置：如我喜欢的工具、我使用的语言、我的日程安排、我说话时的语气
  • 稳定的个人信息：如我的角色、我的时区、我的日常习惯
  • 目标和决策：如选择的选项、计划、以"我决定……"开头的句子
  • 关键里程碑：如工作变动、搬家、截止日期、产品发布
  • 工作和项目上下文：如项目名称、利益相关者、需求、状态如"已完成/下一步"
  • 重复的痛点或强烈观点：这些会改变 LLM 后续的建议
  • 我说"记住这个……"或"不要忘记……"的事情

1.2 什么不值得记忆

我不打算存储任何 LLM 的回答。LLM 对同一问题的回答会随着上下文而变化。因此 LLM 的回答在长期记忆中帮助不大。

除了 LLM 的回答，我也不想保留这些：

  • 一次性的小事情，以后可能不重要
  • 非常敏感的个人数据，如健康诊断、确切地址、政府ID、密码、银行账户
  • 我明确要求不要记住的事情
  • 我已经写在便利贴上的事情

2、设计 LLM 记忆提取的提示词

现在我们知道人类如何处理记忆了。接下来，我想构建一个遵循相同规则并从我的日常聊天中提取记忆的智能体。

关键在于 system prompt。我需要在系统提示中描述所有规则。然后我要求智能体以非常高的一致性遵循这些规则。

在过去，我可能会尝试一些"写1000行提示词"的挑战。现在我不需要那样做。我只需要打开任何 LLM 客户端，粘贴这些规则，然后让 LLM 帮我写一个 system prompt。这只需要不到一分钟。

经过几次尝试，我选了一个我喜欢的。这是那个 system prompt：

你的工作：
仅基于用户的当前输入和现有的相关记忆，决定是否需要添加新的"长期记忆"，如果需要，**只提取一个事实**。你不与用户交谈。你只处理记忆提取和去重。

---

### 1. 核心原则

1. 只保存**将来可能有用**的信息。
2. **每轮最多一个事实**，且必须清楚地出现在当前输入中。
3. **绝不发明或推断任何事情**。你只能重述或轻微改写用户明确说过的话。
4. 如果当前输入没有值得保留的内容，或信息已在相关记忆中，则不要添加新记忆。

---

### 2. 什么算作"长期记忆"

只考虑以下类别，并判断信息是否有长期价值：

由于篇幅原因，这里只展示部分提示词。你可以从文末的源代码中获取完整提示词。

3、为长期记忆构建上下文提供器

完成记忆提取规则后，我们开始为智能体构建长期记忆模块。

为了将来使用，我仍然选择 Microsoft Agent Framework MAF。它提供了 ContextProvider 功能，让我们可以用简单的方式将长期记忆插入到智能体中。

当然，长期记忆的原则保持不变。你可以使用任何你喜欢的智能体框架并构建自己的记忆模块。或者你可以忽略框架，首先构建记忆的存储和检索，然后通过函数调用来调用它们。这也可以。

顺序运行记忆提取

在上篇文章中，我已经用 ContextProvider 构建了一个长期记忆模块。新版本看起来很相似。但这次，我们使用 LLM 来提取记忆。因此在我们设置 ContextProvider 后，我们首先使用 system prompt 构建一个记忆提取智能体。

如果你还不知道如何使用 ContextProvider，我建议你再读一遍我上篇文章。那篇文章详细解释了 Microsoft Agent Framework 中的 ChatMessageStore 和 ContextProvider：

为了避免从 Redis 获取太多不相关的数据，我将 distance_threshold 值设置得相当小。但不是太小。如果太小，就失去了意义。你可以选择你喜欢的值。

class LongTermMemory(ContextProvider):
    def __init__(
        self,
        thread_id: str | None = None,
        session_tag: str | None = None,
        distance_threshold: float = 0.3,
        context_prompt: str = ContextProvider.DEFAULT_CONTEXT_PROMPT,
        redis_url: str = "redis://localhost:6379",
        embedding_model: str = "BAAI/bge-m3",
        llm_model: str = Qwen3.NEXT,
        llm_api_key: str | None = None,
        llm_base_url: str | None = None,
    ):
        ...
        self._init_extractor()

    def _init_extractor(self):
        with open("prompt.md", "r", encoding="utf-8") as f:
                system_prompt = f.read()

        self._extractor = OpenAILikeChatClient(
            model_id=self._llm_model,
        ).as_agent(
            name="extractor",
            instructions=system_prompt,
            default_options={
                "response_format": ExtractResult,
                "extra_body": {"enable_thinking": False}
            },
        )

接下来我们实现 invoking 方法。这个方法在用户智能体调用 LLM 之前运行。在这个方法中，我们提取并存储长期记忆。

为了让逻辑清晰，我按顺序实现 invoking 方法，如图中所示：

当新的用户请求进入 ContextProvider 时，我们首先通过语义搜索在 RedisVL 中搜索最相似的记忆。

class LongTermMemory(ContextProvider):
    ...

    async def invoking(
        self,
        messages: ChatMessage | MutableSequence[ChatMessage],
        **kwargs: Any
    ) -> Context:
        if isinstance(messages, ChatMessage):
            messages = [messages]
        prompt = "\n".join([m.text for m in messages])

        line_sep_memories = self._get_line_sep_memories(prompt)
        ...

    def _get_line_sep_memories(self, prompt: str) -> str:
        context = self._semantic_store.get_relevant(prompt, role="user", session_tag=self._session_tag)
        line_sep_memories = "\n".join([f"* {str(m.get("content", ""))}" for m in context])

        return line_sep_memories

接下来，我们将这些现有记忆加上用户请求发送给记忆提取智能体。该智能体首先根据规则检查是否有值得保存的内容。然后它从用户请求中提取一个新的有用的记忆并将其保存到 RedisVL 中。

class LongTermMemory(ContextProvider):
    ...

    async def invoking(
        self,
        messages: ChatMessage | MutableSequence[ChatMessage],
        **kwargs: Any
    ) -> Context:
        ...
        await self._save_memory(messages, line_sep_memories)
        ...

    async def _save_memory(
        self,
        messages: ChatMessage | MutableSequence[ChatMessage],
        relevant_memory: str | None = None,
    ) -> None:
        detect_messages = (
            [
            ChatMessage(role=Role.USER, text=f"Existing related memories：\n\n{relevant_memory}"),
            ] + list(messages)
            if relevant_memory.strip()
            else list(messages)
        )
        response = await self._extractor.run(detect_messages)

        extract_result: ExtractResult = cast(ExtractResult, response.value)
        if extract_result.should_write_memory:
            self._semantic_store.add_messages(
                messages=[
                    {"role": "user", "content": extract_result.memory_to_write}
                ],
                session_tag=self._session_tag,
            )

最后，我们将来自 RedisVL 的记忆作为额外上下文放入 Context 中。这些记忆消息被合并到真实聊天智能体的历史中。它们给聊天智能体提供额外的背景来生成回答。

class LongTermMemory(ContextProvider):
    ...

    async def invoking(
        self,
        messages: ChatMessage | MutableSequence[ChatMessage],
        **kwargs: Any
    ) -> Context:
        ...

        return Context(messages=[
            ChatMessage(role="user", text=f"{self._context_prompt}\n{line_sep_memories}")
        ] if len(line_sep_memories)>0 else None)

现在我们可以构建一个简单的聊天智能体来测试新的长期记忆模块：

agent = OpenAILikeChatClient(
    model_id=Qwen3.MAX
).as_agent(
    name="assistant",
    instructions="You are a helpful assistant.",
    context_provider=LongTermMemory(),
)

async def main():
    thread = agent.get_new_thread()

原文链接: Advanced RedisVL Long-term Memory Tutorial: Using an LLM to Extract Memories

汇智网翻译整理，转载请标明出处