mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-17 21:25:46 +02:00
Sydney RunkleandGitHub
5e7566f4a3
* new union syntax * fix test * second round of conversions by injecting future annotations * format + add top level makefile
54 lines
1.4 KiB
Python
54 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Protocol
|
|
|
|
from langgraph.checkpoint.base import Checkpoint, EmptyChannelError
|
|
from langgraph.checkpoint.base.id import uuid6
|
|
|
|
|
|
class ChannelProtocol(Protocol):
|
|
def checkpoint(self) -> Any | None: ...
|
|
|
|
|
|
def empty_checkpoint() -> Checkpoint:
|
|
return Checkpoint(
|
|
v=1,
|
|
id=str(uuid6(clock_seq=-2)),
|
|
ts=datetime.now(timezone.utc).isoformat(),
|
|
channel_values={},
|
|
channel_versions={},
|
|
versions_seen={},
|
|
)
|
|
|
|
|
|
def create_checkpoint(
|
|
checkpoint: Checkpoint,
|
|
channels: Mapping[str, ChannelProtocol] | None,
|
|
step: int,
|
|
*,
|
|
id: str | None = 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"],
|
|
)
|