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

先决条件

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

LangGraph API 用户不需要

如果您使用 LangGraph API，则无需手动实现 checkpointer。该 API 会自动为您处理 checkpointing。本指南适用于您在自己的自定义服务器中实现 LangGraph 时。

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

创建 LangGraph 工作流时，您可以使用 checkpointer 来设置其结果的持久化：

创建 checkpointer 实例：

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,
    *,
    # 您可以 J选择性地在工作流函数签名中指定 `previous`
    # 以访问上次执行时工作流的返回值
    previous
):
    previous = previous or []
    combined_inputs = previous + inputs
    result = do_something(combined_inputs)
    ...

可选地选择哪些值将从工作流返回，哪些将由 checkpointer 保存为 previous：

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

本指南介绍了如何为您的工作流添加线程级别持久化。

注意

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

注意

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

设置¶

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

pip install --quiet -U langgraph langchain_anthropic

接下来，我们需要为 Anthropic（我们将使用的 LLM）设置 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 构建的 LLM 应用 — 在此了解更多入门信息。

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

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

首先，我们来定义将要使用的模型：

^{API Reference: ChatAnthropic}

from langchain_anthropic import ChatAnthropic

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

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

^{API Reference: BaseMessage | add_messages | entrypoint | task | MemorySaver}

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))

如果我们尝试使用此工作流，对话的上下文将在交互之间持续存在：

Note

如果你正在使用 LangGraph Platform 或 LangGraph Studio，你__不需要__将 checkpointer 传递给 entrypoint 装饰器，因为它会自动完成。

现在我们可以与代理进行交互，并且可以看到它记住了以前的消息！

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()

================================== Ai Message ==================================

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()

================================== Ai Message ==================================

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()

================================== Ai Message ==================================

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.

Streaming tokens

如果您希望从聊天机器人中流式传输 LLM 令牌，可以使用 stream_mode="messages"。请参阅本操作指南以了解更多信息。