跳至内容

检查点器

CheckpointMetadata

基类: TypedDict

与检查点关联的元数据。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
class CheckpointMetadata(TypedDict, total=False):
    """Metadata associated with a checkpoint."""

    source: Literal["input", "loop", "update"]
    """The source of the checkpoint.

    - "input": The checkpoint was created from an input to invoke/stream/batch.
    - "loop": The checkpoint was created from inside the pregel loop.
    - "update": The checkpoint was created from a manual state update.
    """
    step: int
    """The step number of the checkpoint.

    -1 for the first "input" checkpoint.
    0 for the first "loop" checkpoint.
    ... for the nth checkpoint afterwards.
    """
    writes: dict[str, Any]
    """The writes that were made between the previous checkpoint and this one.

    Mapping from node name to writes emitted by that node.
    """
    parents: dict[str, str]
    """The IDs of the parent checkpoints.

    Mapping from checkpoint namespace to checkpoint ID.
    """

source: Literal['input', 'loop', 'update'] instance-attribute

检查点的来源。

  • "input": 检查点是从输入创建的,用于调用/流/批处理。
  • "loop": 检查点是从 pregel 循环内部创建的。
  • "update": 检查点是从手动状态更新创建的。

step: int instance-attribute

检查点的步骤号。

-1 表示第一个 "input" 检查点。0 表示第一个 "loop" 检查点。... 表示之后的第 n 个检查点。

writes: dict[str, Any] instance-attribute

在之前的检查点和当前检查点之间所做的写入操作。

从节点名称到该节点发出的写入操作的映射。

parents: dict[str, str] instance-attribute

父检查点的 ID。

从检查点命名空间到检查点 ID 的映射。

Checkpoint

基类: TypedDict

在特定时间点的状态快照。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
class Checkpoint(TypedDict):
    """State snapshot at a given point in time."""

    v: int
    """The version of the checkpoint format. Currently 1."""
    id: str
    """The ID of the checkpoint. This is both unique and monotonically
    increasing, so can be used for sorting checkpoints from first to last."""
    ts: str
    """The timestamp of the checkpoint in ISO 8601 format."""
    channel_values: dict[str, Any]
    """The values of the channels at the time of the checkpoint.
    Mapping from channel name to deserialized channel snapshot value.
    """
    channel_versions: ChannelVersions
    """The versions of the channels at the time of the checkpoint.
    The keys are channel names and the values are monotonically increasing
    version strings for each channel.
    """
    versions_seen: dict[str, ChannelVersions]
    """Map from node ID to map from channel name to version seen.
    This keeps track of the versions of the channels that each node has seen.
    Used to determine which nodes to execute next.
    """
    pending_sends: List[SendProtocol]
    """List of inputs pushed to nodes but not yet processed.
    Cleared by the next checkpoint."""

v: int instance-attribute

检查点格式的版本。当前为 1。

id: str instance-attribute

检查点的 ID。它既是唯一的又是单调递增的,因此可以用于对检查点进行排序,从第一个到最后一个。

ts: str instance-attribute

检查点的 ISO 8601 格式的时间戳。

channel_values: dict[str, Any] instance-attribute

检查点时通道的值。从通道名称到反序列化的通道快照值的映射。

channel_versions: ChannelVersions instance-attribute

检查点时通道的版本。键是通道名称,值是每个通道的单调递增版本字符串。

versions_seen: dict[str, ChannelVersions] instance-attribute

从节点 ID 到从通道名称到已查看版本的映射。这会跟踪每个节点已查看的通道的版本。用于确定接下来要执行哪些节点。

pending_sends: List[SendProtocol] instance-attribute

推送到节点但尚未处理的输入列表。在下一个检查点被清除。

BaseCheckpointSaver

基类: Generic[V]

用于创建图形检查点器的基类。

检查点器允许 LangGraph 代理在多个交互中以及跨多个交互持久化其状态。

属性

  • serde (SerializerProtocol) –

    用于编码/解码检查点的序列化器。

注意

在创建自定义检查点保存器时,请考虑实现异步版本,以避免阻塞主线程。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
class BaseCheckpointSaver(Generic[V]):
    """Base class for creating a graph checkpointer.

    Checkpointers allow LangGraph agents to persist their state
    within and across multiple interactions.

    Attributes:
        serde (SerializerProtocol): Serializer for encoding/decoding checkpoints.

    Note:
        When creating a custom checkpoint saver, consider implementing async
        versions to avoid blocking the main thread.
    """

    serde: SerializerProtocol = JsonPlusSerializer()

    def __init__(
        self,
        *,
        serde: Optional[SerializerProtocol] = None,
    ) -> None:
        self.serde = maybe_add_typed_methods(serde or self.serde)

    @property
    def config_specs(self) -> list[ConfigurableFieldSpec]:
        """Define the configuration options for the checkpoint saver.

        Returns:
            list[ConfigurableFieldSpec]: List of configuration field specs.
        """
        return [CheckpointThreadId, CheckpointNS, CheckpointId]

    def get(self, config: RunnableConfig) -> Optional[Checkpoint]:
        """Fetch a checkpoint using the given configuration.

        Args:
            config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

        Returns:
            Optional[Checkpoint]: The requested checkpoint, or None if not found.
        """
        if value := self.get_tuple(config):
            return value.checkpoint

    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Fetch a checkpoint tuple using the given configuration.

        Args:
            config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

        Returns:
            Optional[CheckpointTuple]: The requested checkpoint tuple, or None if not found.

        Raises:
            NotImplementedError: Implement this method in your custom checkpoint saver.
        """
        raise NotImplementedError

    def list(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[Dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> Iterator[CheckpointTuple]:
        """List checkpoints that match the given criteria.

        Args:
            config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
            filter (Optional[Dict[str, Any]]): Additional filtering criteria.
            before (Optional[RunnableConfig]): List checkpoints created before this configuration.
            limit (Optional[int]): Maximum number of checkpoints to return.

        Returns:
            Iterator[CheckpointTuple]: Iterator of matching checkpoint tuples.

        Raises:
            NotImplementedError: Implement this method in your custom checkpoint saver.
        """
        raise NotImplementedError

    def put(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Store a checkpoint with its configuration and metadata.

        Args:
            config (RunnableConfig): Configuration for the checkpoint.
            checkpoint (Checkpoint): The checkpoint to store.
            metadata (CheckpointMetadata): Additional metadata for the checkpoint.
            new_versions (ChannelVersions): New channel versions as of this write.

        Returns:
            RunnableConfig: Updated configuration after storing the checkpoint.

        Raises:
            NotImplementedError: Implement this method in your custom checkpoint saver.
        """
        raise NotImplementedError

    def put_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[Tuple[str, Any]],
        task_id: str,
    ) -> None:
        """Store intermediate writes linked to a checkpoint.

        Args:
            config (RunnableConfig): Configuration of the related checkpoint.
            writes (List[Tuple[str, Any]]): List of writes to store.
            task_id (str): Identifier for the task creating the writes.

        Raises:
            NotImplementedError: Implement this method in your custom checkpoint saver.
        """
        raise NotImplementedError

    async def aget(self, config: RunnableConfig) -> Optional[Checkpoint]:
        """Asynchronously fetch a checkpoint using the given configuration.

        Args:
            config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

        Returns:
            Optional[Checkpoint]: The requested checkpoint, or None if not found.
        """
        if value := await self.aget_tuple(config):
            return value.checkpoint

    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Asynchronously fetch a checkpoint tuple using the given configuration.

        Args:
            config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

        Returns:
            Optional[CheckpointTuple]: The requested checkpoint tuple, or None if not found.

        Raises:
            NotImplementedError: Implement this method in your custom checkpoint saver.
        """
        raise NotImplementedError

    async def alist(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[Dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> AsyncIterator[CheckpointTuple]:
        """Asynchronously list checkpoints that match the given criteria.

        Args:
            config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
            filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
            before (Optional[RunnableConfig]): List checkpoints created before this configuration.
            limit (Optional[int]): Maximum number of checkpoints to return.

        Returns:
            AsyncIterator[CheckpointTuple]: Async iterator of matching checkpoint tuples.

        Raises:
            NotImplementedError: Implement this method in your custom checkpoint saver.
        """
        raise NotImplementedError
        yield

    async def aput(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Asynchronously store a checkpoint with its configuration and metadata.

        Args:
            config (RunnableConfig): Configuration for the checkpoint.
            checkpoint (Checkpoint): The checkpoint to store.
            metadata (CheckpointMetadata): Additional metadata for the checkpoint.
            new_versions (ChannelVersions): New channel versions as of this write.

        Returns:
            RunnableConfig: Updated configuration after storing the checkpoint.

        Raises:
            NotImplementedError: Implement this method in your custom checkpoint saver.
        """
        raise NotImplementedError

    async def aput_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[Tuple[str, Any]],
        task_id: str,
    ) -> None:
        """Asynchronously store intermediate writes linked to a checkpoint.

        Args:
            config (RunnableConfig): Configuration of the related checkpoint.
            writes (List[Tuple[str, Any]]): List of writes to store.
            task_id (str): Identifier for the task creating the writes.

        Raises:
            NotImplementedError: Implement this method in your custom checkpoint saver.
        """
        raise NotImplementedError

    def get_next_version(self, current: Optional[V], channel: ChannelProtocol) -> V:
        """Generate the next version ID for a channel.

        Default is to use integer versions, incrementing by 1. If you override, you can use str/int/float versions,
        as long as they are monotonically increasing.

        Args:
            current (Optional[V]): The current version identifier (int, float, or str).
            channel (BaseChannel): The channel being versioned.

        Returns:
            V: The next version identifier, which must be increasing.
        """
        if isinstance(current, str):
            raise NotImplementedError
        elif current is None:
            return 1
        else:
            return current + 1

config_specs: list[ConfigurableFieldSpec] property

定义检查点保存器的配置选项。

返回值

  • list[ConfigurableFieldSpec]

    list[ConfigurableFieldSpec]: 配置字段规范的列表。

get(config: RunnableConfig) -> Optional[Checkpoint]

使用给定的配置获取检查点。

参数

  • config (RunnableConfig) –

    指定要检索的检查点的配置。

返回值

  • Optional[Checkpoint]

    Optional[Checkpoint]: 请求的检查点,如果未找到则为 None。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
def get(self, config: RunnableConfig) -> Optional[Checkpoint]:
    """Fetch a checkpoint using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[Checkpoint]: The requested checkpoint, or None if not found.
    """
    if value := self.get_tuple(config):
        return value.checkpoint

get_tuple(config: RunnableConfig) -> Optional[CheckpointTuple]

使用给定的配置获取检查点元组。

参数

  • config (RunnableConfig) –

    指定要检索的检查点的配置。

返回值

  • Optional[CheckpointTuple]

    Optional[CheckpointTuple]: 请求的检查点元组,如果未找到则为 None。

引发

  • NotImplementedError

    在您的自定义检查点保存器中实现此方法。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Fetch a checkpoint tuple using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[CheckpointTuple]: The requested checkpoint tuple, or None if not found.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

list(config: Optional[RunnableConfig], *, filter: Optional[Dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None) -> Iterator[CheckpointTuple]

列出符合给定条件的检查点。

参数

  • config (Optional[RunnableConfig]) –

    用于过滤检查点的基本配置。

  • filter (Optional[Dict[str, Any]], 默认: None ) –

    其他过滤条件。

  • before (Optional[RunnableConfig], 默认: None ) –

    列出在此配置之前创建的检查点。

  • limit (Optional[int], 默认: None ) –

    要返回的最大检查点数。

返回值

  • Iterator[CheckpointTuple]

    Iterator[CheckpointTuple]: 匹配的检查点元组的迭代器。

引发

  • NotImplementedError

    在您的自定义检查点保存器中实现此方法。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
def list(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> Iterator[CheckpointTuple]:
    """List checkpoints that match the given criteria.

    Args:
        config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
        filter (Optional[Dict[str, Any]]): Additional filtering criteria.
        before (Optional[RunnableConfig]): List checkpoints created before this configuration.
        limit (Optional[int]): Maximum number of checkpoints to return.

    Returns:
        Iterator[CheckpointTuple]: Iterator of matching checkpoint tuples.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

put(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions) -> RunnableConfig

使用其配置和元数据存储检查点。

参数

  • config (RunnableConfig) –

    检查点的配置。

  • checkpoint (Checkpoint) –

    要存储的检查点。

  • metadata (CheckpointMetadata) –

    检查点的其他元数据。

  • new_versions (ChannelVersions) –

    截至此次写入的新的通道版本。

返回值

  • RunnableConfig ( RunnableConfig ) –

    存储检查点后的更新配置。

引发

  • NotImplementedError

    在您的自定义检查点保存器中实现此方法。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
def put(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Store a checkpoint with its configuration and metadata.

    Args:
        config (RunnableConfig): Configuration for the checkpoint.
        checkpoint (Checkpoint): The checkpoint to store.
        metadata (CheckpointMetadata): Additional metadata for the checkpoint.
        new_versions (ChannelVersions): New channel versions as of this write.

    Returns:
        RunnableConfig: Updated configuration after storing the checkpoint.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

put_writes(config: RunnableConfig, writes: Sequence[Tuple[str, Any]], task_id: str) -> None

存储与检查点链接的中间写入内容。

参数

  • config (RunnableConfig) –

    相关检查点的配置。

  • writes (List[Tuple[str, Any]]) –

    要存储的写入内容列表。

  • task_id (str) –

    创建写入内容的任务标识符。

引发

  • NotImplementedError

    在您的自定义检查点保存器中实现此方法。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
def put_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[Tuple[str, Any]],
    task_id: str,
) -> None:
    """Store intermediate writes linked to a checkpoint.

    Args:
        config (RunnableConfig): Configuration of the related checkpoint.
        writes (List[Tuple[str, Any]]): List of writes to store.
        task_id (str): Identifier for the task creating the writes.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

aget(config: RunnableConfig) -> Optional[Checkpoint] async

使用给定配置异步获取检查点。

参数

  • config (RunnableConfig) –

    指定要检索的检查点的配置。

返回值

  • Optional[Checkpoint]

    Optional[Checkpoint]: 请求的检查点,如果未找到则为 None。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
async def aget(self, config: RunnableConfig) -> Optional[Checkpoint]:
    """Asynchronously fetch a checkpoint using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[Checkpoint]: The requested checkpoint, or None if not found.
    """
    if value := await self.aget_tuple(config):
        return value.checkpoint

aget_tuple(config: RunnableConfig) -> Optional[CheckpointTuple] async

使用给定配置异步获取检查点元组。

参数

  • config (RunnableConfig) –

    指定要检索的检查点的配置。

返回值

  • Optional[CheckpointTuple]

    Optional[CheckpointTuple]: 请求的检查点元组,如果未找到则为 None。

引发

  • NotImplementedError

    在您的自定义检查点保存器中实现此方法。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Asynchronously fetch a checkpoint tuple using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[CheckpointTuple]: The requested checkpoint tuple, or None if not found.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

alist(config: Optional[RunnableConfig], *, filter: Optional[Dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None) -> AsyncIterator[CheckpointTuple] async

异步列出符合给定条件的检查点。

参数

  • config (Optional[RunnableConfig]) –

    用于过滤检查点的基本配置。

  • filter (Optional[Dict[str, Any]], 默认: None ) –

    用于元数据的其他过滤条件。

  • before (Optional[RunnableConfig], 默认: None ) –

    列出在此配置之前创建的检查点。

  • limit (Optional[int], 默认: None ) –

    要返回的最大检查点数。

返回值

  • AsyncIterator[CheckpointTuple]

    AsyncIterator[CheckpointTuple]: 匹配的检查点元组的异步迭代器。

引发

  • NotImplementedError

    在您的自定义检查点保存器中实现此方法。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
async def alist(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> AsyncIterator[CheckpointTuple]:
    """Asynchronously list checkpoints that match the given criteria.

    Args:
        config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
        filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
        before (Optional[RunnableConfig]): List checkpoints created before this configuration.
        limit (Optional[int]): Maximum number of checkpoints to return.

    Returns:
        AsyncIterator[CheckpointTuple]: Async iterator of matching checkpoint tuples.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError
    yield

aput(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions) -> RunnableConfig async

异步使用其配置和元数据存储检查点。

参数

  • config (RunnableConfig) –

    检查点的配置。

  • checkpoint (Checkpoint) –

    要存储的检查点。

  • metadata (CheckpointMetadata) –

    检查点的其他元数据。

  • new_versions (ChannelVersions) –

    截至此次写入的新的通道版本。

返回值

  • RunnableConfig ( RunnableConfig ) –

    存储检查点后的更新配置。

引发

  • NotImplementedError

    在您的自定义检查点保存器中实现此方法。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
async def aput(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Asynchronously store a checkpoint with its configuration and metadata.

    Args:
        config (RunnableConfig): Configuration for the checkpoint.
        checkpoint (Checkpoint): The checkpoint to store.
        metadata (CheckpointMetadata): Additional metadata for the checkpoint.
        new_versions (ChannelVersions): New channel versions as of this write.

    Returns:
        RunnableConfig: Updated configuration after storing the checkpoint.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

aput_writes(config: RunnableConfig, writes: Sequence[Tuple[str, Any]], task_id: str) -> None async

异步存储与检查点链接的中间写入内容。

参数

  • config (RunnableConfig) –

    相关检查点的配置。

  • writes (List[Tuple[str, Any]]) –

    要存储的写入内容列表。

  • task_id (str) –

    创建写入内容的任务标识符。

引发

  • NotImplementedError

    在您的自定义检查点保存器中实现此方法。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
async def aput_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[Tuple[str, Any]],
    task_id: str,
) -> None:
    """Asynchronously store intermediate writes linked to a checkpoint.

    Args:
        config (RunnableConfig): Configuration of the related checkpoint.
        writes (List[Tuple[str, Any]]): List of writes to store.
        task_id (str): Identifier for the task creating the writes.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

get_next_version(current: Optional[V], channel: ChannelProtocol) -> V

为通道生成下一个版本 ID。

默认使用整数版本,每次递增 1。如果覆盖,可以使用 str/int/float 版本,只要它们是单调递增的即可。

参数

  • current (Optional[V]) –

    当前版本标识符(int、float 或 str)。

  • channel (BaseChannel) –

    要进行版本控制的通道。

返回值

  • V ( V ) –

    下一个版本标识符,必须递增。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
def get_next_version(self, current: Optional[V], channel: ChannelProtocol) -> V:
    """Generate the next version ID for a channel.

    Default is to use integer versions, incrementing by 1. If you override, you can use str/int/float versions,
    as long as they are monotonically increasing.

    Args:
        current (Optional[V]): The current version identifier (int, float, or str).
        channel (BaseChannel): The channel being versioned.

    Returns:
        V: The next version identifier, which must be increasing.
    """
    if isinstance(current, str):
        raise NotImplementedError
    elif current is None:
        return 1
    else:
        return current + 1

create_checkpoint(checkpoint: Checkpoint, channels: Optional[Mapping[str, ChannelProtocol]], step: int, *, id: Optional[str] = None) -> Checkpoint

为给定的通道创建检查点。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
def create_checkpoint(
    checkpoint: Checkpoint,
    channels: Optional[Mapping[str, ChannelProtocol]],
    step: int,
    *,
    id: Optional[str] = None,
) -> Checkpoint:
    """Create a checkpoint for the given channels."""
    ts = datetime.now(timezone.utc).isoformat()
    if channels is None:
        values = checkpoint["channel_values"]
    else:
        values = {}
        for k, v in channels.items():
            if k not in checkpoint["channel_versions"]:
                continue
            try:
                values[k] = v.checkpoint()
            except EmptyChannelError:
                pass
    return Checkpoint(
        v=1,
        ts=ts,
        id=id or str(uuid6(clock_seq=step)),
        channel_values=values,
        channel_versions=checkpoint["channel_versions"],
        versions_seen=checkpoint["versions_seen"],
        pending_sends=checkpoint.get("pending_sends", []),
    )

SerializerProtocol

基础: Protocol

用于对象序列化和反序列化的协议。

  • dumps: 将对象序列化为字节。
  • dumps_typed: 将对象序列化为元组(类型,字节)。
  • loads: 从字节反序列化对象。
  • loads_typed: 从元组(类型,字节)反序列化对象。

有效的实现包括 picklejsonorjson 模块。

源代码位于 libs/checkpoint/langgraph/checkpoint/serde/base.py
class SerializerProtocol(Protocol):
    """Protocol for serialization and deserialization of objects.

    - `dumps`: Serialize an object to bytes.
    - `dumps_typed`: Serialize an object to a tuple (type, bytes).
    - `loads`: Deserialize an object from bytes.
    - `loads_typed`: Deserialize an object from a tuple (type, bytes).

    Valid implementations include the `pickle`, `json` and `orjson` modules.
    """

    def dumps(self, obj: Any) -> bytes: ...

    def dumps_typed(self, obj: Any) -> tuple[str, bytes]: ...

    def loads(self, data: bytes) -> Any: ...

    def loads_typed(self, data: tuple[str, bytes]) -> Any: ...

JsonPlusSerializer

基础: SerializerProtocol

源代码位于 libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py
class JsonPlusSerializer(SerializerProtocol):
    def _encode_constructor_args(
        self,
        constructor: Union[Callable, type[Any]],
        *,
        method: Union[None, str, Sequence[Union[None, str]]] = None,
        args: Optional[Sequence[Any]] = None,
        kwargs: Optional[dict[str, Any]] = None,
    ) -> dict[str, Any]:
        out = {
            "lc": 2,
            "type": "constructor",
            "id": (*constructor.__module__.split("."), constructor.__name__),
        }
        if method is not None:
            out["method"] = method
        if args is not None:
            out["args"] = args
        if kwargs is not None:
            out["kwargs"] = kwargs
        return out

    def _default(self, obj: Any) -> Union[str, dict[str, Any]]:
        if isinstance(obj, Serializable):
            return cast(dict[str, Any], obj.to_json())
        elif hasattr(obj, "model_dump") and callable(obj.model_dump):
            return self._encode_constructor_args(
                obj.__class__, method=(None, "model_construct"), kwargs=obj.model_dump()
            )
        elif hasattr(obj, "dict") and callable(obj.dict):
            return self._encode_constructor_args(
                obj.__class__, method=(None, "construct"), kwargs=obj.dict()
            )
        elif hasattr(obj, "_asdict") and callable(obj._asdict):
            return self._encode_constructor_args(obj.__class__, kwargs=obj._asdict())
        elif isinstance(obj, pathlib.Path):
            return self._encode_constructor_args(pathlib.Path, args=obj.parts)
        elif isinstance(obj, re.Pattern):
            return self._encode_constructor_args(
                re.compile, args=(obj.pattern, obj.flags)
            )
        elif isinstance(obj, UUID):
            return self._encode_constructor_args(UUID, args=(obj.hex,))
        elif isinstance(obj, decimal.Decimal):
            return self._encode_constructor_args(decimal.Decimal, args=(str(obj),))
        elif isinstance(obj, (set, frozenset, deque)):
            return self._encode_constructor_args(type(obj), args=(tuple(obj),))
        elif isinstance(obj, (IPv4Address, IPv4Interface, IPv4Network)):
            return self._encode_constructor_args(obj.__class__, args=(str(obj),))
        elif isinstance(obj, (IPv6Address, IPv6Interface, IPv6Network)):
            return self._encode_constructor_args(obj.__class__, args=(str(obj),))

        elif isinstance(obj, datetime):
            return self._encode_constructor_args(
                datetime, method="fromisoformat", args=(obj.isoformat(),)
            )
        elif isinstance(obj, timezone):
            return self._encode_constructor_args(
                timezone,
                args=obj.__getinitargs__(),  # type: ignore[attr-defined]
            )
        elif isinstance(obj, ZoneInfo):
            return self._encode_constructor_args(ZoneInfo, args=(obj.key,))
        elif isinstance(obj, timedelta):
            return self._encode_constructor_args(
                timedelta, args=(obj.days, obj.seconds, obj.microseconds)
            )
        elif isinstance(obj, date):
            return self._encode_constructor_args(
                date, args=(obj.year, obj.month, obj.day)
            )
        elif isinstance(obj, time):
            return self._encode_constructor_args(
                time,
                args=(obj.hour, obj.minute, obj.second, obj.microsecond, obj.tzinfo),
                kwargs={"fold": obj.fold},
            )
        elif dataclasses.is_dataclass(obj):
            return self._encode_constructor_args(
                obj.__class__,
                kwargs={
                    field.name: getattr(obj, field.name)
                    for field in dataclasses.fields(obj)
                },
            )
        elif isinstance(obj, Enum):
            return self._encode_constructor_args(obj.__class__, args=(obj.value,))
        elif isinstance(obj, SendProtocol):
            return self._encode_constructor_args(
                obj.__class__, kwargs={"node": obj.node, "arg": obj.arg}
            )
        elif isinstance(obj, (bytes, bytearray)):
            return self._encode_constructor_args(
                obj.__class__, method="fromhex", args=(obj.hex(),)
            )
        elif isinstance(obj, BaseException):
            return repr(obj)
        else:
            raise TypeError(
                f"Object of type {obj.__class__.__name__} is not JSON serializable"
            )

    def _reviver(self, value: dict[str, Any]) -> Any:
        if (
            value.get("lc", None) == 2
            and value.get("type", None) == "constructor"
            and value.get("id", None) is not None
        ):
            try:
                # Get module and class name
                [*module, name] = value["id"]
                # Import module
                mod = importlib.import_module(".".join(module))
                # Import class
                cls = getattr(mod, name)
                # Instantiate class
                method = value.get("method")
                if isinstance(method, str):
                    methods = [getattr(cls, method)]
                elif isinstance(method, list):
                    methods = [
                        cls if method is None else getattr(cls, method)
                        for method in method
                    ]
                else:
                    methods = [cls]
                args = value.get("args")
                kwargs = value.get("kwargs")
                for method in methods:
                    try:
                        if isclass(method) and issubclass(method, BaseException):
                            return None
                        if args and kwargs:
                            return method(*args, **kwargs)
                        elif args:
                            return method(*args)
                        elif kwargs:
                            return method(**kwargs)
                        else:
                            return method()
                    except Exception:
                        continue
            except Exception:
                return None

        return LC_REVIVER(value)

    def dumps(self, obj: Any) -> bytes:
        return json.dumps(obj, default=self._default, ensure_ascii=False).encode(
            "utf-8", "ignore"
        )

    def dumps_typed(self, obj: Any) -> tuple[str, bytes]:
        if isinstance(obj, bytes):
            return "bytes", obj
        elif isinstance(obj, bytearray):
            return "bytearray", obj
        else:
            try:
                return "msgpack", _msgpack_enc(obj)
            except UnicodeEncodeError:
                return "json", self.dumps(obj)

    def loads(self, data: bytes) -> Any:
        return json.loads(data, object_hook=self._reviver)

    def loads_typed(self, data: tuple[str, bytes]) -> Any:
        type_, data_ = data
        if type_ == "bytes":
            return data_
        elif type_ == "bytearray":
            return bytearray(data_)
        elif type_ == "json":
            return self.loads(data_)
        elif type_ == "msgpack":
            return msgpack.unpackb(data_, ext_hook=_msgpack_ext_hook)
        else:
            raise NotImplementedError(f"Unknown serialization type: {type_}")

MemorySaver

基础: BaseCheckpointSaver[str], AbstractContextManager, AbstractAsyncContextManager

内存中的检查点保存器。

此检查点保存器使用 defaultdict 将检查点存储在内存中。

注意

仅将 MemorySaver 用于调试或测试目的。对于生产用例,我们建议安装 langgraph-checkpoint-postgres 并使用 PostgresSaver / AsyncPostgresSaver

参数

  • serde (Optional[SerializerProtocol], 默认: None ) –

    用于序列化和反序列化检查点的序列化程序。默认为 None。

示例

    import asyncio

    from langgraph.checkpoint.memory import MemorySaver
    from langgraph.graph import StateGraph

    builder = StateGraph(int)
    builder.add_node("add_one", lambda x: x + 1)
    builder.set_entry_point("add_one")
    builder.set_finish_point("add_one")

    memory = MemorySaver()
    graph = builder.compile(checkpointer=memory)
    coro = graph.ainvoke(1, {"configurable": {"thread_id": "thread-1"}})
    asyncio.run(coro)  # Output: 2
源代码位于 libs/checkpoint/langgraph/checkpoint/memory/__init__.py
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
class MemorySaver(
    BaseCheckpointSaver[str], AbstractContextManager, AbstractAsyncContextManager
):
    """An in-memory checkpoint saver.

    This checkpoint saver stores checkpoints in memory using a defaultdict.

    Note:
        Only use `MemorySaver` for debugging or testing purposes.
        For production use cases we recommend installing [langgraph-checkpoint-postgres](https://pypi.ac.cn/project/langgraph-checkpoint-postgres/) and using `PostgresSaver` / `AsyncPostgresSaver`.

    Args:
        serde (Optional[SerializerProtocol]): The serializer to use for serializing and deserializing checkpoints. Defaults to None.

    Examples:

            import asyncio

            from langgraph.checkpoint.memory import MemorySaver
            from langgraph.graph import StateGraph

            builder = StateGraph(int)
            builder.add_node("add_one", lambda x: x + 1)
            builder.set_entry_point("add_one")
            builder.set_finish_point("add_one")

            memory = MemorySaver()
            graph = builder.compile(checkpointer=memory)
            coro = graph.ainvoke(1, {"configurable": {"thread_id": "thread-1"}})
            asyncio.run(coro)  # Output: 2
    """

    # thread ID ->  checkpoint NS -> checkpoint ID -> checkpoint mapping
    storage: defaultdict[
        str,
        dict[
            str, dict[str, tuple[tuple[str, bytes], tuple[str, bytes], Optional[str]]]
        ],
    ]
    writes: defaultdict[
        tuple[str, str, str], dict[tuple[str, int], tuple[str, str, tuple[str, bytes]]]
    ]

    def __init__(
        self,
        *,
        serde: Optional[SerializerProtocol] = None,
    ) -> None:
        super().__init__(serde=serde)
        self.storage = defaultdict(lambda: defaultdict(dict))
        self.writes = defaultdict(dict)

    def __enter__(self) -> "MemorySaver":
        return self

    def __exit__(
        self,
        exc_type: Optional[type[BaseException]],
        exc_value: Optional[BaseException],
        traceback: Optional[TracebackType],
    ) -> Optional[bool]:
        return

    async def __aenter__(self) -> "MemorySaver":
        return self

    async def __aexit__(
        self,
        __exc_type: Optional[type[BaseException]],
        __exc_value: Optional[BaseException],
        __traceback: Optional[TracebackType],
    ) -> Optional[bool]:
        return

    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Get a checkpoint tuple from the in-memory storage.

        This method retrieves a checkpoint tuple from the in-memory storage based on the
        provided config. If the config contains a "checkpoint_id" key, the checkpoint with
        the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint
        for the given thread ID is retrieved.

        Args:
            config (RunnableConfig): The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
        """
        thread_id = config["configurable"]["thread_id"]
        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
        if checkpoint_id := get_checkpoint_id(config):
            if saved := self.storage[thread_id][checkpoint_ns].get(checkpoint_id):
                checkpoint, metadata, parent_checkpoint_id = saved
                writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
                if parent_checkpoint_id:
                    sends = [
                        w[2]
                        for w in self.writes[
                            (thread_id, checkpoint_ns, parent_checkpoint_id)
                        ].values()
                        if w[1] == TASKS
                    ]
                else:
                    sends = []
                return CheckpointTuple(
                    config=config,
                    checkpoint={
                        **self.serde.loads_typed(checkpoint),
                        "pending_sends": [self.serde.loads_typed(s) for s in sends],
                    },
                    metadata=self.serde.loads_typed(metadata),
                    pending_writes=[
                        (id, c, self.serde.loads_typed(v)) for id, c, v in writes
                    ],
                    parent_config={
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": parent_checkpoint_id,
                        }
                    }
                    if parent_checkpoint_id
                    else None,
                )
        else:
            if checkpoints := self.storage[thread_id][checkpoint_ns]:
                checkpoint_id = max(checkpoints.keys())
                checkpoint, metadata, parent_checkpoint_id = checkpoints[checkpoint_id]
                writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
                if parent_checkpoint_id:
                    sends = [
                        w[2]
                        for w in self.writes[
                            (thread_id, checkpoint_ns, parent_checkpoint_id)
                        ].values()
                        if w[1] == TASKS
                    ]
                else:
                    sends = []
                return CheckpointTuple(
                    config={
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": checkpoint_id,
                        }
                    },
                    checkpoint={
                        **self.serde.loads_typed(checkpoint),
                        "pending_sends": [self.serde.loads_typed(s) for s in sends],
                    },
                    metadata=self.serde.loads_typed(metadata),
                    pending_writes=[
                        (id, c, self.serde.loads_typed(v)) for id, c, v in writes
                    ],
                    parent_config={
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": parent_checkpoint_id,
                        }
                    }
                    if parent_checkpoint_id
                    else None,
                )

    def list(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[Dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> Iterator[CheckpointTuple]:
        """List checkpoints from the in-memory storage.

        This method retrieves a list of checkpoint tuples from the in-memory storage based
        on the provided criteria.

        Args:
            config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
            filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
            before (Optional[RunnableConfig]): List checkpoints created before this configuration.
            limit (Optional[int]): Maximum number of checkpoints to return.

        Yields:
            Iterator[CheckpointTuple]: An iterator of matching checkpoint tuples.
        """
        thread_ids = (config["configurable"]["thread_id"],) if config else self.storage
        config_checkpoint_ns = (
            config["configurable"].get("checkpoint_ns") if config else None
        )
        config_checkpoint_id = get_checkpoint_id(config) if config else None
        for thread_id in thread_ids:
            for checkpoint_ns in self.storage[thread_id].keys():
                if (
                    config_checkpoint_ns is not None
                    and checkpoint_ns != config_checkpoint_ns
                ):
                    continue

                for checkpoint_id, (
                    checkpoint,
                    metadata_b,
                    parent_checkpoint_id,
                ) in sorted(
                    self.storage[thread_id][checkpoint_ns].items(),
                    key=lambda x: x[0],
                    reverse=True,
                ):
                    # filter by checkpoint ID from config
                    if config_checkpoint_id and checkpoint_id != config_checkpoint_id:
                        continue

                    # filter by checkpoint ID from `before` config
                    if (
                        before
                        and (before_checkpoint_id := get_checkpoint_id(before))
                        and checkpoint_id >= before_checkpoint_id
                    ):
                        continue

                    # filter by metadata
                    metadata = self.serde.loads_typed(metadata_b)
                    if filter and not all(
                        query_value == metadata.get(query_key)
                        for query_key, query_value in filter.items()
                    ):
                        continue

                    # limit search results
                    if limit is not None and limit <= 0:
                        break
                    elif limit is not None:
                        limit -= 1

                    writes = self.writes[
                        (thread_id, checkpoint_ns, checkpoint_id)
                    ].values()

                    if parent_checkpoint_id:
                        sends = [
                            w[2]
                            for w in self.writes[
                                (thread_id, checkpoint_ns, parent_checkpoint_id)
                            ].values()
                            if w[1] == TASKS
                        ]
                    else:
                        sends = []

                    yield CheckpointTuple(
                        config={
                            "configurable": {
                                "thread_id": thread_id,
                                "checkpoint_ns": checkpoint_ns,
                                "checkpoint_id": checkpoint_id,
                            }
                        },
                        checkpoint={
                            **self.serde.loads_typed(checkpoint),
                            "pending_sends": [self.serde.loads_typed(s) for s in sends],
                        },
                        metadata=metadata,
                        parent_config={
                            "configurable": {
                                "thread_id": thread_id,
                                "checkpoint_ns": checkpoint_ns,
                                "checkpoint_id": parent_checkpoint_id,
                            }
                        }
                        if parent_checkpoint_id
                        else None,
                        pending_writes=[
                            (id, c, self.serde.loads_typed(v)) for id, c, v in writes
                        ],
                    )

    def put(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Save a checkpoint to the in-memory storage.

        This method saves a checkpoint to the in-memory storage. The checkpoint is associated
        with the provided config.

        Args:
            config (RunnableConfig): The config to associate with the checkpoint.
            checkpoint (Checkpoint): The checkpoint to save.
            metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
            new_versions (dict): New versions as of this write

        Returns:
            RunnableConfig: The updated config containing the saved checkpoint's timestamp.
        """
        c = checkpoint.copy()
        c.pop("pending_sends")  # type: ignore[misc]
        thread_id = config["configurable"]["thread_id"]
        checkpoint_ns = config["configurable"]["checkpoint_ns"]
        self.storage[thread_id][checkpoint_ns].update(
            {
                checkpoint["id"]: (
                    self.serde.dumps_typed(c),
                    self.serde.dumps_typed(metadata),
                    config["configurable"].get("checkpoint_id"),  # parent
                )
            }
        )
        return {
            "configurable": {
                "thread_id": thread_id,
                "checkpoint_ns": checkpoint_ns,
                "checkpoint_id": checkpoint["id"],
            }
        }

    def put_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[Tuple[str, Any]],
        task_id: str,
    ) -> None:
        """Save a list of writes to the in-memory storage.

        This method saves a list of writes to the in-memory storage. The writes are associated
        with the provided config.

        Args:
            config (RunnableConfig): The config to associate with the writes.
            writes (list[tuple[str, Any]]): The writes to save.
            task_id (str): Identifier for the task creating the writes.

        Returns:
            RunnableConfig: The updated config containing the saved writes' timestamp.
        """
        thread_id = config["configurable"]["thread_id"]
        checkpoint_ns = config["configurable"]["checkpoint_ns"]
        checkpoint_id = config["configurable"]["checkpoint_id"]
        outer_key = (thread_id, checkpoint_ns, checkpoint_id)
        for idx, (c, v) in enumerate(writes):
            inner_key = (task_id, WRITES_IDX_MAP.get(c, idx))
            self.writes[outer_key][inner_key] = (task_id, c, self.serde.dumps_typed(v))

    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Asynchronous version of get_tuple.

        This method is an asynchronous wrapper around get_tuple that runs the synchronous
        method in a separate thread using asyncio.

        Args:
            config (RunnableConfig): The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
        """
        return await asyncio.get_running_loop().run_in_executor(
            None, self.get_tuple, config
        )

    async def alist(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[Dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> AsyncIterator[CheckpointTuple]:
        """Asynchronous version of list.

        This method is an asynchronous wrapper around list that runs the synchronous
        method in a separate thread using asyncio.

        Args:
            config (RunnableConfig): The config to use for listing the checkpoints.

        Yields:
            AsyncIterator[CheckpointTuple]: An asynchronous iterator of checkpoint tuples.
        """
        loop = asyncio.get_running_loop()
        iter = await loop.run_in_executor(
            None,
            partial(
                self.list,
                before=before,
                limit=limit,
                filter=filter,
            ),
            config,
        )
        while True:
            # handling StopIteration exception inside coroutine won't work
            # as expected, so using next() with default value to break the loop
            if item := await loop.run_in_executor(None, next, iter, None):
                yield item
            else:
                break

    async def aput(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Asynchronous version of put.

        Args:
            config (RunnableConfig): The config to associate with the checkpoint.
            checkpoint (Checkpoint): The checkpoint to save.
            metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
            new_versions (dict): New versions as of this write

        Returns:
            RunnableConfig: The updated config containing the saved checkpoint's timestamp.
        """
        return await asyncio.get_running_loop().run_in_executor(
            None, self.put, config, checkpoint, metadata, new_versions
        )

    async def aput_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[Tuple[str, Any]],
        task_id: str,
    ) -> None:
        """Asynchronous version of put_writes.

        This method is an asynchronous wrapper around put_writes that runs the synchronous
        method in a separate thread using asyncio.

        Args:
            config (RunnableConfig): The config to associate with the writes.
            writes (List[Tuple[str, Any]]): The writes to save, each as a (channel, value) pair.
            task_id (str): Identifier for the task creating the writes.
        """
        return await asyncio.get_running_loop().run_in_executor(
            None, self.put_writes, config, writes, task_id
        )

    def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
        if current is None:
            current_v = 0
        elif isinstance(current, int):
            current_v = current
        else:
            current_v = int(current.split(".")[0])
        next_v = current_v + 1
        next_h = random.random()
        return f"{next_v:032}.{next_h:016}"

config_specs: list[ConfigurableFieldSpec] 属性

定义检查点保存器的配置选项。

返回值

  • list[ConfigurableFieldSpec]

    list[ConfigurableFieldSpec]: 配置字段规范的列表。

get(config: RunnableConfig) -> Optional[Checkpoint]

使用给定的配置获取检查点。

参数

  • config (RunnableConfig) –

    指定要检索的检查点的配置。

返回值

  • Optional[Checkpoint]

    Optional[Checkpoint]: 请求的检查点,如果未找到则为 None。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
def get(self, config: RunnableConfig) -> Optional[Checkpoint]:
    """Fetch a checkpoint using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[Checkpoint]: The requested checkpoint, or None if not found.
    """
    if value := self.get_tuple(config):
        return value.checkpoint

aget(config: RunnableConfig) -> Optional[Checkpoint] 异步

使用给定配置异步获取检查点。

参数

  • config (RunnableConfig) –

    指定要检索的检查点的配置。

返回值

  • Optional[Checkpoint]

    Optional[Checkpoint]: 请求的检查点,如果未找到则为 None。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
async def aget(self, config: RunnableConfig) -> Optional[Checkpoint]:
    """Asynchronously fetch a checkpoint using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[Checkpoint]: The requested checkpoint, or None if not found.
    """
    if value := await self.aget_tuple(config):
        return value.checkpoint

get_tuple(config: RunnableConfig) -> Optional[CheckpointTuple]

从内存存储中获取检查点元组。

此方法根据提供的配置从内存存储中检索检查点元组。如果配置包含“checkpoint_id”键,则检索与匹配的线程 ID 和时间戳对应的检查点。否则,将检索给定线程 ID 的最新检查点。

参数

  • config (RunnableConfig) –

    用于检索检查点的配置。

返回值

  • Optional[CheckpointTuple]

    Optional[CheckpointTuple]: 检索到的检查点元组,如果未找到匹配的检查点,则为 None。

源代码位于 libs/checkpoint/langgraph/checkpoint/memory/__init__.py
def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Get a checkpoint tuple from the in-memory storage.

    This method retrieves a checkpoint tuple from the in-memory storage based on the
    provided config. If the config contains a "checkpoint_id" key, the checkpoint with
    the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint
    for the given thread ID is retrieved.

    Args:
        config (RunnableConfig): The config to use for retrieving the checkpoint.

    Returns:
        Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
    """
    thread_id = config["configurable"]["thread_id"]
    checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
    if checkpoint_id := get_checkpoint_id(config):
        if saved := self.storage[thread_id][checkpoint_ns].get(checkpoint_id):
            checkpoint, metadata, parent_checkpoint_id = saved
            writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
            if parent_checkpoint_id:
                sends = [
                    w[2]
                    for w in self.writes[
                        (thread_id, checkpoint_ns, parent_checkpoint_id)
                    ].values()
                    if w[1] == TASKS
                ]
            else:
                sends = []
            return CheckpointTuple(
                config=config,
                checkpoint={
                    **self.serde.loads_typed(checkpoint),
                    "pending_sends": [self.serde.loads_typed(s) for s in sends],
                },
                metadata=self.serde.loads_typed(metadata),
                pending_writes=[
                    (id, c, self.serde.loads_typed(v)) for id, c, v in writes
                ],
                parent_config={
                    "configurable": {
                        "thread_id": thread_id,
                        "checkpoint_ns": checkpoint_ns,
                        "checkpoint_id": parent_checkpoint_id,
                    }
                }
                if parent_checkpoint_id
                else None,
            )
    else:
        if checkpoints := self.storage[thread_id][checkpoint_ns]:
            checkpoint_id = max(checkpoints.keys())
            checkpoint, metadata, parent_checkpoint_id = checkpoints[checkpoint_id]
            writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
            if parent_checkpoint_id:
                sends = [
                    w[2]
                    for w in self.writes[
                        (thread_id, checkpoint_ns, parent_checkpoint_id)
                    ].values()
                    if w[1] == TASKS
                ]
            else:
                sends = []
            return CheckpointTuple(
                config={
                    "configurable": {
                        "thread_id": thread_id,
                        "checkpoint_ns": checkpoint_ns,
                        "checkpoint_id": checkpoint_id,
                    }
                },
                checkpoint={
                    **self.serde.loads_typed(checkpoint),
                    "pending_sends": [self.serde.loads_typed(s) for s in sends],
                },
                metadata=self.serde.loads_typed(metadata),
                pending_writes=[
                    (id, c, self.serde.loads_typed(v)) for id, c, v in writes
                ],
                parent_config={
                    "configurable": {
                        "thread_id": thread_id,
                        "checkpoint_ns": checkpoint_ns,
                        "checkpoint_id": parent_checkpoint_id,
                    }
                }
                if parent_checkpoint_id
                else None,
            )

list(config: Optional[RunnableConfig], *, filter: Optional[Dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None) -> Iterator[CheckpointTuple]

列出内存存储中的检查点。

此方法根据提供的条件从内存存储中检索检查点元组列表。

参数

  • config (Optional[RunnableConfig]) –

    用于过滤检查点的基本配置。

  • filter (Optional[Dict[str, Any]], 默认: None ) –

    用于元数据的其他过滤条件。

  • before (Optional[RunnableConfig], 默认: None ) –

    列出在此配置之前创建的检查点。

  • limit (Optional[int], 默认: None ) –

    要返回的最大检查点数。

生成

  • CheckpointTuple

    Iterator[CheckpointTuple]: 匹配检查点元组的迭代器。

源代码位于 libs/checkpoint/langgraph/checkpoint/memory/__init__.py
def list(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> Iterator[CheckpointTuple]:
    """List checkpoints from the in-memory storage.

    This method retrieves a list of checkpoint tuples from the in-memory storage based
    on the provided criteria.

    Args:
        config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
        filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
        before (Optional[RunnableConfig]): List checkpoints created before this configuration.
        limit (Optional[int]): Maximum number of checkpoints to return.

    Yields:
        Iterator[CheckpointTuple]: An iterator of matching checkpoint tuples.
    """
    thread_ids = (config["configurable"]["thread_id"],) if config else self.storage
    config_checkpoint_ns = (
        config["configurable"].get("checkpoint_ns") if config else None
    )
    config_checkpoint_id = get_checkpoint_id(config) if config else None
    for thread_id in thread_ids:
        for checkpoint_ns in self.storage[thread_id].keys():
            if (
                config_checkpoint_ns is not None
                and checkpoint_ns != config_checkpoint_ns
            ):
                continue

            for checkpoint_id, (
                checkpoint,
                metadata_b,
                parent_checkpoint_id,
            ) in sorted(
                self.storage[thread_id][checkpoint_ns].items(),
                key=lambda x: x[0],
                reverse=True,
            ):
                # filter by checkpoint ID from config
                if config_checkpoint_id and checkpoint_id != config_checkpoint_id:
                    continue

                # filter by checkpoint ID from `before` config
                if (
                    before
                    and (before_checkpoint_id := get_checkpoint_id(before))
                    and checkpoint_id >= before_checkpoint_id
                ):
                    continue

                # filter by metadata
                metadata = self.serde.loads_typed(metadata_b)
                if filter and not all(
                    query_value == metadata.get(query_key)
                    for query_key, query_value in filter.items()
                ):
                    continue

                # limit search results
                if limit is not None and limit <= 0:
                    break
                elif limit is not None:
                    limit -= 1

                writes = self.writes[
                    (thread_id, checkpoint_ns, checkpoint_id)
                ].values()

                if parent_checkpoint_id:
                    sends = [
                        w[2]
                        for w in self.writes[
                            (thread_id, checkpoint_ns, parent_checkpoint_id)
                        ].values()
                        if w[1] == TASKS
                    ]
                else:
                    sends = []

                yield CheckpointTuple(
                    config={
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": checkpoint_id,
                        }
                    },
                    checkpoint={
                        **self.serde.loads_typed(checkpoint),
                        "pending_sends": [self.serde.loads_typed(s) for s in sends],
                    },
                    metadata=metadata,
                    parent_config={
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": parent_checkpoint_id,
                        }
                    }
                    if parent_checkpoint_id
                    else None,
                    pending_writes=[
                        (id, c, self.serde.loads_typed(v)) for id, c, v in writes
                    ],
                )

put(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions) -> RunnableConfig

将检查点保存到内存存储中。

此方法将检查点保存到内存存储中。检查点与提供的配置相关联。

参数

  • config (RunnableConfig) –

    与检查点关联的配置。

  • checkpoint (Checkpoint) –

    要保存的检查点。

  • metadata (CheckpointMetadata) –

    与检查点一起保存的附加元数据。

  • new_versions (dict) –

    本次写入的新版本

返回值

  • RunnableConfig ( RunnableConfig ) –

    包含保存的检查点时间戳的更新配置。

源代码位于 libs/checkpoint/langgraph/checkpoint/memory/__init__.py
def put(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Save a checkpoint to the in-memory storage.

    This method saves a checkpoint to the in-memory storage. The checkpoint is associated
    with the provided config.

    Args:
        config (RunnableConfig): The config to associate with the checkpoint.
        checkpoint (Checkpoint): The checkpoint to save.
        metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
        new_versions (dict): New versions as of this write

    Returns:
        RunnableConfig: The updated config containing the saved checkpoint's timestamp.
    """
    c = checkpoint.copy()
    c.pop("pending_sends")  # type: ignore[misc]
    thread_id = config["configurable"]["thread_id"]
    checkpoint_ns = config["configurable"]["checkpoint_ns"]
    self.storage[thread_id][checkpoint_ns].update(
        {
            checkpoint["id"]: (
                self.serde.dumps_typed(c),
                self.serde.dumps_typed(metadata),
                config["configurable"].get("checkpoint_id"),  # parent
            )
        }
    )
    return {
        "configurable": {
            "thread_id": thread_id,
            "checkpoint_ns": checkpoint_ns,
            "checkpoint_id": checkpoint["id"],
        }
    }

put_writes(config: RunnableConfig, writes: Sequence[Tuple[str, Any]], task_id: str) -> None

将写入列表保存到内存存储中。

此方法将写入列表保存到内存存储中。写入与提供的配置相关联。

参数

  • config (RunnableConfig) –

    与写入关联的配置。

  • writes (list[tuple[str, Any]]) –

    要保存的写入内容。

  • task_id (str) –

    创建写入内容的任务标识符。

返回值

  • RunnableConfig ( None ) –

    包含已保存写入内容时间戳的更新配置。

源代码位于 libs/checkpoint/langgraph/checkpoint/memory/__init__.py
def put_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[Tuple[str, Any]],
    task_id: str,
) -> None:
    """Save a list of writes to the in-memory storage.

    This method saves a list of writes to the in-memory storage. The writes are associated
    with the provided config.

    Args:
        config (RunnableConfig): The config to associate with the writes.
        writes (list[tuple[str, Any]]): The writes to save.
        task_id (str): Identifier for the task creating the writes.

    Returns:
        RunnableConfig: The updated config containing the saved writes' timestamp.
    """
    thread_id = config["configurable"]["thread_id"]
    checkpoint_ns = config["configurable"]["checkpoint_ns"]
    checkpoint_id = config["configurable"]["checkpoint_id"]
    outer_key = (thread_id, checkpoint_ns, checkpoint_id)
    for idx, (c, v) in enumerate(writes):
        inner_key = (task_id, WRITES_IDX_MAP.get(c, idx))
        self.writes[outer_key][inner_key] = (task_id, c, self.serde.dumps_typed(v))

aget_tuple(config: RunnableConfig) -> Optional[CheckpointTuple] 异步

get_tuple 的异步版本。

此方法是 get_tuple 的异步包装器,它使用 asyncio 在单独的线程中运行同步方法。

参数

  • config (RunnableConfig) –

    用于检索检查点的配置。

返回值

  • Optional[CheckpointTuple]

    Optional[CheckpointTuple]: 检索到的检查点元组,如果未找到匹配的检查点,则为 None。

源代码位于 libs/checkpoint/langgraph/checkpoint/memory/__init__.py
async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Asynchronous version of get_tuple.

    This method is an asynchronous wrapper around get_tuple that runs the synchronous
    method in a separate thread using asyncio.

    Args:
        config (RunnableConfig): The config to use for retrieving the checkpoint.

    Returns:
        Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
    """
    return await asyncio.get_running_loop().run_in_executor(
        None, self.get_tuple, config
    )

alist(config: Optional[RunnableConfig], *, filter: Optional[Dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None) -> AsyncIterator[CheckpointTuple] 异步

list 的异步版本。

此方法是 list 的异步包装器,它使用 asyncio 在单独的线程中运行同步方法。

参数

  • config (RunnableConfig) –

    用于列出检查点的配置。

生成

  • AsyncIterator[CheckpointTuple]

    AsyncIterator[CheckpointTuple]: 检查点元组的异步迭代器。

源代码位于 libs/checkpoint/langgraph/checkpoint/memory/__init__.py
async def alist(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> AsyncIterator[CheckpointTuple]:
    """Asynchronous version of list.

    This method is an asynchronous wrapper around list that runs the synchronous
    method in a separate thread using asyncio.

    Args:
        config (RunnableConfig): The config to use for listing the checkpoints.

    Yields:
        AsyncIterator[CheckpointTuple]: An asynchronous iterator of checkpoint tuples.
    """
    loop = asyncio.get_running_loop()
    iter = await loop.run_in_executor(
        None,
        partial(
            self.list,
            before=before,
            limit=limit,
            filter=filter,
        ),
        config,
    )
    while True:
        # handling StopIteration exception inside coroutine won't work
        # as expected, so using next() with default value to break the loop
        if item := await loop.run_in_executor(None, next, iter, None):
            yield item
        else:
            break

aput(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions) -> RunnableConfig 异步

put 的异步版本。

参数

  • config (RunnableConfig) –

    与检查点关联的配置。

  • checkpoint (Checkpoint) –

    要保存的检查点。

  • metadata (CheckpointMetadata) –

    与检查点一起保存的附加元数据。

  • new_versions (dict) –

    本次写入的新版本

返回值

  • RunnableConfig ( RunnableConfig ) –

    包含保存的检查点时间戳的更新配置。

源代码位于 libs/checkpoint/langgraph/checkpoint/memory/__init__.py
async def aput(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Asynchronous version of put.

    Args:
        config (RunnableConfig): The config to associate with the checkpoint.
        checkpoint (Checkpoint): The checkpoint to save.
        metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
        new_versions (dict): New versions as of this write

    Returns:
        RunnableConfig: The updated config containing the saved checkpoint's timestamp.
    """
    return await asyncio.get_running_loop().run_in_executor(
        None, self.put, config, checkpoint, metadata, new_versions
    )

aput_writes(config: RunnableConfig, writes: Sequence[Tuple[str, Any]], task_id: str) -> None 异步

put_writes 的异步版本。

此方法是 put_writes 的异步包装器,它使用 asyncio 在单独的线程中运行同步方法。

参数

  • config (RunnableConfig) –

    与写入关联的配置。

  • writes (List[Tuple[str, Any]]) –

    要保存的写入内容,每个写入内容都是一个 (channel, value) 对。

  • task_id (str) –

    创建写入内容的任务标识符。

源代码位于 libs/checkpoint/langgraph/checkpoint/memory/__init__.py
async def aput_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[Tuple[str, Any]],
    task_id: str,
) -> None:
    """Asynchronous version of put_writes.

    This method is an asynchronous wrapper around put_writes that runs the synchronous
    method in a separate thread using asyncio.

    Args:
        config (RunnableConfig): The config to associate with the writes.
        writes (List[Tuple[str, Any]]): The writes to save, each as a (channel, value) pair.
        task_id (str): Identifier for the task creating the writes.
    """
    return await asyncio.get_running_loop().run_in_executor(
        None, self.put_writes, config, writes, task_id
    )

SqliteSaver

基类: BaseCheckpointSaver[str]

将检查点存储在 SQLite 数据库中的检查点保存器。

注意

此类适用于轻量级、同步用例(演示和小型项目),并且无法扩展到多个线程。对于具有 async 支持的类似 sqlite 保存器,请考虑使用 AsyncSqliteSaver

参数

  • conn (Connection) –

    SQLite 数据库连接。

  • serde (Optional[SerializerProtocol], 默认: None ) –

    用于序列化和反序列化检查点的序列化程序。默认为 JsonPlusSerializerCompat。

示例

>>> import sqlite3
>>> from langgraph.checkpoint.sqlite import SqliteSaver
>>> from langgraph.graph import StateGraph
>>>
>>> builder = StateGraph(int)
>>> builder.add_node("add_one", lambda x: x + 1)
>>> builder.set_entry_point("add_one")
>>> builder.set_finish_point("add_one")
>>> conn = sqlite3.connect("checkpoints.sqlite")
>>> memory = SqliteSaver(conn)
>>> graph = builder.compile(checkpointer=memory)
>>> config = {"configurable": {"thread_id": "1"}}
>>> graph.get_state(config)
>>> result = graph.invoke(3, config)
>>> graph.get_state(config)
StateSnapshot(values=4, next=(), config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '0c62ca34-ac19-445d-bbb0-5b4984975b2a'}}, parent_config=None)
源代码位于 libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
class SqliteSaver(BaseCheckpointSaver[str]):
    """A checkpoint saver that stores checkpoints in a SQLite database.

    Note:
        This class is meant for lightweight, synchronous use cases
        (demos and small projects) and does not
        scale to multiple threads.
        For a similar sqlite saver with `async` support,
        consider using [AsyncSqliteSaver][langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver].

    Args:
        conn (sqlite3.Connection): The SQLite database connection.
        serde (Optional[SerializerProtocol]): The serializer to use for serializing and deserializing checkpoints. Defaults to JsonPlusSerializerCompat.

    Examples:

        >>> import sqlite3
        >>> from langgraph.checkpoint.sqlite import SqliteSaver
        >>> from langgraph.graph import StateGraph
        >>>
        >>> builder = StateGraph(int)
        >>> builder.add_node("add_one", lambda x: x + 1)
        >>> builder.set_entry_point("add_one")
        >>> builder.set_finish_point("add_one")
        >>> conn = sqlite3.connect("checkpoints.sqlite")
        >>> memory = SqliteSaver(conn)
        >>> graph = builder.compile(checkpointer=memory)
        >>> config = {"configurable": {"thread_id": "1"}}
        >>> graph.get_state(config)
        >>> result = graph.invoke(3, config)
        >>> graph.get_state(config)
        StateSnapshot(values=4, next=(), config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '0c62ca34-ac19-445d-bbb0-5b4984975b2a'}}, parent_config=None)
    """  # noqa

    conn: sqlite3.Connection
    is_setup: bool

    def __init__(
        self,
        conn: sqlite3.Connection,
        *,
        serde: Optional[SerializerProtocol] = None,
    ) -> None:
        super().__init__(serde=serde)
        self.jsonplus_serde = JsonPlusSerializer()
        self.conn = conn
        self.is_setup = False
        self.lock = threading.Lock()

    @classmethod
    @contextmanager
    def from_conn_string(cls, conn_string: str) -> Iterator["SqliteSaver"]:
        """Create a new SqliteSaver instance from a connection string.

        Args:
            conn_string (str): The SQLite connection string.

        Yields:
            SqliteSaver: A new SqliteSaver instance.

        Examples:

            In memory:

                with SqliteSaver.from_conn_string(":memory:") as memory:
                    ...

            To disk:

                with SqliteSaver.from_conn_string("checkpoints.sqlite") as memory:
                    ...
        """
        with closing(
            sqlite3.connect(
                conn_string,
                # https://ricardoanderegg.com/posts/python-sqlite-thread-safety/
                check_same_thread=False,
            )
        ) as conn:
            yield SqliteSaver(conn)

    def setup(self) -> None:
        """Set up the checkpoint database.

        This method creates the necessary tables in the SQLite database if they don't
        already exist. It is called automatically when needed and should not be called
        directly by the user.
        """
        if self.is_setup:
            return

        self.conn.executescript(
            """
            PRAGMA journal_mode=WAL;
            CREATE TABLE IF NOT EXISTS checkpoints (
                thread_id TEXT NOT NULL,
                checkpoint_ns TEXT NOT NULL DEFAULT '',
                checkpoint_id TEXT NOT NULL,
                parent_checkpoint_id TEXT,
                type TEXT,
                checkpoint BLOB,
                metadata BLOB,
                PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
            );
            CREATE TABLE IF NOT EXISTS writes (
                thread_id TEXT NOT NULL,
                checkpoint_ns TEXT NOT NULL DEFAULT '',
                checkpoint_id TEXT NOT NULL,
                task_id TEXT NOT NULL,
                idx INTEGER NOT NULL,
                channel TEXT NOT NULL,
                type TEXT,
                value BLOB,
                PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
            );
            """
        )

        self.is_setup = True

    @contextmanager
    def cursor(self, transaction: bool = True) -> Iterator[sqlite3.Cursor]:
        """Get a cursor for the SQLite database.

        This method returns a cursor for the SQLite database. It is used internally
        by the SqliteSaver and should not be called directly by the user.

        Args:
            transaction (bool): Whether to commit the transaction when the cursor is closed. Defaults to True.

        Yields:
            sqlite3.Cursor: A cursor for the SQLite database.
        """
        with self.lock:
            self.setup()
            cur = self.conn.cursor()
            try:
                yield cur
            finally:
                if transaction:
                    self.conn.commit()
                cur.close()

    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Get a checkpoint tuple from the database.

        This method retrieves a checkpoint tuple from the SQLite database based on the
        provided config. If the config contains a "checkpoint_id" key, the checkpoint with
        the matching thread ID and checkpoint ID is retrieved. Otherwise, the latest checkpoint
        for the given thread ID is retrieved.

        Args:
            config (RunnableConfig): The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

        Examples:

            Basic:
            >>> config = {"configurable": {"thread_id": "1"}}
            >>> checkpoint_tuple = memory.get_tuple(config)
            >>> print(checkpoint_tuple)
            CheckpointTuple(...)

            With checkpoint ID:

            >>> config = {
            ...    "configurable": {
            ...        "thread_id": "1",
            ...        "checkpoint_ns": "",
            ...        "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875",
            ...    }
            ... }
            >>> checkpoint_tuple = memory.get_tuple(config)
            >>> print(checkpoint_tuple)
            CheckpointTuple(...)
        """  # noqa
        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
        with self.cursor(transaction=False) as cur:
            # find the latest checkpoint for the thread_id
            if checkpoint_id := get_checkpoint_id(config):
                cur.execute(
                    "SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?",
                    (
                        str(config["configurable"]["thread_id"]),
                        checkpoint_ns,
                        checkpoint_id,
                    ),
                )
            else:
                cur.execute(
                    "SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? ORDER BY checkpoint_id DESC LIMIT 1",
                    (str(config["configurable"]["thread_id"]), checkpoint_ns),
                )
            # if a checkpoint is found, return it
            if value := cur.fetchone():
                (
                    thread_id,
                    checkpoint_id,
                    parent_checkpoint_id,
                    type,
                    checkpoint,
                    metadata,
                ) = value
                if not get_checkpoint_id(config):
                    config = {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": checkpoint_id,
                        }
                    }
                # find any pending writes
                cur.execute(
                    "SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
                    (
                        str(config["configurable"]["thread_id"]),
                        checkpoint_ns,
                        str(config["configurable"]["checkpoint_id"]),
                    ),
                )
                # deserialize the checkpoint and metadata
                return CheckpointTuple(
                    config,
                    self.serde.loads_typed((type, checkpoint)),
                    self.jsonplus_serde.loads(metadata) if metadata is not None else {},
                    (
                        {
                            "configurable": {
                                "thread_id": thread_id,
                                "checkpoint_ns": checkpoint_ns,
                                "checkpoint_id": parent_checkpoint_id,
                            }
                        }
                        if parent_checkpoint_id
                        else None
                    ),
                    [
                        (task_id, channel, self.serde.loads_typed((type, value)))
                        for task_id, channel, type, value in cur
                    ],
                )

    def list(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[Dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> Iterator[CheckpointTuple]:
        """List checkpoints from the database.

        This method retrieves a list of checkpoint tuples from the SQLite database based
        on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

        Args:
            config (RunnableConfig): The config to use for listing the checkpoints.
            filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata. Defaults to None.
            before (Optional[RunnableConfig]): If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.
            limit (Optional[int]): The maximum number of checkpoints to return. Defaults to None.

        Yields:
            Iterator[CheckpointTuple]: An iterator of checkpoint tuples.

        Examples:
            >>> from langgraph.checkpoint.sqlite import SqliteSaver
            >>> with SqliteSaver.from_conn_string(":memory:") as memory:
            ... # Run a graph, then list the checkpoints
            >>>     config = {"configurable": {"thread_id": "1"}}
            >>>     checkpoints = list(memory.list(config, limit=2))
            >>> print(checkpoints)
            [CheckpointTuple(...), CheckpointTuple(...)]

            >>> config = {"configurable": {"thread_id": "1"}}
            >>> before = {"configurable": {"checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875"}}
            >>> with SqliteSaver.from_conn_string(":memory:") as memory:
            ... # Run a graph, then list the checkpoints
            >>>     checkpoints = list(memory.list(config, before=before))
            >>> print(checkpoints)
            [CheckpointTuple(...), ...]
        """
        where, param_values = search_where(config, filter, before)
        query = f"""SELECT thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata
        FROM checkpoints
        {where}
        ORDER BY checkpoint_id DESC"""
        if limit:
            query += f" LIMIT {limit}"
        with self.cursor(transaction=False) as cur, closing(self.conn.cursor()) as wcur:
            cur.execute(query, param_values)
            for (
                thread_id,
                checkpoint_ns,
                checkpoint_id,
                parent_checkpoint_id,
                type,
                checkpoint,
                metadata,
            ) in cur:
                wcur.execute(
                    "SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
                    (thread_id, checkpoint_ns, checkpoint_id),
                )
                yield CheckpointTuple(
                    {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": checkpoint_id,
                        }
                    },
                    self.serde.loads_typed((type, checkpoint)),
                    self.jsonplus_serde.loads(metadata) if metadata is not None else {},
                    (
                        {
                            "configurable": {
                                "thread_id": thread_id,
                                "checkpoint_ns": checkpoint_ns,
                                "checkpoint_id": parent_checkpoint_id,
                            }
                        }
                        if parent_checkpoint_id
                        else None
                    ),
                    [
                        (task_id, channel, self.serde.loads_typed((type, value)))
                        for task_id, channel, type, value in wcur
                    ],
                )

    def put(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Save a checkpoint to the database.

        This method saves a checkpoint to the SQLite database. The checkpoint is associated
        with the provided config and its parent config (if any).

        Args:
            config (RunnableConfig): The config to associate with the checkpoint.
            checkpoint (Checkpoint): The checkpoint to save.
            metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
            new_versions (ChannelVersions): New channel versions as of this write.

        Returns:
            RunnableConfig: Updated configuration after storing the checkpoint.

        Examples:

            >>> from langgraph.checkpoint.sqlite import SqliteSaver
            >>> with SqliteSaver.from_conn_string(":memory:") as memory:
            >>>     config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
            >>>     checkpoint = {"ts": "2024-05-04T06:32:42.235444+00:00", "id": "1ef4f797-8335-6428-8001-8a1503f9b875", "data": {"key": "value"}}
            >>>     saved_config = memory.put(config, checkpoint, {"source": "input", "step": 1, "writes": {"key": "value"}}, {})
            >>> print(saved_config)
            {'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef4f797-8335-6428-8001-8a1503f9b875'}}
        """
        thread_id = config["configurable"]["thread_id"]
        checkpoint_ns = config["configurable"]["checkpoint_ns"]
        type_, serialized_checkpoint = self.serde.dumps_typed(checkpoint)
        serialized_metadata = self.jsonplus_serde.dumps(metadata)
        with self.cursor() as cur:
            cur.execute(
                "INSERT OR REPLACE INTO checkpoints (thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata) VALUES (?, ?, ?, ?, ?, ?, ?)",
                (
                    str(config["configurable"]["thread_id"]),
                    checkpoint_ns,
                    checkpoint["id"],
                    config["configurable"].get("checkpoint_id"),
                    type_,
                    serialized_checkpoint,
                    serialized_metadata,
                ),
            )
        return {
            "configurable": {
                "thread_id": thread_id,
                "checkpoint_ns": checkpoint_ns,
                "checkpoint_id": checkpoint["id"],
            }
        }

    def put_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[Tuple[str, Any]],
        task_id: str,
    ) -> None:
        """Store intermediate writes linked to a checkpoint.

        This method saves intermediate writes associated with a checkpoint to the SQLite database.

        Args:
            config (RunnableConfig): Configuration of the related checkpoint.
            writes (Sequence[Tuple[str, Any]]): List of writes to store, each as (channel, value) pair.
            task_id (str): Identifier for the task creating the writes.
        """
        query = (
            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
            if all(w[0] in WRITES_IDX_MAP for w in writes)
            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
        )
        with self.cursor() as cur:
            cur.executemany(
                query,
                [
                    (
                        str(config["configurable"]["thread_id"]),
                        str(config["configurable"]["checkpoint_ns"]),
                        str(config["configurable"]["checkpoint_id"]),
                        task_id,
                        WRITES_IDX_MAP.get(channel, idx),
                        channel,
                        *self.serde.dumps_typed(value),
                    )
                    for idx, (channel, value) in enumerate(writes)
                ],
            )

    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Get a checkpoint tuple from the database asynchronously.

        Note:
            This async method is not supported by the SqliteSaver class.
            Use get_tuple() instead, or consider using [AsyncSqliteSaver][langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver].
        """
        raise NotImplementedError(_AIO_ERROR_MSG)

    async def alist(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[Dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> AsyncIterator[CheckpointTuple]:
        """List checkpoints from the database asynchronously.

        Note:
            This async method is not supported by the SqliteSaver class.
            Use list() instead, or consider using [AsyncSqliteSaver][langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver].
        """
        raise NotImplementedError(_AIO_ERROR_MSG)
        yield

    async def aput(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Save a checkpoint to the database asynchronously.

        Note:
            This async method is not supported by the SqliteSaver class.
            Use put() instead, or consider using [AsyncSqliteSaver][langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver].
        """
        raise NotImplementedError(_AIO_ERROR_MSG)

    def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
        """Generate the next version ID for a channel.

        This method creates a new version identifier for a channel based on its current version.

        Args:
            current (Optional[str]): The current version identifier of the channel.
            channel (BaseChannel): The channel being versioned.

        Returns:
            str: The next version identifier, which is guaranteed to be monotonically increasing.
        """
        if current is None:
            current_v = 0
        elif isinstance(current, int):
            current_v = current
        else:
            current_v = int(current.split(".")[0])
        next_v = current_v + 1
        next_h = random.random()
        return f"{next_v:032}.{next_h:016}"

config_specs: list[ConfigurableFieldSpec] 属性

定义检查点保存器的配置选项。

返回值

  • list[ConfigurableFieldSpec]

    list[ConfigurableFieldSpec]: 配置字段规范的列表。

get(config: RunnableConfig) -> Optional[Checkpoint]

使用给定的配置获取检查点。

参数

  • config (RunnableConfig) –

    指定要检索的检查点的配置。

返回值

  • Optional[Checkpoint]

    Optional[Checkpoint]: 请求的检查点,如果未找到则为 None。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
def get(self, config: RunnableConfig) -> Optional[Checkpoint]:
    """Fetch a checkpoint using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[Checkpoint]: The requested checkpoint, or None if not found.
    """
    if value := self.get_tuple(config):
        return value.checkpoint

aget(config: RunnableConfig) -> Optional[Checkpoint] 异步

使用给定配置异步获取检查点。

参数

  • config (RunnableConfig) –

    指定要检索的检查点的配置。

返回值

  • Optional[Checkpoint]

    Optional[Checkpoint]: 请求的检查点,如果未找到则为 None。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
async def aget(self, config: RunnableConfig) -> Optional[Checkpoint]:
    """Asynchronously fetch a checkpoint using the given configuration.

    Args:
        config (RunnableConfig): Configuration specifying which checkpoint to retrieve.

    Returns:
        Optional[Checkpoint]: The requested checkpoint, or None if not found.
    """
    if value := await self.aget_tuple(config):
        return value.checkpoint

aput_writes(config: RunnableConfig, writes: Sequence[Tuple[str, Any]], task_id: str) -> None async

异步存储与检查点链接的中间写入内容。

参数

  • config (RunnableConfig) –

    相关检查点的配置。

  • writes (List[Tuple[str, Any]]) –

    要存储的写入内容列表。

  • task_id (str) –

    创建写入内容的任务标识符。

引发

  • NotImplementedError

    在您的自定义检查点保存器中实现此方法。

源代码在 libs/checkpoint/langgraph/checkpoint/base/__init__.py
async def aput_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[Tuple[str, Any]],
    task_id: str,
) -> None:
    """Asynchronously store intermediate writes linked to a checkpoint.

    Args:
        config (RunnableConfig): Configuration of the related checkpoint.
        writes (List[Tuple[str, Any]]): List of writes to store.
        task_id (str): Identifier for the task creating the writes.

    Raises:
        NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

from_conn_string(conn_string: str) -> Iterator[SqliteSaver] classmethod

从连接字符串创建一个新的 SqliteSaver 实例。

参数

  • conn_string (str) –

    SQLite 连接字符串。

生成

  • SqliteSaver ( SqliteSaver ) –

    一个新的 SqliteSaver 实例。

示例

In memory:

    with SqliteSaver.from_conn_string(":memory:") as memory:
        ...

To disk:

    with SqliteSaver.from_conn_string("checkpoints.sqlite") as memory:
        ...
源代码位于 libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py
@classmethod
@contextmanager
def from_conn_string(cls, conn_string: str) -> Iterator["SqliteSaver"]:
    """Create a new SqliteSaver instance from a connection string.

    Args:
        conn_string (str): The SQLite connection string.

    Yields:
        SqliteSaver: A new SqliteSaver instance.

    Examples:

        In memory:

            with SqliteSaver.from_conn_string(":memory:") as memory:
                ...

        To disk:

            with SqliteSaver.from_conn_string("checkpoints.sqlite") as memory:
                ...
    """
    with closing(
        sqlite3.connect(
            conn_string,
            # https://ricardoanderegg.com/posts/python-sqlite-thread-safety/
            check_same_thread=False,
        )
    ) as conn:
        yield SqliteSaver(conn)

setup() -> None

设置检查点数据库。

此方法在 SQLite 数据库中创建必要的表,如果它们不存在。它在需要时自动调用,用户不应该直接调用。

源代码位于 libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py
def setup(self) -> None:
    """Set up the checkpoint database.

    This method creates the necessary tables in the SQLite database if they don't
    already exist. It is called automatically when needed and should not be called
    directly by the user.
    """
    if self.is_setup:
        return

    self.conn.executescript(
        """
        PRAGMA journal_mode=WAL;
        CREATE TABLE IF NOT EXISTS checkpoints (
            thread_id TEXT NOT NULL,
            checkpoint_ns TEXT NOT NULL DEFAULT '',
            checkpoint_id TEXT NOT NULL,
            parent_checkpoint_id TEXT,
            type TEXT,
            checkpoint BLOB,
            metadata BLOB,
            PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
        );
        CREATE TABLE IF NOT EXISTS writes (
            thread_id TEXT NOT NULL,
            checkpoint_ns TEXT NOT NULL DEFAULT '',
            checkpoint_id TEXT NOT NULL,
            task_id TEXT NOT NULL,
            idx INTEGER NOT NULL,
            channel TEXT NOT NULL,
            type TEXT,
            value BLOB,
            PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
        );
        """
    )

    self.is_setup = True

cursor(transaction: bool = True) -> Iterator[sqlite3.Cursor]

获取 SQLite 数据库的游标。

此方法返回 SQLite 数据库的游标。它由 SqliteSaver 内部使用,用户不应该直接调用。

参数

  • transaction (bool, default: True ) –

    是否在游标关闭时提交事务。默认为 True。

生成

  • Cursor

    sqlite3.Cursor: SQLite 数据库的游标。

源代码位于 libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py
@contextmanager
def cursor(self, transaction: bool = True) -> Iterator[sqlite3.Cursor]:
    """Get a cursor for the SQLite database.

    This method returns a cursor for the SQLite database. It is used internally
    by the SqliteSaver and should not be called directly by the user.

    Args:
        transaction (bool): Whether to commit the transaction when the cursor is closed. Defaults to True.

    Yields:
        sqlite3.Cursor: A cursor for the SQLite database.
    """
    with self.lock:
        self.setup()
        cur = self.conn.cursor()
        try:
            yield cur
        finally:
            if transaction:
                self.conn.commit()
            cur.close()

get_tuple(config: RunnableConfig) -> Optional[CheckpointTuple]

从数据库中获取检查点元组。

此方法根据提供的配置从 SQLite 数据库中检索检查点元组。如果配置包含“checkpoint_id”键,则检索与匹配线程 ID 和检查点 ID 相匹配的检查点。否则,检索给定线程 ID 的最新检查点。

参数

  • config (RunnableConfig) –

    用于检索检查点的配置。

返回值

  • Optional[CheckpointTuple]

    Optional[CheckpointTuple]: 检索到的检查点元组,如果未找到匹配的检查点,则为 None。

示例

Basic:
>>> config = {"configurable": {"thread_id": "1"}}
>>> checkpoint_tuple = memory.get_tuple(config)
>>> print(checkpoint_tuple)
CheckpointTuple(...)

With checkpoint ID:

>>> config = {
...    "configurable": {
...        "thread_id": "1",
...        "checkpoint_ns": "",
...        "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875",
...    }
... }
>>> checkpoint_tuple = memory.get_tuple(config)
>>> print(checkpoint_tuple)
CheckpointTuple(...)
源代码位于 libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py
def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Get a checkpoint tuple from the database.

    This method retrieves a checkpoint tuple from the SQLite database based on the
    provided config. If the config contains a "checkpoint_id" key, the checkpoint with
    the matching thread ID and checkpoint ID is retrieved. Otherwise, the latest checkpoint
    for the given thread ID is retrieved.

    Args:
        config (RunnableConfig): The config to use for retrieving the checkpoint.

    Returns:
        Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

    Examples:

        Basic:
        >>> config = {"configurable": {"thread_id": "1"}}
        >>> checkpoint_tuple = memory.get_tuple(config)
        >>> print(checkpoint_tuple)
        CheckpointTuple(...)

        With checkpoint ID:

        >>> config = {
        ...    "configurable": {
        ...        "thread_id": "1",
        ...        "checkpoint_ns": "",
        ...        "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        ...    }
        ... }
        >>> checkpoint_tuple = memory.get_tuple(config)
        >>> print(checkpoint_tuple)
        CheckpointTuple(...)
    """  # noqa
    checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
    with self.cursor(transaction=False) as cur:
        # find the latest checkpoint for the thread_id
        if checkpoint_id := get_checkpoint_id(config):
            cur.execute(
                "SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?",
                (
                    str(config["configurable"]["thread_id"]),
                    checkpoint_ns,
                    checkpoint_id,
                ),
            )
        else:
            cur.execute(
                "SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? ORDER BY checkpoint_id DESC LIMIT 1",
                (str(config["configurable"]["thread_id"]), checkpoint_ns),
            )
        # if a checkpoint is found, return it
        if value := cur.fetchone():
            (
                thread_id,
                checkpoint_id,
                parent_checkpoint_id,
                type,
                checkpoint,
                metadata,
            ) = value
            if not get_checkpoint_id(config):
                config = {
                    "configurable": {
                        "thread_id": thread_id,
                        "checkpoint_ns": checkpoint_ns,
                        "checkpoint_id": checkpoint_id,
                    }
                }
            # find any pending writes
            cur.execute(
                "SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
                (
                    str(config["configurable"]["thread_id"]),
                    checkpoint_ns,
                    str(config["configurable"]["checkpoint_id"]),
                ),
            )
            # deserialize the checkpoint and metadata
            return CheckpointTuple(
                config,
                self.serde.loads_typed((type, checkpoint)),
                self.jsonplus_serde.loads(metadata) if metadata is not None else {},
                (
                    {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": parent_checkpoint_id,
                        }
                    }
                    if parent_checkpoint_id
                    else None
                ),
                [
                    (task_id, channel, self.serde.loads_typed((type, value)))
                    for task_id, channel, type, value in cur
                ],
            )

list(config: Optional[RunnableConfig], *, filter: Optional[Dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None) -> Iterator[CheckpointTuple]

列出数据库中的检查点。

此方法根据提供的配置从 SQLite 数据库中检索检查点元组列表。检查点按检查点 ID 降序排列(最新的排在最前面)。

参数

  • config (RunnableConfig) –

    用于列出检查点的配置。

  • filter (Optional[Dict[str, Any]], 默认: None ) –

    元数据的附加过滤条件。默认为 None。

  • before (Optional[RunnableConfig], 默认: None ) –

    如果提供,则仅返回指定检查点 ID 之前的检查点。默认为 None。

  • limit (Optional[int], 默认: None ) –

    要返回的检查点数的最大值。默认为 None。

生成

  • CheckpointTuple

    Iterator[CheckpointTuple]: 检查点元组的迭代器。

示例

>>> from langgraph.checkpoint.sqlite import SqliteSaver
>>> with SqliteSaver.from_conn_string(":memory:") as memory:
... # Run a graph, then list the checkpoints
>>>     config = {"configurable": {"thread_id": "1"}}
>>>     checkpoints = list(memory.list(config, limit=2))
>>> print(checkpoints)
[CheckpointTuple(...), CheckpointTuple(...)]
>>> config = {"configurable": {"thread_id": "1"}}
>>> before = {"configurable": {"checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875"}}
>>> with SqliteSaver.from_conn_string(":memory:") as memory:
... # Run a graph, then list the checkpoints
>>>     checkpoints = list(memory.list(config, before=before))
>>> print(checkpoints)
[CheckpointTuple(...), ...]
源代码位于 libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py
def list(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> Iterator[CheckpointTuple]:
    """List checkpoints from the database.

    This method retrieves a list of checkpoint tuples from the SQLite database based
    on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

    Args:
        config (RunnableConfig): The config to use for listing the checkpoints.
        filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata. Defaults to None.
        before (Optional[RunnableConfig]): If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.
        limit (Optional[int]): The maximum number of checkpoints to return. Defaults to None.

    Yields:
        Iterator[CheckpointTuple]: An iterator of checkpoint tuples.

    Examples:
        >>> from langgraph.checkpoint.sqlite import SqliteSaver
        >>> with SqliteSaver.from_conn_string(":memory:") as memory:
        ... # Run a graph, then list the checkpoints
        >>>     config = {"configurable": {"thread_id": "1"}}
        >>>     checkpoints = list(memory.list(config, limit=2))
        >>> print(checkpoints)
        [CheckpointTuple(...), CheckpointTuple(...)]

        >>> config = {"configurable": {"thread_id": "1"}}
        >>> before = {"configurable": {"checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875"}}
        >>> with SqliteSaver.from_conn_string(":memory:") as memory:
        ... # Run a graph, then list the checkpoints
        >>>     checkpoints = list(memory.list(config, before=before))
        >>> print(checkpoints)
        [CheckpointTuple(...), ...]
    """
    where, param_values = search_where(config, filter, before)
    query = f"""SELECT thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata
    FROM checkpoints
    {where}
    ORDER BY checkpoint_id DESC"""
    if limit:
        query += f" LIMIT {limit}"
    with self.cursor(transaction=False) as cur, closing(self.conn.cursor()) as wcur:
        cur.execute(query, param_values)
        for (
            thread_id,
            checkpoint_ns,
            checkpoint_id,
            parent_checkpoint_id,
            type,
            checkpoint,
            metadata,
        ) in cur:
            wcur.execute(
                "SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
                (thread_id, checkpoint_ns, checkpoint_id),
            )
            yield CheckpointTuple(
                {
                    "configurable": {
                        "thread_id": thread_id,
                        "checkpoint_ns": checkpoint_ns,
                        "checkpoint_id": checkpoint_id,
                    }
                },
                self.serde.loads_typed((type, checkpoint)),
                self.jsonplus_serde.loads(metadata) if metadata is not None else {},
                (
                    {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": parent_checkpoint_id,
                        }
                    }
                    if parent_checkpoint_id
                    else None
                ),
                [
                    (task_id, channel, self.serde.loads_typed((type, value)))
                    for task_id, channel, type, value in wcur
                ],
            )

put(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions) -> RunnableConfig

将检查点保存到数据库。

此方法将检查点保存到 SQLite 数据库。检查点与提供的配置及其父配置(如果有)相关联。

参数

  • config (RunnableConfig) –

    与检查点关联的配置。

  • checkpoint (Checkpoint) –

    要保存的检查点。

  • metadata (CheckpointMetadata) –

    与检查点一起保存的附加元数据。

  • new_versions (ChannelVersions) –

    截至此次写入的新的通道版本。

返回值

  • RunnableConfig ( RunnableConfig ) –

    存储检查点后的更新配置。

示例

>>> from langgraph.checkpoint.sqlite import SqliteSaver
>>> with SqliteSaver.from_conn_string(":memory:") as memory:
>>>     config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
>>>     checkpoint = {"ts": "2024-05-04T06:32:42.235444+00:00", "id": "1ef4f797-8335-6428-8001-8a1503f9b875", "data": {"key": "value"}}
>>>     saved_config = memory.put(config, checkpoint, {"source": "input", "step": 1, "writes": {"key": "value"}}, {})
>>> print(saved_config)
{'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef4f797-8335-6428-8001-8a1503f9b875'}}
源代码位于 libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py
def put(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Save a checkpoint to the database.

    This method saves a checkpoint to the SQLite database. The checkpoint is associated
    with the provided config and its parent config (if any).

    Args:
        config (RunnableConfig): The config to associate with the checkpoint.
        checkpoint (Checkpoint): The checkpoint to save.
        metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
        new_versions (ChannelVersions): New channel versions as of this write.

    Returns:
        RunnableConfig: Updated configuration after storing the checkpoint.

    Examples:

        >>> from langgraph.checkpoint.sqlite import SqliteSaver
        >>> with SqliteSaver.from_conn_string(":memory:") as memory:
        >>>     config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
        >>>     checkpoint = {"ts": "2024-05-04T06:32:42.235444+00:00", "id": "1ef4f797-8335-6428-8001-8a1503f9b875", "data": {"key": "value"}}
        >>>     saved_config = memory.put(config, checkpoint, {"source": "input", "step": 1, "writes": {"key": "value"}}, {})
        >>> print(saved_config)
        {'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef4f797-8335-6428-8001-8a1503f9b875'}}
    """
    thread_id = config["configurable"]["thread_id"]
    checkpoint_ns = config["configurable"]["checkpoint_ns"]
    type_, serialized_checkpoint = self.serde.dumps_typed(checkpoint)
    serialized_metadata = self.jsonplus_serde.dumps(metadata)
    with self.cursor() as cur:
        cur.execute(
            "INSERT OR REPLACE INTO checkpoints (thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata) VALUES (?, ?, ?, ?, ?, ?, ?)",
            (
                str(config["configurable"]["thread_id"]),
                checkpoint_ns,
                checkpoint["id"],
                config["configurable"].get("checkpoint_id"),
                type_,
                serialized_checkpoint,
                serialized_metadata,
            ),
        )
    return {
        "configurable": {
            "thread_id": thread_id,
            "checkpoint_ns": checkpoint_ns,
            "checkpoint_id": checkpoint["id"],
        }
    }

put_writes(config: RunnableConfig, writes: Sequence[Tuple[str, Any]], task_id: str) -> None

存储与检查点链接的中间写入内容。

此方法将与检查点相关的中间写入保存到 SQLite 数据库。

参数

  • config (RunnableConfig) –

    相关检查点的配置。

  • writes (Sequence[Tuple[str, Any]]) –

    要存储的写入列表,每个写入都是 (通道,值) 对。

  • task_id (str) –

    创建写入内容的任务标识符。

源代码位于 libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py
def put_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[Tuple[str, Any]],
    task_id: str,
) -> None:
    """Store intermediate writes linked to a checkpoint.

    This method saves intermediate writes associated with a checkpoint to the SQLite database.

    Args:
        config (RunnableConfig): Configuration of the related checkpoint.
        writes (Sequence[Tuple[str, Any]]): List of writes to store, each as (channel, value) pair.
        task_id (str): Identifier for the task creating the writes.
    """
    query = (
        "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
        if all(w[0] in WRITES_IDX_MAP for w in writes)
        else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
    )
    with self.cursor() as cur:
        cur.executemany(
            query,
            [
                (
                    str(config["configurable"]["thread_id"]),
                    str(config["configurable"]["checkpoint_ns"]),
                    str(config["configurable"]["checkpoint_id"]),
                    task_id,
                    WRITES_IDX_MAP.get(channel, idx),
                    channel,
                    *self.serde.dumps_typed(value),
                )
                for idx, (channel, value) in enumerate(writes)
            ],
        )

aget_tuple(config: RunnableConfig) -> Optional[CheckpointTuple] async

异步从数据库中获取检查点元组。

注意

此异步方法不受 SqliteSaver 类支持。使用 get_tuple() 代替,或考虑使用 AsyncSqliteSaver

源代码位于 libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py
async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Get a checkpoint tuple from the database asynchronously.

    Note:
        This async method is not supported by the SqliteSaver class.
        Use get_tuple() instead, or consider using [AsyncSqliteSaver][langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver].
    """
    raise NotImplementedError(_AIO_ERROR_MSG)

alist(config: Optional[RunnableConfig], *, filter: Optional[Dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None) -> AsyncIterator[CheckpointTuple] async

异步列出数据库中的检查点。

注意

此异步方法不受 SqliteSaver 类支持。使用 list() 代替,或考虑使用 AsyncSqliteSaver

源代码位于 libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py
async def alist(
    self,
    config: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> AsyncIterator[CheckpointTuple]:
    """List checkpoints from the database asynchronously.

    Note:
        This async method is not supported by the SqliteSaver class.
        Use list() instead, or consider using [AsyncSqliteSaver][langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver].
    """
    raise NotImplementedError(_AIO_ERROR_MSG)
    yield

aput(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions) -> RunnableConfig async

异步将检查点保存到数据库。

注意

此异步方法不受 SqliteSaver 类支持。使用 put() 代替,或考虑使用 AsyncSqliteSaver

源代码位于 libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py
async def aput(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Save a checkpoint to the database asynchronously.

    Note:
        This async method is not supported by the SqliteSaver class.
        Use put() instead, or consider using [AsyncSqliteSaver][langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver].
    """
    raise NotImplementedError(_AIO_ERROR_MSG)

get_next_version(current: Optional[str], channel: ChannelProtocol) -> str

为通道生成下一个版本 ID。

此方法根据通道的当前版本创建通道的新版本标识符。

参数

  • current (Optional[str]) –

    通道的当前版本标识符。

  • channel (BaseChannel) –

    要进行版本控制的通道。

返回值

  • str ( str ) –

    下一个版本标识符,保证单调递增。

源代码位于 libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py
def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
    """Generate the next version ID for a channel.

    This method creates a new version identifier for a channel based on its current version.

    Args:
        current (Optional[str]): The current version identifier of the channel.
        channel (BaseChannel): The channel being versioned.

    Returns:
        str: The next version identifier, which is guaranteed to be monotonically increasing.
    """
    if current is None:
        current_v = 0
    elif isinstance(current, int):
        current_v = current
    else:
        current_v = int(current.split(".")[0])
    next_v = current_v + 1
    next_h = random.random()
    return f"{next_v:032}.{next_h:016}"

AsyncSqliteSaver

基类: BaseCheckpointSaver[str]

一个异步检查点保存器,将检查点存储在 SQLite 数据库中。

此类提供了一个异步接口,用于使用 SQLite 数据库保存和检索检查点。它专为异步环境设计,与同步替代方案相比,在 I/O 绑定操作方面提供了更好的性能。

属性

  • conn (Connection) –

    异步 SQLite 数据库连接。

  • serde (SerializerProtocol) –

    用于对检查点进行编码/解码的序列化器。

提示

需要 aiosqlite 包。使用 pip install aiosqlite 安装它。

警告

虽然此类支持异步检查点,但不建议用于生产工作负载,因为 SQLite 的写入性能有限。对于生产用途,请考虑更强大的数据库,例如 PostgreSQL。

提示

请务必在执行代码后关闭数据库连接,否则您可能会看到图表在执行后“挂起”(因为程序在连接关闭之前不会退出)。

最简单的方法是使用示例中所示的async with语句。

async with AsyncSqliteSaver.from_conn_string("checkpoints.sqlite") as saver:
    # Your code here
    graph = builder.compile(checkpointer=saver)
    config = {"configurable": {"thread_id": "thread-1"}}
    async for event in graph.astream_events(..., config, version="v1"):
        print(event)

示例

StateGraph 中的使用

>>> import asyncio
>>>
>>> from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
>>> from langgraph.graph import StateGraph
>>>
>>> builder = StateGraph(int)
>>> builder.add_node("add_one", lambda x: x + 1)
>>> builder.set_entry_point("add_one")
>>> builder.set_finish_point("add_one")
>>> async with AsyncSqliteSaver.from_conn_string("checkpoints.db") as memory:
>>>     graph = builder.compile(checkpointer=memory)
>>>     coro = graph.ainvoke(1, {"configurable": {"thread_id": "thread-1"}})
>>>     print(asyncio.run(coro))
Output: 2
原始用法

>>> import asyncio
>>> import aiosqlite
>>> from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
>>>
>>> async def main():
>>>     async with aiosqlite.connect("checkpoints.db") as conn:
...         saver = AsyncSqliteSaver(conn)
...         config = {"configurable": {"thread_id": "1"}}
...         checkpoint = {"ts": "2023-05-03T10:00:00Z", "data": {"key": "value"}}
...         saved_config = await saver.aput(config, checkpoint, {}, {})
...         print(saved_config)
>>> asyncio.run(main())
{"configurable": {"thread_id": "1", "checkpoint_id": "0c62ca34-ac19-445d-bbb0-5b4984975b2a"}}
libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py 中的源代码
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
class AsyncSqliteSaver(BaseCheckpointSaver[str]):
    """An asynchronous checkpoint saver that stores checkpoints in a SQLite database.

    This class provides an asynchronous interface for saving and retrieving checkpoints
    using a SQLite database. It's designed for use in asynchronous environments and
    offers better performance for I/O-bound operations compared to synchronous alternatives.

    Attributes:
        conn (aiosqlite.Connection): The asynchronous SQLite database connection.
        serde (SerializerProtocol): The serializer used for encoding/decoding checkpoints.

    Tip:
        Requires the [aiosqlite](https://pypi.ac.cn/project/aiosqlite/) package.
        Install it with `pip install aiosqlite`.

    Warning:
        While this class supports asynchronous checkpointing, it is not recommended
        for production workloads due to limitations in SQLite's write performance.
        For production use, consider a more robust database like PostgreSQL.

    Tip:
        Remember to **close the database connection** after executing your code,
        otherwise, you may see the graph "hang" after execution (since the program
        will not exit until the connection is closed).

        The easiest way is to use the `async with` statement as shown in the examples.

        ```python
        async with AsyncSqliteSaver.from_conn_string("checkpoints.sqlite") as saver:
            # Your code here
            graph = builder.compile(checkpointer=saver)
            config = {"configurable": {"thread_id": "thread-1"}}
            async for event in graph.astream_events(..., config, version="v1"):
                print(event)
        ```

    Examples:
        Usage within StateGraph:

        ```pycon
        >>> import asyncio
        >>>
        >>> from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
        >>> from langgraph.graph import StateGraph
        >>>
        >>> builder = StateGraph(int)
        >>> builder.add_node("add_one", lambda x: x + 1)
        >>> builder.set_entry_point("add_one")
        >>> builder.set_finish_point("add_one")
        >>> async with AsyncSqliteSaver.from_conn_string("checkpoints.db") as memory:
        >>>     graph = builder.compile(checkpointer=memory)
        >>>     coro = graph.ainvoke(1, {"configurable": {"thread_id": "thread-1"}})
        >>>     print(asyncio.run(coro))
        Output: 2
        ```
        Raw usage:

        ```pycon
        >>> import asyncio
        >>> import aiosqlite
        >>> from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
        >>>
        >>> async def main():
        >>>     async with aiosqlite.connect("checkpoints.db") as conn:
        ...         saver = AsyncSqliteSaver(conn)
        ...         config = {"configurable": {"thread_id": "1"}}
        ...         checkpoint = {"ts": "2023-05-03T10:00:00Z", "data": {"key": "value"}}
        ...         saved_config = await saver.aput(config, checkpoint, {}, {})
        ...         print(saved_config)
        >>> asyncio.run(main())
        {"configurable": {"thread_id": "1", "checkpoint_id": "0c62ca34-ac19-445d-bbb0-5b4984975b2a"}}
        ```
    """

    lock: asyncio.Lock
    is_setup: bool

    def __init__(
        self,
        conn: aiosqlite.Connection,
        *,
        serde: Optional[SerializerProtocol] = None,
    ):
        super().__init__(serde=serde)
        self.jsonplus_serde = JsonPlusSerializer()
        self.conn = conn
        self.lock = asyncio.Lock()
        self.loop = asyncio.get_running_loop()
        self.is_setup = False

    @classmethod
    @asynccontextmanager
    async def from_conn_string(
        cls, conn_string: str
    ) -> AsyncIterator["AsyncSqliteSaver"]:
        """Create a new AsyncSqliteSaver instance from a connection string.

        Args:
            conn_string (str): The SQLite connection string.

        Yields:
            AsyncSqliteSaver: A new AsyncSqliteSaver instance.
        """
        async with aiosqlite.connect(conn_string) as conn:
            yield AsyncSqliteSaver(conn)

    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Get a checkpoint tuple from the database.

        This method retrieves a checkpoint tuple from the SQLite database based on the
        provided config. If the config contains a "checkpoint_id" key, the checkpoint with
        the matching thread ID and checkpoint ID is retrieved. Otherwise, the latest checkpoint
        for the given thread ID is retrieved.

        Args:
            config (RunnableConfig): The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
        """
        try:
            # check if we are in the main thread, only bg threads can block
            # we don't check in other methods to avoid the overhead
            if asyncio.get_running_loop() is self.loop:
                raise asyncio.InvalidStateError(
                    "Synchronous calls to AsyncSqliteSaver are only allowed from a "
                    "different thread. From the main thread, use the async interface."
                    "For example, use `await checkpointer.aget_tuple(...)` or `await "
                    "graph.ainvoke(...)`."
                )
        except RuntimeError:
            pass
        return asyncio.run_coroutine_threadsafe(
            self.aget_tuple(config), self.loop
        ).result()

    def list(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[Dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> Iterator[CheckpointTuple]:
        """List checkpoints from the database asynchronously.

        This method retrieves a list of checkpoint tuples from the SQLite database based
        on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

        Args:
            config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
            filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
            before (Optional[RunnableConfig]): If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.
            limit (Optional[int]): Maximum number of checkpoints to return.

        Yields:
            Iterator[CheckpointTuple]: An iterator of matching checkpoint tuples.
        """
        aiter_ = self.alist(config, filter=filter, before=before, limit=limit)
        while True:
            try:
                yield asyncio.run_coroutine_threadsafe(
                    anext(aiter_),
                    self.loop,
                ).result()
            except StopAsyncIteration:
                break

    def put(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Save a checkpoint to the database.

        This method saves a checkpoint to the SQLite database. The checkpoint is associated
        with the provided config and its parent config (if any).

        Args:
            config (RunnableConfig): The config to associate with the checkpoint.
            checkpoint (Checkpoint): The checkpoint to save.
            metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
            new_versions (ChannelVersions): New channel versions as of this write.

        Returns:
            RunnableConfig: Updated configuration after storing the checkpoint.
        """
        return asyncio.run_coroutine_threadsafe(
            self.aput(config, checkpoint, metadata, new_versions), self.loop
        ).result()

    def put_writes(
        self, config: RunnableConfig, writes: Sequence[Tuple[str, Any]], task_id: str
    ) -> None:
        return asyncio.run_coroutine_threadsafe(
            self.aput_writes(config, writes, task_id), self.loop
        ).result()

    async def setup(self) -> None:
        """Set up the checkpoint database asynchronously.

        This method creates the necessary tables in the SQLite database if they don't
        already exist. It is called automatically when needed and should not be called
        directly by the user.
        """
        async with self.lock:
            if self.is_setup:
                return
            if not self.conn.is_alive():
                await self.conn
            async with self.conn.executescript(
                """
                PRAGMA journal_mode=WAL;
                CREATE TABLE IF NOT EXISTS checkpoints (
                    thread_id TEXT NOT NULL,
                    checkpoint_ns TEXT NOT NULL DEFAULT '',
                    checkpoint_id TEXT NOT NULL,
                    parent_checkpoint_id TEXT,
                    type TEXT,
                    checkpoint BLOB,
                    metadata BLOB,
                    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
                );
                CREATE TABLE IF NOT EXISTS writes (
                    thread_id TEXT NOT NULL,
                    checkpoint_ns TEXT NOT NULL DEFAULT '',
                    checkpoint_id TEXT NOT NULL,
                    task_id TEXT NOT NULL,
                    idx INTEGER NOT NULL,
                    channel TEXT NOT NULL,
                    type TEXT,
                    value BLOB,
                    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
                );
                """
            ):
                await self.conn.commit()

            self.is_setup = True

    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Get a checkpoint tuple from the database asynchronously.

        This method retrieves a checkpoint tuple from the SQLite database based on the
        provided config. If the config contains a "checkpoint_id" key, the checkpoint with
        the matching thread ID and checkpoint ID is retrieved. Otherwise, the latest checkpoint
        for the given thread ID is retrieved.

        Args:
            config (RunnableConfig): The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
        """
        await self.setup()
        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
        async with self.lock, self.conn.cursor() as cur:
            # find the latest checkpoint for the thread_id
            if checkpoint_id := get_checkpoint_id(config):
                await cur.execute(
                    "SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?",
                    (
                        str(config["configurable"]["thread_id"]),
                        checkpoint_ns,
                        checkpoint_id,
                    ),
                )
            else:
                await cur.execute(
                    "SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? ORDER BY checkpoint_id DESC LIMIT 1",
                    (str(config["configurable"]["thread_id"]), checkpoint_ns),
                )
            # if a checkpoint is found, return it
            if value := await cur.fetchone():
                (
                    thread_id,
                    checkpoint_id,
                    parent_checkpoint_id,
                    type,
                    checkpoint,
                    metadata,
                ) = value
                if not get_checkpoint_id(config):
                    config = {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": checkpoint_id,
                        }
                    }
                # find any pending writes
                await cur.execute(
                    "SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
                    (
                        str(config["configurable"]["thread_id"]),
                        checkpoint_ns,
                        str(config["configurable"]["checkpoint_id"]),
                    ),
                )
                # deserialize the checkpoint and metadata
                return CheckpointTuple(
                    config,
                    self.serde.loads_typed((type, checkpoint)),
                    self.jsonplus_serde.loads(metadata) if metadata is not None else {},
                    (
                        {
                            "configurable": {
                                "thread_id": thread_id,
                                "checkpoint_ns": checkpoint_ns,
                                "checkpoint_id": parent_checkpoint_id,
                            }
                        }
                        if parent_checkpoint_id
                        else None
                    ),
                    [
                        (task_id, channel, self.serde.loads_typed((type, value)))
                        async for task_id, channel, type, value in cur
                    ],
                )

    async def alist(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[Dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> AsyncIterator[CheckpointTuple]:
        """List checkpoints from the database asynchronously.

        This method retrieves a list of checkpoint tuples from the SQLite database based
        on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

        Args:
            config (Optional[RunnableConfig]): Base configuration for filtering checkpoints.
            filter (Optional[Dict[str, Any]]): Additional filtering criteria for metadata.
            before (Optional[RunnableConfig]): If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.
            limit (Optional[int]): Maximum number of checkpoints to return.

        Yields:
            AsyncIterator[CheckpointTuple]: An asynchronous iterator of matching checkpoint tuples.
        """
        await self.setup()
        where, params = search_where(config, filter, before)
        query = f"""SELECT thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata
        FROM checkpoints
        {where}
        ORDER BY checkpoint_id DESC"""
        if limit:
            query += f" LIMIT {limit}"
        async with self.lock, self.conn.execute(
            query, params
        ) as cur, self.conn.cursor() as wcur:
            async for (
                thread_id,
                checkpoint_ns,
                checkpoint_id,
                parent_checkpoint_id,
                type,
                checkpoint,
                metadata,
            ) in cur:
                await wcur.execute(
                    "SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
                    (thread_id, checkpoint_ns, checkpoint_id),
                )
                yield CheckpointTuple(
                    {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": checkpoint_id,
                        }
                    },
                    self.serde.loads_typed((type, checkpoint)),
                    self.jsonplus_serde.loads(metadata) if metadata is not None else {},
                    (
                        {
                            "configurable": {
                                "thread_id": thread_id,
                                "checkpoint_ns": checkpoint_ns,
                                "checkpoint_id": parent_checkpoint_id,
                            }
                        }
                        if parent_checkpoint_id
                        else None
                    ),
                    [
                        (task_id, channel, self.serde.loads_typed((type, value)))
                        async for task_id, channel, type, value in wcur
                    ],
                )

    async def aput(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Save a checkpoint to the database asynchronously.

        This method saves a checkpoint to the SQLite database. The checkpoint is associated
        with the provided config and its parent config (if any).

        Args:
            config (RunnableConfig): The config to associate with the checkpoint.
            checkpoint (Checkpoint): The checkpoint to save.
            metadata (CheckpointMetadata): Additional metadata to save with the checkpoint.
            new_versions (ChannelVersions): New channel versions as of this write.

        Returns:
            RunnableConfig: Updated configuration after storing the checkpoint.
        """
        await self.setup()
        thread_id = config["configurable"]["thread_id"]
        checkpoint_ns = config["configurable"]["checkpoint_ns"]
        type_, serialized_checkpoint = self.serde.dumps_typed(checkpoint)
        serialized_metadata = self.jsonplus_serde.dumps(metadata)
        async with self.lock, self.conn.execute(
            "INSERT OR REPLACE INTO checkpoints (thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata) VALUES (?, ?, ?, ?, ?, ?, ?)",
            (
                str(config["configurable"]["thread_id"]),
                checkpoint_ns,
                checkpoint["id"],
                config["configurable"].get("checkpoint_id"),
                type_,
                serialized_checkpoint,
                serialized_metadata,
            ),
        ):
            await self.conn.commit()
        return {
            "configurable": {
                "thread_id": thread_id,
                "checkpoint_ns": checkpoint_ns,
                "checkpoint_id": checkpoint["id"],
            }
        }

    async def aput_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[Tuple[str, Any]],
        task_id: str,
    ) -> None:
        """Store intermediate writes linked to a checkpoint asynchronously.

        This method saves intermediate writes associated with a checkpoint to the database.

        Args:
            config (RunnableConfig): Configuration of the related checkpoint.
            writes (Sequence[Tuple[str, Any]]): List of writes to store, each as (channel, value) pair.
            task_id (str): Identifier for the task creating the writes.
        """
        query = (
            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
            if all(w[0] in WRITES_IDX_MAP for w in writes)
            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
        )
        await self.setup()
        async with self.lock, self.conn.cursor() as cur:
            await cur.executemany(
                query,
                [
                    (
                        str(config["configurable"]["thread_id"]),
                        str(config["configurable"]["checkpoint_ns"]),
                        str(config["configurable"]["checkpoint_id"]),
                        task_id,
                        WRITES_IDX_MAP.get(channel, idx),
                        channel,
                        *self.serde.dumps_typed(value),
                    )
                    for idx, (channel, value) in enumerate(writes)
                ],
            )

    def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
        """Generate the next version ID for a channel.

        This method creates a new version identifier for a channel based on its current version.

        Args:
            current (Optional[str]): The current version identifier of the channel.
            channel (BaseChannel): The channel being versioned.

        Returns:
            str: The next version identifier, which is guaranteed to be monotonically increasing.
        """
        if current is None:
            current_v = 0
        elif isinstance(current, int):
            current_v = current
        else:
            current_v = int(current.split(".")[0])