如何添加线程级持久化（函数式 API）¶

前提条件

本指南假设你熟悉以下内容：

LangGraph API 用户无需操作

如果你使用的是 LangGraph API，则无需手动实现检查点保存器。该 API 会自动为你处理检查点保存。本指南适用于在你自己的自定义服务器中实现 LangGraph 的情况。

许多 AI 应用程序需要内存来在同一线程上的多次交互（例如，对话的多个回合）之间共享上下文。在 LangGraph 函数式 API 中，可以使用线程级持久化将这种内存添加到任何 entrypoint() 工作流中。

在创建 LangGraph 工作流时，你可以通过使用检查点保存器来设置其持久化结果：

创建一个检查点保存器实例：

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

将 checkpointer 实例传递给 entrypoint() 装饰器：

from langgraph.func import entrypoint

@entrypoint(checkpointer=checkpointer)
def workflow(inputs)
    ...

可选择在工作流函数签名中暴露 previous 参数：

@entrypoint(checkpointer=checkpointer)
def workflow(
    inputs,
    *,
    # 你可以选择在工作流函数签名中指定 `previous`
    # 以访问工作流上一次执行的返回值
    previous
):
    previous = previous or []
    combined_inputs = previous + inputs
    result = do_something(combined_inputs)
    ...

可选择指定哪些值将从工作流中返回，哪些值将由检查点保存器作为 previous 保存：

@entrypoint(checkpointer=checkpointer)
def workflow(inputs, *, previous):
    ...
    result = do_something(...)
    return entrypoint.final(value=result, save=combine(inputs, result))

本指南展示了如何为你的工作流添加线程级持久化。

注意

如果你需要在多个对话或用户之间**共享**的内存（跨线程持久化），请查看此操作指南。

注意

如果你需要为 StateGraph 添加线程级持久化，请查看此操作指南。

安装设置¶

首先，我们需要安装所需的软件包。

%%capture --no-stderr
%pip install --quiet -U langgraph langchain_anthropic

接下来，我们需要为 Anthropic（我们将使用的大语言模型）设置 API 密钥。

import getpass
import os


def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")


_set_env("ANTHROPIC_API_KEY")

为 LangGraph 开发设置 LangSmith

注册 LangSmith 以快速发现问题并提升你的 LangGraph 项目的性能。LangSmith 允许你使用跟踪数据来调试、测试和监控使用 LangGraph 构建的大语言模型应用程序 — 点击此处了解更多关于如何开始使用的信息。

示例：具有短期记忆的简单聊天机器人¶

我们将使用一个包含单个任务的工作流，该任务会调用一个聊天模型。

让我们首先定义我们将使用的模型：

from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(model="claude-3-5-sonnet-latest")

现在我们可以定义我们的任务和工作流了。为了添加持久化功能，我们需要将一个 Checkpointer 传递给 entrypoint() 装饰器。

from langchain_core.messages import BaseMessage
from langgraph.graph import add_messages
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import MemorySaver


@task
def call_model(messages: list[BaseMessage]):
    response = model.invoke(messages)
    return response


checkpointer = MemorySaver()


@entrypoint(checkpointer=checkpointer)
def workflow(inputs: list[BaseMessage], *, previous: list[BaseMessage]):
    if previous:
        inputs = add_messages(previous, inputs)

    response = call_model(inputs).result()
    return entrypoint.final(value=response, save=add_messages(inputs, response))

API Reference: BaseMessage

如果我们尝试使用此工作流程，对话的上下文将在多次交互中得以保留：

Note

如果您正在使用 LangGraph Cloud 或 LangGraph Studio，您__无需__将检查点器（checkpointer）传递给入口点装饰器，因为这会自动完成。

现在我们可以与智能体进行交互，并发现它能记住之前的消息！

config = {"configurable": {"thread_id": "1"}}
input_message = {"role": "user", "content": "hi! I'm bob"}
for chunk in workflow.stream([input_message], config, stream_mode="values"):
    chunk.pretty_print()

==================================[1m Ai Message [0m==================================

Hi Bob! I'm Claude. Nice to meet you! How are you today?

你始终可以恢复之前的对话线程：

input_message = {"role": "user", "content": "what's my name?"}
for chunk in workflow.stream([input_message], config, stream_mode="values"):
    chunk.pretty_print()

==================================[1m Ai Message [0m==================================

Your name is Bob.

如果我们想开启一个新的对话，我们可以传入一个不同的 thread_id。噗！所有的记忆都消失了！

input_message = {"role": "user", "content": "what's my name?"}
for chunk in workflow.stream(
    [input_message],
    {"configurable": {"thread_id": "2"}},
    stream_mode="values",
):
    chunk.pretty_print()

==================================[1m Ai Message [0m==================================

I don't know your name unless you tell me. Each conversation I have starts fresh, so I don't have access to any previous interactions or personal information unless you share it with me.

流式传输令牌

如果你想从你的聊天机器人流式传输大语言模型（LLM）令牌，可以使用 stream_mode="messages"。查看此操作指南以了解更多信息。

如何添加线程级持久化（函数式 API）¶

安装设置¶

示例：具有短期记忆的简单聊天机器人¶

Comments