Skip to content

内存管理

什么是记忆?

记忆是一种认知功能,它允许人们存储、检索并使用信息来理解他们的现在和未来。想象一下与一个同事合作时的挫败感,他忘记了你告诉他的所有事情,导致你需要不断地重复!随着AI代理承担越来越多涉及大量用户交互的复杂任务,为它们配备记忆变得同样重要,以提高效率和用户体验。有了记忆,代理可以从反馈中学习,并适应用户的偏好。本指南涵盖了两种基于回忆范围的记忆类型:

短期记忆,或称为线程范围内的记忆,可以在单个对话线程内随时调用。LangGraph将短期记忆作为代理的状态的一部分进行管理。状态通过检查点器保存到数据库中,因此可以随时恢复线程。每当图被调用或步骤完成时,短期记忆就会更新,并且在每个步骤开始时读取状态。

**长期记忆**是跨多个对话线程共享的。它可以随时在**任何线程**中调用。记忆被限定在任何自定义命名空间中,而不仅仅是在单一线程ID内。LangGraph提供了存储参考文档),让你能够保存和调用长期记忆。

这两种记忆类型都对理解和实现你的应用程序非常重要。

短期记忆与长期记忆对比

短期记忆

短期记忆可以让您的应用程序记住单个线程或对话中的先前交互。一个线程组织会话中的多个交互,类似于电子邮件将消息分组在单个对话中。

LangGraph管理代理状态的一部分作为短期记忆,并通过线程范围检查点持久化。此状态通常包括对话历史记录以及其他有状态数据,如上传的文件、检索的文档或生成的工件。通过将这些信息存储在图的状态中,机器人可以访问给定对话的完整上下文,同时保持不同线程之间的分离。

由于对话历史记录是表示短期记忆最常见的形式,在下一节中,我们将介绍管理对话历史记录的技术,当消息列表变得**很长**时。如果您想继续了解高层次的概念,请继续阅读长期记忆部分。

管理长对话历史记录

长时间的对话对当今的大型语言模型(LLM)构成了挑战。完整的对话历史可能甚至无法适应LLM的上下文窗口,导致不可恢复的错误。即使您的LLM从技术上支持整个上下文长度,大多数LLM仍然在长上下文中表现不佳。它们会被过时或离题的内容所“分散注意力”,同时响应时间变慢且成本更高。

管理短期记忆是一种平衡精度与召回率以及应用程序的其他性能要求(延迟与成本)的过程。正如我们一直强调的那样,重要的是要批判性地思考您如何为您的LLM表示信息,并查看您的数据。以下是一些常见的管理消息列表的技术,希望为您提供足够的背景信息,以便为您的应用程序选择最佳的权衡:

编辑消息列表

聊天模型接受使用消息提供的上下文,这些消息包括开发者提供的指令(系统消息)和用户输入(人类消息)。在聊天应用中,消息交替出现在人类输入和模型响应之间,随着时间推移形成一个不断增长的消息列表。由于上下文窗口有限且包含丰富标记的消息列表可能代价高昂,许多应用程序可以从手动删除或忘记过时信息的技术中受益。

最直接的方法是从列表中删除旧消息(类似于最近最少使用缓存)。

在LangGraph中,从列表中删除内容的典型方法是由节点返回一个更新,告诉系统删除列表的一部分。您可以定义这个更新的样子,但常见的方式是让您返回一个对象或字典,指定要保留哪些值。

def manage_list(existing: list, updates: Union[list, dict]):
    if isinstance(updates, list):
        # 正常情况,添加到历史记录
        return existing + updates
    elif isinstance(updates, dict) and updates["type"] == "keep":
        # 您可以决定这看起来像什么。
        # 例如,您可以简化并只接受字符串 "DELETE"
        # 并清除整个列表。
        return existing[updates["from"]:updates["to"]]
    # 等等。我们定义如何解释更新

class State(TypedDict):
    my_list: Annotated[list, manage_list]

def my_node(state: State):
    return {
        # 我们返回一个更新,说要保留索引 -5 到末尾的值(删除其余部分)
        "my_list": {"type": "keep", "from": -5, "to": None}
    }

LangGraph将在每次返回键为 "my_list" 的更新时调用 manage_list "reducer" 函数。在此函数内,我们定义要接受的更新类型。通常情况下,消息将被添加到现有列表中(对话将增长);然而,我们也添加了支持接受一个字典,使您能够“保留”某些状态部分的功能。这允许您程序化地丢弃旧消息上下文。

另一种常见的方法是让您返回一个包含“删除”对象的列表,这些对象指定了所有要删除的消息的ID。如果您正在使用LangChain的消息和LangGraph中的add_messages reducer(或使用相同底层功能的MessagesState),则可以使用一个RemoveMessage来实现这一点。

API Reference: RemoveMessage | AIMessage | add_messages

from langchain_core.messages import RemoveMessage, AIMessage
from langgraph.graph import add_messages
# ... 其他导入

class State(TypedDict):
    # add_messages 将默认根据ID插入消息到现有列表中
    # 如果返回一个 RemoveMessage,则会删除列表中具有该ID的消息
    messages: Annotated[list, add_messages]

def my_node_1(state: State):
    # 向状态中的 `messages` 列表添加一个AI消息
    return {"messages": [AIMessage(content="Hi")]}

def my_node_2(state: State):
    # 删除状态中的 `messages` 列表除最后两个消息外的所有消息
    delete_messages = [RemoveMessage(id=m.id) for m in state['messages'][:-2]]
    return {"messages": delete_messages}

在上述示例中,add_messages reducer 允许我们在 messages 状态键中追加新消息,如 my_node_1 所示。当它看到一个 RemoveMessage 时,它会删除列表中具有该ID的消息(然后丢弃该 RemoveMessage)。有关LangChain特定的消息处理的更多信息,请参阅此关于使用 RemoveMessage 的指南

参见此指南教程和我们的LangChain学院课程模块2中的示例用法。

总结过去的对话

如上所示,修剪或删除消息的问题在于我们可能会因消息队列的剪裁而丢失信息。因此,一些应用程序从更复杂的方法中受益,即使用聊天模型总结消息历史记录。

简单的提示和编排逻辑可以用来实现这一点。例如,在LangGraph中,我们可以扩展MessagesState,以包括一个 summary 键。

from langgraph.graph import MessagesState
class State(MessagesState):
    summary: str

然后,我们可以生成对话历史记录的摘要,使用任何现有的摘要作为下一个摘要的上下文。这个 summarize_conversation 节点可以在一些消息累积在 messages 状态键之后被调用。

def summarize_conversation(state: State):

    # 首先,获取任何现有的摘要
    summary = state.get("summary", "")

    # 创建我们的摘要提示
    if summary:

        # 已经存在摘要
        summary_message = (
            f"This is a summary of the conversation to date: {summary}\n\n"
            "Extend the summary by taking into account the new messages above:"
        )

    else:
        summary_message = "Create a summary of the conversation above:"

    # 将提示添加到我们的历史记录中
    messages = state["messages"] + [HumanMessage(content=summary_message)]
    response = model.invoke(messages)

    # 删除除最新两条消息外的所有消息
    delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]
    return {"summary": response.content, "messages": delete_messages}

参见此指南这里和我们的LangChain学院课程模块2中的示例用法。

知道何时删除消息

大多数LLM都有一个最大支持的上下文窗口(以token计数)。一个简单的方法来决定何时截断消息是计算消息历史记录中的token数量,并在接近该限制时截断。朴素的截断很容易自己实现,尽管有一些“陷阱”。一些模型API进一步限制消息类型的序列(必须以人类消息开始,不能有连续相同类型的消息等)。如果您使用LangChain,您可以使用trim_messages工具,并指定要从列表中保留的token数量,以及用于处理边界的策略(例如,保留最后max_tokens)。

下面是示例。

API Reference: trim_messages

from langchain_core.messages import trim_messages
trim_messages(
    messages,
    # 保留消息列表中最后 <= n_count 个token。
    strategy="last",
    # 记得根据您的模型调整
    # 或者传递自定义token_encoder
    token_counter=ChatOpenAI(model="gpt-4"),
    # 记得根据所需的对话长度调整
    max_tokens=45,
    # 大多数聊天模型期望聊天历史记录以以下之一开始:
    # (1) 一条HumanMessage或
    # (2) 一条SystemMessage后跟一条HumanMessage
    start_on="human",
    # 大多数聊天模型期望聊天历史记录以以下之一结束:
    # (1) 一条HumanMessage或
    # (2) 一条ToolMessage
    end_on=("human", "tool"),
    # 通常,我们希望保留原始历史记录中存在的SystemMessage
    # 因为它包含了模型的特殊指令。
    include_system=True,
)

长期记忆

在 LangGraph 中,长期记忆允许系统跨不同对话或会话保留信息。与仅限于线程范围的短期记忆不同,长期记忆保存在自定义的“命名空间”中。

存储记忆

LangGraph 将长期记忆存储为 JSON 文档在一个存储参考文档)中。每个记忆都组织在一个自定义的 namespace(类似于文件夹)和一个独特的 key(类似于文件名)。命名空间通常包括用户或组织ID或其他标签,这使得信息更容易被组织。这种结构使记忆的层次化组织成为可能。通过内容过滤器支持跨命名空间搜索。请参见下面的例子。

from langgraph.store.memory import InMemoryStore


def embed(texts: list[str]) -> list[list[float]]:
    # 用实际的嵌入函数或LangChain嵌入对象替换
    return [[1.0, 2.0] * len(texts)]


# InMemoryStore 将数据保存到内存字典中。在生产环境中使用数据库支持的存储。
store = InMemoryStore(index={"embed": embed, "dims": 2})
user_id = "my-user"
application_context = "chitchat"
namespace = (user_id, application_context)
store.put(
    namespace,
    "a-memory",
    {
        "rules": [
            "用户喜欢简短直接的语言",
            "用户只说英语和Python",
        ],
        "my-key": "my-value",
    },
)
# 根据ID获取“记忆”
item = store.get(namespace, "a-memory")
# 在此命名空间内搜索“记忆”,根据内容等价性筛选,并按向量相似度排序
items = store.search(
    namespace, filter={"my-key": "my-value"}, query="语言偏好"
)

关于长期记忆的思考框架

长期记忆是一个没有通用解决方案的复杂挑战。然而,以下问题提供了一个结构化的框架来帮助您导航不同的技术:

记忆的类型是什么?

人类使用记忆来记住事实经历规则。AI代理也可以以相同的方式使用记忆。例如,AI代理可以使用记忆来记住特定用户的事实以完成任务。我们将在下一部分中扩展几种类型的记忆。

何时更新记忆?

记忆可以在代理的应用逻辑中更新(例如,“在热路径上”)。在这种情况下,代理通常会在响应用户之前决定记住某些事实。或者,记忆可以在后台任务中更新(运行在后台/异步并生成记忆的逻辑)。我们在下一部分中解释了这两种方法之间的权衡。

内存类型

不同的应用程序需要各种类型的内存。虽然这种类比并不完美,但研究人类记忆类型可以提供一些启示。有些研究(例如CoALA论文)甚至将这些人类记忆类型映射到了AI代理中使用的类型。

记忆类型 存储内容 人类示例 代理示例
语义 事实 我在学校学到的东西 用户的事实
事件性 经历 我做过的事情 过去的代理行为
程序性 指令 本能或运动技能 代理系统提示

语义记忆

语义记忆,无论是人类还是AI代理,都涉及特定事实和概念的保留。在人类中,它可能包括在学校学到的信息以及对概念及其关系的理解。对于AI代理,语义记忆通常用于通过记住过去交互中的事实或概念来个性化应用。

注意:不要将其与“语义搜索”混淆,“语义搜索”是一种使用“意义”(通常是嵌入式表示)查找相似内容的技术。语义记忆是心理学中的一个术语,指的是存储事实和知识,而语义搜索是一种基于意义而非精确匹配检索信息的方法。

基本资料

语义记忆可以通过不同的方式进行管理。例如,记忆可以是一个单一的、持续更新的“基本资料”,其中包含关于用户、组织或其他实体(包括代理本身)的具体且特定的信息。一个基本资料通常只是一个JSON文档,包含你选择的各种键值对以代表你的领域。

当你记住一个基本资料时,你希望每次都能**更新**这个基本资料。因此,你需要传递之前的资料,并要求模型生成一个新的基本资料(或者一些JSON补丁应用于旧的基本资料)。随着基本资料变得越来越大,这可能会变得容易出错,并可能受益于将基本资料拆分为多个文档或在生成文档时进行**严格的**解码,以确保记忆模式的有效性。

集合

另一种方法是将记忆视为随着时间不断更新和扩展的一系列文档。每个单独的记忆可以更具体地定义并且更容易生成,这意味着你不太可能随着时间的推移而**丢失**信息。对于大型语言模型来说,为新信息生成新的对象比与现有基本资料进行协调要容易得多。因此,文档集合往往会导致下游更高的召回率

然而,这将一些复杂性转移到了记忆更新上。模型现在必须删除或更新列表中的现有项目,这可能比较棘手。此外,某些模型可能会默认过度插入,而其他模型则会默认过度更新。查看Trustcall包以了解一种管理此问题的方法,并考虑评估(例如,使用LangSmith工具)以帮助你调整行为。

处理文档集合也会将复杂性转移到对列表的记忆**搜索**上。Store目前支持语义搜索根据内容筛选

最后,使用一系列记忆可能会使向模型提供全面上下文变得具有挑战性。尽管单个记忆可能遵循特定的模式,但这结构可能无法捕捉到记忆之间的完整上下文或关系。因此,在使用这些记忆生成响应时,模型可能会缺乏重要的上下文信息,这些信息在统一的基本资料方法中会更加容易获得。

无论采用哪种记忆管理方法,核心要点是代理将使用语义记忆来使其响应扎根,这通常会导致更个性化的相关交互。

事件性记忆

事件性记忆,无论是人类还是AI代理,都涉及回忆过去的事件或行动。CoALA论文对此进行了很好的阐述:事实可以写入语义记忆,而*经历*可以写入事件性记忆。对于AI代理,事件性记忆通常用于帮助代理记住如何完成任务。

实际上,事件性记忆常常通过少量样本示例提示来实现,其中代理从过去的序列中学习正确执行任务。有时,“展示”比“告诉”更容易,LLM从示例中学得很好。少量样本学习让你通过用输入-输出示例更新提示来说明预期行为,从而"编程"你的LLM。虽然可以使用各种最佳实践来生成少量样本示例,但挑战通常在于根据用户输入选择最相关的示例。

需要注意的是,记忆存储只是存储数据作为少量样本示例的一种方式。如果你希望有更多的开发者参与,或将少量样本更紧密地绑定到你的评估框架,你可以使用一个LangSmith数据集来存储你的数据。然后可以使用现成的动态少量样本示例选择器来实现相同的目标。LangSmith将为你索引数据集,并启用基于关键词相似性的检索(使用类似BM25的算法)以获取与用户输入最相关的少量样本示例。

参见这个如何使用视频来了解LangSmith中动态少量样本示例选择的示例用法。同时,参见这篇博客文章展示了如何使用少量样本提示来提高工具调用性能,以及这篇博客文章展示了如何使用少量样本示例来将LLM与人类偏好对齐。

程序性记忆

程序性记忆,无论是人类还是AI代理,都涉及记住用于执行任务的规则。在人类中,程序性记忆就像内化了的知识,比如骑自行车的基本运动技能和平衡感。另一方面,事件性记忆涉及回忆具体的经历,比如第一次成功地骑自行车而没有训练轮或一次难忘的风景骑行。对于AI代理,程序性记忆是模型权重、代理代码和代理提示的组合,共同决定了代理的功能。

实际上,代理修改其模型权重或重写其代码的情况相对较少。然而,代理修改自己的提示更为常见。

一种有效的方法是通过"反思"或元提示来精炼代理的指令。这涉及到用当前指令(如系统提示)以及最近的对话或明确的用户反馈来提示代理。然后,代理根据这些输入来精炼自己的指令。这种方法特别适用于那些难以提前指定指令的任务,因为它允许代理从其交互中学习和适应。

例如,我们构建了一个Tweet生成器,使用外部反馈和提示重写来为Twitter生成高质量的论文摘要。在这种情况下,具体的摘要提示很难事先指定,但用户很容易批评生成的推文并提供反馈以改进摘要过程。

以下伪代码显示了如何使用LangGraph记忆存储来实现这一点,使用存储来保存提示,使用update_instructions节点来获取当前提示(以及从对话中捕获的用户反馈state["messages"]),更新提示,并将新提示保存回存储。然后,call_model从存储中获取更新后的提示并使用它来生成响应。

# 节点使用指令
def call_model(state: State, store: BaseStore):
    namespace = ("agent_instructions", )
    instructions = store.get(namespace, key="agent_a")[0]
    # 应用逻辑
    prompt = prompt_template.format(instructions=instructions.value["instructions"])
    ...

# 节点更新指令
def update_instructions(state: State, store: BaseStore):
    namespace = ("instructions",)
    current_instructions = store.search(namespace)[0]
    # 记忆逻辑
    prompt = prompt_template.format(instructions=instructions.value["instructions"], conversation=state["messages"])
    output = llm.invoke(prompt)
    new_instructions = output['new_instructions']
    store.put(("agent_instructions",), "agent_a", {"instructions": new_instructions})
    ...

写入记忆

虽然人类通常在睡眠期间形成长期记忆,但AI代理需要不同的方法。代理何时以及如何创建新的记忆?至少有两种主要的方法来写入记忆:“在热路径上”和“在后台”。

在热路径上写入记忆

在运行时创建记忆提供了优势同时也带来了挑战。从积极的一面来看,这种方法允许实时更新,使新记忆立即可用于后续交互。它还能够提高透明度,用户可以被通知当记忆被创建并存储时。

然而,这种方法也带来了一些挑战。如果代理需要一个新的工具来决定要保存什么记忆,这可能会增加复杂性。此外,关于要保存什么内容的记忆推理过程可能会影响代理的延迟。最后,代理必须在记忆创建和其他职责之间多任务处理,这可能会影响所创建记忆的数量和质量。

例如,ChatGPT使用一个save_memories工具来更新或插入记忆作为字符串内容,并决定是否以及如何使用此工具来处理每个用户的消息。参见我们的记忆代理模板以获取参考实现。

在后台写入记忆

将记忆创建作为一个单独的后台任务执行提供了一些优势。它可以消除主应用程序中的延迟,分离应用程序逻辑与内存管理,并允许代理专注于完成特定任务。这种方法还提供了在避免重复工作的同时灵活安排记忆创建时间的能力。

然而,这种方法也有其自身的挑战。确定记忆写入的频率变得至关重要,因为不频繁的更新可能会导致其他线程缺乏新的上下文。决定触发记忆形成的时间也很重要。常见的策略包括在设定的时间段后调度(如果有新的事件发生则重新调度)、使用cron调度或者允许用户或应用程序逻辑手动触发。

参见我们的记忆服务模板以获取参考实现。

Comments