From 6669634a98150dedd6ef8ad8bb4fde1df4d2cb77 Mon Sep 17 00:00:00 2001 From: Caspar Broekhuizen Date: Mon, 20 Oct 2025 17:04:06 -0700 Subject: [PATCH] style(langgraph): make format make lint --- .../langgraph/_internal/_constants.py | 3 +++ libs/langgraph/langgraph/pregel/_algo.py | 25 ++++++++++++----- libs/langgraph/langgraph/pregel/_loop.py | 27 ++++++++----------- libs/langgraph/langgraph/pregel/main.py | 2 +- libs/langgraph/langgraph/types.py | 3 --- libs/langgraph/tests/test_pregel.py | 7 +++-- 6 files changed, 36 insertions(+), 31 deletions(-) diff --git a/libs/langgraph/langgraph/_internal/_constants.py b/libs/langgraph/langgraph/_internal/_constants.py index abab56295..aebeb95e4 100644 --- a/libs/langgraph/langgraph/_internal/_constants.py +++ b/libs/langgraph/langgraph/_internal/_constants.py @@ -24,6 +24,7 @@ PREVIOUS = sys.intern("__previous__") # --- Reserved cache namespaces --- CACHE_NS_WRITES = sys.intern("__pregel_ns_writes") + # cache namespace for node writes # --- Reserved config.configurable keys --- @@ -77,6 +78,8 @@ CONF = cast(Literal["configurable"], sys.intern("configurable")) # key for the configurable dict in RunnableConfig NULL_TASK_ID = sys.intern("00000000-0000-0000-0000-000000000000") # the task_id to use for writes that are not associated with a task +RUNTIME_PLACEHOLDER = "__pregel_runtime_placeholder__" +# placeholder for untracked values replaced at runtime # redefined to avoid circular import with langgraph.constants _TAG_HIDDEN = sys.intern("langsmith:hidden") diff --git a/libs/langgraph/langgraph/pregel/_algo.py b/libs/langgraph/langgraph/pregel/_algo.py index caf5bbc95..ef7d31898 100644 --- a/libs/langgraph/langgraph/pregel/_algo.py +++ b/libs/langgraph/langgraph/pregel/_algo.py @@ -57,6 +57,7 @@ from langgraph._internal._constants import ( RESERVED, RESUME, RETURN, + RUNTIME_PLACEHOLDER, TASKS, ) from langgraph._internal._scratchpad import PregelScratchpad @@ -72,7 +73,6 @@ from langgraph.pregel._log import logger from langgraph.pregel._read import INPUT_CACHE_KEY_TYPE, PregelNode from langgraph.runtime import DEFAULT_RUNTIME, Runtime from langgraph.types import ( - RUNTIME_PLACEHOLDER, All, CacheKey, CachePolicy, @@ -1112,11 +1112,14 @@ class LazyAtomicCounter: self._counter = itertools.count(0).__next__ return self._counter() -def sanitize_untracked_values_in_send(packet: Send, channels: Mapping[str, BaseChannel]) -> Send: + +def sanitize_untracked_values_in_send( + packet: Send, channels: Mapping[str, BaseChannel] +) -> Send: """Replace any UntrackedValue contents in Send.arg with RUNTIME_PLACEHOLDER for checkpointing. - + Send is not typed and arg may be a nested dict.""" - + if not isinstance(packet.arg, dict): # Command return packet @@ -1133,7 +1136,10 @@ def sanitize_untracked_values_in_send(packet: Send, channels: Mapping[str, BaseC sanitized_arg = replace(packet.arg) return Send(node=packet.node, arg=sanitized_arg) -def rehydrate_untracked_values_in_send(packet: Send, channels: Mapping[str, BaseChannel]) -> Send: + +def rehydrate_untracked_values_in_send( + packet: Send, channels: Mapping[str, BaseChannel] +) -> Send: """Replace RUNTIME_PLACEHOLDERs in Send.arg with actual untracked values from UntrackedValue channels.""" if not isinstance(packet.arg, dict): @@ -1141,13 +1147,18 @@ def rehydrate_untracked_values_in_send(packet: Send, channels: Mapping[str, Base return packet # deepcopy to avoid mutating the original packet, as it is later persisted in checkpoints - arg_deepcopy = deepcopy(packet.arg) + arg_deepcopy = deepcopy(packet.arg) + def replace(obj: dict[str, Any]) -> dict[str, Any]: for k, v in obj.items(): if isinstance(v, dict): # arg can be nested dicts v = replace(v) - if v is RUNTIME_PLACEHOLDER and k in channels and isinstance(channels[k], UntrackedValue): + if ( + v is RUNTIME_PLACEHOLDER + and k in channels + and isinstance(channels[k], UntrackedValue) + ): obj[k] = channels[k].get() return obj diff --git a/libs/langgraph/langgraph/pregel/_loop.py b/libs/langgraph/langgraph/pregel/_loop.py index ea2a3ad26..5d8491489 100644 --- a/libs/langgraph/langgraph/pregel/_loop.py +++ b/libs/langgraph/langgraph/pregel/_loop.py @@ -20,7 +20,6 @@ from typing import ( TypeVar, cast, ) -from langgraph.channels.untracked_value import UntrackedValue from langchain_core.callbacks import AsyncParentRunManager, ParentRunManager from langchain_core.runnables import RunnableConfig @@ -62,6 +61,7 @@ from langgraph._internal._constants import ( from langgraph._internal._scratchpad import PregelScratchpad from langgraph._internal._typing import EMPTY_SEQ, MISSING from langgraph.channels.base import BaseChannel +from langgraph.channels.untracked_value import UntrackedValue from langgraph.constants import TAG_HIDDEN from langgraph.errors import ( EmptyInputError, @@ -75,12 +75,12 @@ from langgraph.pregel._algo import ( Call, GetNextVersion, PregelTaskWrites, - sanitize_untracked_values_in_send, apply_writes, checkpoint_null_version, increment, prepare_next_tasks, prepare_single_task, + sanitize_untracked_values_in_send, should_interrupt, task_path_str, ) @@ -327,20 +327,15 @@ class PregelLoop: # We never want to persist untracked values in checkpoints # because there is no guarantee that they are serializable - def _sanitize(group: WritesT) -> WritesT: - out: WritesT = [] - for c, v in group: - # Do not persist UntrackedValue channel writes - if isinstance(self.specs.get(c), UntrackedValue): - continue - # Sanitize UntrackedValues that are nested within Send packets - if c == TASKS and isinstance(v, Send): - out.append((c, sanitize_untracked_values_in_send(v, self.channels))) - else: - out.append((c, v)) - return out - - writes_to_save = _sanitize(writes_to_save) + writes_to_save = [ + # Sanitize UntrackedValues that are nested within Send packets + (c, sanitize_untracked_values_in_send(v, self.channels)) + if c == TASKS and isinstance(v, Send) + else (c, v) + for c, v in writes_to_save + # Do not persist UntrackedValue channel writes + if not isinstance(self.specs.get(c), UntrackedValue) + ] # save writes self.checkpoint_pending_writes.extend((task_id, c, v) for c, v in writes) diff --git a/libs/langgraph/langgraph/pregel/main.py b/libs/langgraph/langgraph/pregel/main.py index ea3309587..0a4d06925 100644 --- a/libs/langgraph/langgraph/pregel/main.py +++ b/libs/langgraph/langgraph/pregel/main.py @@ -106,11 +106,11 @@ from langgraph.errors import ( from langgraph.managed.base import ManagedValueSpec from langgraph.pregel._algo import ( PregelTaskWrites, - sanitize_untracked_values_in_send, _scratchpad, apply_writes, local_read, prepare_next_tasks, + sanitize_untracked_values_in_send, ) from langgraph.pregel._call import identifier from langgraph.pregel._checkpoint import ( diff --git a/libs/langgraph/langgraph/types.py b/libs/langgraph/langgraph/types.py index a5b80c662..a060e1446 100644 --- a/libs/langgraph/langgraph/types.py +++ b/libs/langgraph/langgraph/types.py @@ -27,9 +27,6 @@ from langgraph._internal._retry import default_retry_on from langgraph._internal._typing import MISSING, DeprecatedKwargs from langgraph.warnings import LangGraphDeprecatedSinceV10 -# placeholder for untracked values replaced at runtime -RUNTIME_PLACEHOLDER = "__pregel_runtime_placeholder__" - if TYPE_CHECKING: from langgraph.pregel.protocol import PregelProtocol diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index 8ee148d04..3b027af17 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -13,7 +13,6 @@ from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from random import randrange from typing import Annotated, Any, Literal, get_type_hints -from langgraph.channels.untracked_value import UntrackedValue import pytest from langchain_core.language_models import GenericFakeChatModel @@ -45,6 +44,7 @@ from langgraph.channels.binop import BinaryOperatorAggregate from langgraph.channels.ephemeral_value import EphemeralValue from langgraph.channels.last_value import LastValue from langgraph.channels.topic import Topic +from langgraph.channels.untracked_value import UntrackedValue from langgraph.config import get_stream_writer from langgraph.errors import GraphRecursionError, InvalidUpdateError, ParentCommand from langgraph.func import entrypoint, task @@ -8599,6 +8599,7 @@ def test_multiple_writes_same_channel_from_same_node( }, ] + def test_send_with_untracked_value(sync_checkpointer: BaseCheckpointSaver): """Test that Send objects work correctly with untracked values in state.""" @@ -8648,6 +8649,4 @@ def test_send_with_untracked_value(sync_checkpointer: BaseCheckpointSaver): # Check that the untracked resource is NOT in the final state checkpoint state = app.get_state(config) - assert ( - "session_resource" not in state.values - ) + assert "session_resource" not in state.values