diff --git a/libs/langgraph/langgraph/_internal/_typing.py b/libs/langgraph/langgraph/_internal/_typing.py index a708df4ae..adaddccef 100644 --- a/libs/langgraph/langgraph/_internal/_typing.py +++ b/libs/langgraph/langgraph/_internal/_typing.py @@ -41,18 +41,7 @@ StateLike: TypeAlias = TypedDictLikeV1 | TypedDictLikeV2 | DataclassLike | BaseM It can either be a `TypedDict`, `dataclass`, or Pydantic `BaseModel`. Note: we cannot use either `TypedDict` or `dataclass` directly due to limitations in type checking. """ - - -class UnsetSentinel: - """Singleton sentinel for unset optional parameters.""" - - __slots__ = () - - def __repr__(self) -> str: - return "MISSING" - - -MISSING: UnsetSentinel = UnsetSentinel() +MISSING = object() """Unset sentinel value.""" diff --git a/libs/langgraph/langgraph/graph/_node.py b/libs/langgraph/langgraph/graph/_node.py index a9e9e9282..d8238c2b3 100644 --- a/libs/langgraph/langgraph/graph/_node.py +++ b/libs/langgraph/langgraph/graph/_node.py @@ -88,7 +88,6 @@ class StateNodeSpec(Generic[NodeInputT, ContextT]): input_schema: type[NodeInputT] retry_policy: RetryPolicy | Sequence[RetryPolicy] | None cache_policy: CachePolicy | None - cache_policy_opt_out: bool = False is_error_handler: bool = False error_handler_node: str | None = None ends: tuple[str, ...] | dict[str, str] | None = EMPTY_SEQ diff --git a/libs/langgraph/langgraph/graph/state.py b/libs/langgraph/langgraph/graph/state.py index ba94c3276..bc37e9bc6 100644 --- a/libs/langgraph/langgraph/graph/state.py +++ b/libs/langgraph/langgraph/graph/state.py @@ -47,12 +47,7 @@ from langgraph._internal._fields import ( from langgraph._internal._pydantic import create_model from langgraph._internal._runnable import coerce_to_runnable from langgraph._internal._timeout import coerce_timeout_policy -from langgraph._internal._typing import ( - EMPTY_SEQ, - MISSING, - DeprecatedKwargs, - UnsetSentinel, -) +from langgraph._internal._typing import EMPTY_SEQ, MISSING, DeprecatedKwargs from langgraph.channels.base import BaseChannel from langgraph.channels.binop import BinaryOperatorAggregate from langgraph.channels.delta import DeltaChannel @@ -103,6 +98,15 @@ _CHANNEL_BRANCH_TO = "branch:to:{}" _DEFAULT_ERROR_HANDLER_NODE = "__default_error_handler__" +class _CachePolicyUnset: + """Sentinel: `cache_policy` was omitted on `add_node` (inherit defaults).""" + + __slots__ = () + + +_CACHE_POLICY_UNSET = _CachePolicyUnset() + + @dataclass(slots=True) class _NodeDefaults: """Default node policies applied to every node at compile time.""" @@ -268,6 +272,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]): self.context_schema = context_schema self._node_defaults: _NodeDefaults = _NodeDefaults() + self._cache_policy_opt_out: set[str] = set() self._add_schema(self.state_schema) self._add_schema(self.input_schema, allow_managed=False) @@ -674,7 +679,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]): metadata: dict[str, Any] | None = None, input_schema: type[NodeInputT] | None = None, retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None, - cache_policy: CachePolicy | None | UnsetSentinel = MISSING, + cache_policy: CachePolicy | None | _CachePolicyUnset = _CACHE_POLICY_UNSET, error_handler: StateNode[Any, ContextT] | None = None, destinations: dict[str, str] | tuple[str, ...] | None = None, timeout: float | timedelta | TimeoutPolicy | None = None, @@ -774,13 +779,13 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]): input_schema = cast(type[NodeInputT] | None, input_) timeout = coerce_timeout_policy(timeout) - cache_policy_opt_out = False - if cache_policy is MISSING: + explicit_cache_opt_out = False + if isinstance(cache_policy, _CachePolicyUnset): resolved_cache_policy: CachePolicy | None = None elif isinstance(cache_policy, CachePolicy): resolved_cache_policy = cache_policy else: - cache_policy_opt_out = True + explicit_cache_opt_out = True resolved_cache_policy = None if not isinstance(node, str): @@ -807,6 +812,10 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]): ) if action is None: raise RuntimeError + if explicit_cache_opt_out: + self._cache_policy_opt_out.add(node) + else: + self._cache_policy_opt_out.discard(node) if node in self.nodes: raise ValueError(f"Node `{node}` already present.") if node == END or node == START: @@ -894,7 +903,6 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]): input_schema=input_schema, retry_policy=retry_policy, cache_policy=resolved_cache_policy, - cache_policy_opt_out=cache_policy_opt_out, error_handler_node=handler_node_name, ends=ends, defer=defer, @@ -907,7 +915,6 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]): input_schema=inferred_input_schema, retry_policy=retry_policy, cache_policy=resolved_cache_policy, - cache_policy_opt_out=cache_policy_opt_out, error_handler_node=handler_node_name, ends=ends, defer=defer, @@ -920,7 +927,6 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]): input_schema=self.state_schema, retry_policy=retry_policy, cache_policy=resolved_cache_policy, - cache_policy_opt_out=cache_policy_opt_out, error_handler_node=handler_node_name, ends=ends, defer=defer, @@ -1318,7 +1324,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]): ) # Apply builder defaults to node specs. Per-node values always win. - for spec in self.nodes.values(): + for node_name, spec in self.nodes.items(): # error_handler: regular nodes only — handlers must never # catch themselves or other handlers. if ( @@ -1338,7 +1344,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]): not spec.is_error_handler and defaults.cache_policy is not None and spec.cache_policy is None - and not spec.cache_policy_opt_out + and node_name not in self._cache_policy_opt_out ): spec.cache_policy = defaults.cache_policy # timeout: all nodes — a stuck handler should be cancelled the diff --git a/libs/sdk-py/tests/integration/test_subgraphs.py b/libs/sdk-py/tests/integration/test_subgraphs.py index dda35f380..4b0e28e36 100644 --- a/libs/sdk-py/tests/integration/test_subgraphs.py +++ b/libs/sdk-py/tests/integration/test_subgraphs.py @@ -6,6 +6,9 @@ researcher, so this suite is hermetic (no LLM API key required). from __future__ import annotations +import asyncio +import threading + import pytest from .conftest import ASSISTANT_ID, DEEP_AGENT_ASSISTANT_ID @@ -13,12 +16,20 @@ from .conftest import ASSISTANT_ID, DEEP_AGENT_ASSISTANT_ID pytestmark = pytest.mark.integration +async def _collect_subgraphs_async(thread) -> list: + return [h async for h in thread.subgraphs] + + +def _collect_subgraphs_sync(thread) -> list: + return list(thread.subgraphs) + + async def test_subgraphs_agent_async(async_threads) -> None: """Plain nested `StateGraph.invoke` does not produce a scoped child handle.""" threads, _ = async_threads async with threads.stream(assistant_id=ASSISTANT_ID) as thread: await thread.run.start(input={"messages": [], "value": "init", "items": []}) - handles = [h async for h in thread.subgraphs] + handles = await _collect_subgraphs_async(thread) # Documented behavior: plain nested invokes do not show up as scoped # child handles; the canonical signal is `create_deep_agent`. assert handles == [] @@ -28,25 +39,47 @@ def test_subgraphs_agent_sync(sync_threads) -> None: threads, _ = sync_threads with threads.stream(assistant_id=ASSISTANT_ID) as thread: thread.run.start(input={"messages": [], "value": "init", "items": []}) - handles = list(thread.subgraphs) + handles = _collect_subgraphs_sync(thread) assert handles == [] async def test_subgraphs_deep_agent_async(async_threads) -> None: threads, _ = async_threads async with threads.stream(assistant_id=DEEP_AGENT_ASSISTANT_ID) as thread: + # Subscribe before / during the run so child-namespace lifecycle events + # are not missed when the graph finishes quickly. + collect = asyncio.create_task(_collect_subgraphs_async(thread)) await thread.run.start( input={"messages": [{"role": "user", "content": "research the v3 spec"}]}, ) - handles = [h async for h in thread.subgraphs] + handles = await collect assert handles, "deep_agent should produce at least one direct-child handle" def test_subgraphs_deep_agent_sync(sync_threads) -> None: threads, _ = sync_threads with threads.stream(assistant_id=DEEP_AGENT_ASSISTANT_ID) as thread: - thread.run.start( - input={"messages": [{"role": "user", "content": "research the v3 spec"}]}, - ) - handles = list(thread.subgraphs) + handles: list = [] + error: list[BaseException] = [] + + def collect() -> None: + try: + handles.extend(_collect_subgraphs_sync(thread)) + except BaseException as exc: + error.append(exc) + + worker = threading.Thread(target=collect) + worker.start() + try: + thread.run.start( + input={ + "messages": [{"role": "user", "content": "research the v3 spec"}] + }, + ) + worker.join(timeout=30) + finally: + if worker.is_alive(): + worker.join(timeout=1) + if error: + raise error[0] assert handles, "deep_agent should produce at least one direct-child handle"