如何添加线程级持久化（函数式 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,
    *,
    # you can optionally specify `previous` in the workflow function signature
    # to access the return value from the workflow as of the last execution
    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 添加线程级持久化，请查阅此操作指南。

设置¶

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

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

设置LangSmith进行 LangGraph 开发

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

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

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

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

^{API 参考：ChatAnthropic}

from langchain_anthropic import ChatAnthropic

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

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

^{API 参考：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))

如果我们尝试使用此工作流，对话的上下文将跨交互持久化

注意

如果您使用 LangGraph Platform 或 LangGraph Studio，您无需将检查点器传递给 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.

流式令牌

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