跳到内容

记忆

什么是记忆?

记忆 是一种认知功能,它使人们能够存储、检索和使用信息来理解他们的现在和未来。想象一下,与一位总是忘记你告诉他的一切、需要不断重复的同事一起工作是多么令人沮丧!随着 AI 智能体承担涉及大量用户交互的更复杂任务,为它们配备记忆对于效率和用户满意度同样至关重要。有了记忆,智能体可以从反馈中学习并适应用户的偏好。本指南涵盖基于回忆范围的两种记忆类型

短时记忆,或称为 线程 范围内的记忆,可以随时在与用户的单一对话线程内部进行回忆。LangGraph 将短时记忆作为智能体 状态 的一部分进行管理。状态通过 检查点 持久化到数据库中,以便线程可以随时恢复。当调用图或完成一个步骤时,短时记忆会更新,并在每个步骤开始时读取状态。

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

理解和实现这两者对于你的应用都很重要。

短时记忆

短时记忆让你的应用能够记住单一 线程 或对话中的先前交互。一个 线程 将一个会话中的多次交互组织起来,类似于电子邮件将消息分组到单一对话中的方式。

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

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

管理长对话历史记录

长对话对当前的 LLM 提出了挑战。完整的历史记录甚至可能无法完全放入 LLM 的上下文窗口内,导致不可恢复的错误。即使你的 LLM 在技术上支持完整的上下文长度,大多数 LLM 在处理长上下文时表现仍然不佳。它们会被陈旧或无关的内容“分心”,同时还面临响应时间变慢和成本升高的问题。

管理短时记忆是在 精确率与召回率 和应用的其他性能要求(延迟和成本)之间进行平衡的实践。一如既往,批判性地思考如何为你的 LLM 表示信息并查看你的数据非常重要。我们在下面介绍了一些管理消息列表的常用技术,希望为你提供足够的上下文,以便你为你的应用选择最佳权衡方案

编辑消息列表

聊天模型使用 消息 来接受上下文,其中包括开发者提供的指令(系统消息)和用户输入(人类消息)。在聊天应用中,消息在人类输入和模型响应之间交替,导致消息列表随着时间的推移而增长。由于上下文窗口有限且包含大量 token 的消息列表可能成本高昂,许多应用可以受益于使用手动移除或遗忘陈旧信息的技术。

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

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

def manage_list(existing: list, updates: Union[list, dict]):
    if isinstance(updates, list):
        # Normal case, add to the history
        return existing + updates
    elif isinstance(updates, dict) and updates["type"] == "keep":
        # You get to decide what this looks like.
        # For example, you could simplify and just accept a string "DELETE"
        # and clear the entire list.
        return existing[updates["from"]:updates["to"]]
    # etc. We define how to interpret updates

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

def my_node(state: State):
    return {
        # We return an update for the field "my_list" saying to
        # keep only values from index -5 to the end (deleting the rest)
        "my_list": {"type": "keep", "from": -5, "to": None}
    }

LangGraph 会在键为 "my_list" 下返回更新时调用 manage_listreducer” 函数。在该函数中,我们定义接受哪种类型的更新。通常,消息会被添加到现有列表中(对话会增长);然而,我们也添加了支持接受一个字典,让你能够“保留”状态的某些部分。这让你能够以编程方式丢弃旧的消息上下文。

另一种常见的方法是允许你返回一个“移除”对象列表,指定所有要删除的消息的 ID。如果你在 LangGraph 中使用 LangChain 消息和 add_messages reducer(或使用相同底层功能的 MessagesState),你可以使用 RemoveMessage 来完成此操作。

API 参考:RemoveMessage | AIMessage | add_messages

from langchain_core.messages import RemoveMessage, AIMessage
from langgraph.graph import add_messages
# ... other imports

class State(TypedDict):
    # add_messages will default to upserting messages by ID to the existing list
    # if a RemoveMessage is returned, it will delete the message in the list by ID
    messages: Annotated[list, add_messages]

def my_node_1(state: State):
    # Add an AI message to the `messages` list in the state
    return {"messages": [AIMessage(content="Hi")]}

def my_node_2(state: State):
    # Delete all but the last 2 messages from the `messages` list in the state
    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):

    # First, we get any existing summary
    summary = state.get("summary", "")

    # Create our summarization prompt
    if summary:

        # A summary already exists
        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:"

    # Add prompt to our history
    messages = state["messages"] + [HumanMessage(content=summary_message)]
    response = model.invoke(messages)

    # Delete all but the 2 most recent 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 数量,以及用于处理边界的 strategy(例如,保留最后 max_tokens 个)。

下面是一个示例。

API 参考:trim_messages

from langchain_core.messages import trim_messages
trim_messages(
    messages,
    # Keep the last <= n_count tokens of the messages.
    strategy="last",
    # Remember to adjust based on your model
    # or else pass a custom token_encoder
    token_counter=ChatOpenAI(model="gpt-4"),
    # Remember to adjust based on the desired conversation
    # length
    max_tokens=45,
    # Most chat models expect that chat history starts with either:
    # (1) a HumanMessage or
    # (2) a SystemMessage followed by a HumanMessage
    start_on="human",
    # Most chat models expect that chat history ends with either:
    # (1) a HumanMessage or
    # (2) a ToolMessage
    end_on=("human", "tool"),
    # Usually, we want to keep the SystemMessage
    # if it's present in the original history.
    # The SystemMessage has special instructions for the model.
    include_system=True,
)

长时记忆

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

存储记忆

LangGraph 将长时记忆作为 JSON 文档存储在 存储参考文档)中。每个记忆都组织在一个自定义的 namespace(类似于文件夹)和一个独特的 key(类似于文件名)下。命名空间通常包含用户或组织 ID 或其他标签,以便更容易组织信息。这种结构实现了记忆的层次化组织。然后通过内容过滤器支持跨命名空间搜索。请参阅下面的示例。

from langgraph.store.memory import InMemoryStore


def embed(texts: list[str]) -> list[list[float]]:
    # Replace with an actual embedding function or LangChain embeddings object
    return [[1.0, 2.0] * len(texts)]


# InMemoryStore saves data to an in-memory dictionary. Use a DB-backed store in production use.
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": [
            "User likes short, direct language",
            "User only speaks English & python",
        ],
        "my-key": "my-value",
    },
)
# get the "memory" by ID
item = store.get(namespace, "a-memory")
# search for "memories" within this namespace, filtering on content equivalence, sorted by vector similarity
items = store.search(
    namespace, filter={"my-key": "my-value"}, query="language preferences"
)

关于长时记忆的思考框架

长时记忆是一个复杂的挑战,没有一刀切的解决方案。然而,以下问题提供了一个结构化框架,可以帮助你探索不同的技术

记忆的类型是什么?

人类利用记忆来记住 事实经历规则。AI 智能体也可以以同样的方式使用记忆。例如,AI 智能体可以使用记忆来记住关于用户的具体事实,从而完成任务。我们在下面一节中详细阐述了几种记忆类型。

你想在何时更新记忆?

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

记忆类型

不同的应用需要各种类型的记忆。尽管这种类比并不完美,但研究 人类记忆类型 可以提供深刻见解。一些研究(例如,CoALA 论文)甚至将这些人类记忆类型映射到 AI 智能体中使用的记忆类型。

记忆类型 存储内容 人类示例 智能体示例
语义 事实 我在学校学到的知识 关于用户的事实
情景 经历 我做过的事情 过去的智能体动作
程序 指令 直觉或运动技能 智能体系统提示

语义记忆

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

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

档案

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

在记忆档案时,你会希望确保每次都更新档案。因此,你会想要传入之前的档案,并要求模型生成一个新的档案(或应用于旧档案的一些 JSON patch)。随着档案变得更大,这可能会变得容易出错,并且可能会受益于将档案拆分成多个文档,或者在生成文档时进行严格解码,以确保记忆模式保持有效。

集合

或者,记忆可以是一个文档集合,随着时间的推移不断更新和扩展。每个单独的记忆都可以范围更窄且更容易生成,这意味着你随时间推移丢失信息的可能性较小。LLM 生成新的信息对象比将新信息与现有档案协调起来更容易。因此,文档集合倾向于导致下游的 更高召回率

然而,这会将记忆更新的一些复杂性转移出去。模型现在必须删除更新列表中的现有项,这可能很棘手。此外,一些模型可能默认为过度插入,而另一些模型可能默认为过度更新。请参阅 Trustcall 包以了解一种管理方法,并考虑使用评估工具(例如 LangSmith)来帮助你调整行为。

处理文档集合也会将复杂性转移到列表上的记忆搜索Store 当前同时支持 语义搜索按内容过滤

最后,使用记忆集合可能会导致难以向模型提供全面的上下文。虽然单个记忆可能遵循特定的模式,但这种结构可能无法捕获记忆之间的完整上下文或关系。因此,在使用这些记忆生成响应时,模型可能会缺乏重要的上下文信息,而这些信息在统一档案方法中更容易获得。

无论采用何种记忆管理方法,核心要点是智能体将使用语义记忆来生成其响应,这通常会导致更个性化和更相关的交互。

情景记忆

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

在实践中,情景记忆通常通过 少样本示例提示 来实现,智能体从过去的序列中学习以正确执行任务。有时,“展示”比“讲述”更容易,而且 LLM 善于从示例中学习。少样本学习允许你通过使用输入输出示例更新提示来“编程”你的 LLM,以说明预期行为。虽然可以使用各种 最佳实践 来生成少样本示例,但挑战通常在于根据用户输入选择最相关的示例。

请注意,记忆 存储 只是将数据存储为少样本示例的一种方式。如果你想让开发者更多地参与,或者将少样本更紧密地与你的评估工具关联起来,你也可以使用 LangSmith 数据集 来存储你的数据。然后可以直接使用动态少样本示例选择器来实现相同目标。LangSmith 会为你索引数据集,并根据关键字相似性(使用类似于 BM25 的算法进行基于关键字的相似性匹配)检索与用户输入最相关的少样本示例。

有关 LangSmith 中动态少样本示例选择的示例用法,请参见此操作 视频。此外,请参见这篇 博客文章,展示了如何使用少样本提示来提高工具调用性能,以及这篇 博客文章,展示了如何使用少样本示例使 LLM 与人类偏好对齐。

程序记忆

程序记忆,无论是人类还是 AI 智能体,都涉及记住用于执行任务的规则。在人类中,程序记忆就像如何执行任务的内化知识,例如通过基本的运动技能和平衡来骑自行车。另一方面,情景记忆涉及回忆特定的经历,例如第一次成功脱离辅助轮骑自行车,或一次难忘的风景优美的自行车旅行。对于 AI 智能体,程序记忆是模型权重、智能体代码和智能体提示的组合,它们共同决定了智能体的功能。

在实践中,智能体修改其模型权重或重写其代码的情况相当罕见。然而,智能体修改其自身提示的情况更为常见。

完善智能体指令的一种有效方法是通过 “反思” 或元提示。这涉及向智能体提供其当前指令(例如系统提示)以及最近的对话或明确的用户反馈。然后智能体根据此输入修改其自身指令。这种方法对于那些指令难以预先指定的任务特别有用,因为它允许智能体从交互中学习和适应。

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

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

# Node that *uses* the instructions
def call_model(state: State, store: BaseStore):
    namespace = ("agent_instructions", )
    instructions = store.get(namespace, key="agent_a")[0]
    # Application logic
    prompt = prompt_template.format(instructions=instructions.value["instructions"])
    ...

# Node that updates instructions
def update_instructions(state: State, store: BaseStore):
    namespace = ("instructions",)
    current_instructions = store.search(namespace)[0]
    # Memory logic
    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 工具将记忆作为内容字符串进行 upsert,并决定是否以及如何对每个用户消息使用此工具。请参阅我们的 memory-agent 模板作为参考实现。

在后台写入记忆

将记忆创建作为一个独立的后台任务具有多个优点。它消除了主应用的延迟,将应用逻辑与记忆管理分开,并允许智能体更专注于完成任务。这种方法还为记忆创建的时机提供了灵活性,以避免重复工作。

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

请参阅我们的 memory-service 模板作为参考实现。

评论