Reduce cpu time spent on langchain-core utilities

- shaves off 2s of 11s runtime of a simple graph with 1,000 subgraphs

libs/langgraph/langgraph/graph/graph.py

@@ -38,7 +38,7 @@ from langgraph.pregel import Channel, Pregel
 from langgraph.pregel.read import PregelNode
 from langgraph.pregel.types import All
 from langgraph.pregel.write import ChannelWrite, ChannelWriteEntry
-from langgraph.utils import RunnableCallable, coerce_to_runnable
+from langgraph.utils.runnable import RunnableCallable, coerce_to_runnable
 
 logger = logging.getLogger(__name__)
 

libs/langgraph/langgraph/graph/state.py

@@ -44,7 +44,11 @@ from langgraph.pregel.read import ChannelRead, PregelNode
 from langgraph.pregel.types import All, RetryPolicy
 from langgraph.pregel.write import SKIP_WRITE, ChannelWrite, ChannelWriteEntry
 from langgraph.store.base import BaseStore
-from langgraph.utils import RunnableCallable, coerce_to_runnable, get_field_default
+from langgraph.utils.fields import get_field_default
+from langgraph.utils.runnable import (
+    RunnableCallable,
+    coerce_to_runnable,
+)
 
 logger = logging.getLogger(__name__)
 

libs/langgraph/langgraph/prebuilt/tool_executor.py

@@ -6,7 +6,7 @@ from langchain_core.tools import BaseTool
 from langchain_core.tools import tool as create_tool
 
 from langgraph._api.deprecation import deprecated
-from langgraph.utils import RunnableCallable
+from langgraph.utils.runnable import RunnableCallable
 
 INVALID_TOOL_MSG_TEMPLATE = (
     "{requested_tool_name} is not a valid tool, "

libs/langgraph/langgraph/prebuilt/tool_node.py

@@ -21,7 +21,7 @@ from langchain_core.tools import BaseTool, InjectedToolArg
 from langchain_core.tools import tool as create_tool
 from typing_extensions import get_args
 
-from langgraph.utils import RunnableCallable
+from langgraph.utils.runnable import RunnableCallable
 
 INVALID_TOOL_NAME_ERROR_TEMPLATE = (
     "Error: {requested_tool} is not a valid tool, try one of [{available_tools}]."

libs/langgraph/langgraph/prebuilt/tool_validator.py

@@ -33,7 +33,7 @@ from langchain_core.tools import BaseTool, create_schema_from_function
 from pydantic import BaseModel as BaseModelV2
 from pydantic import ValidationError as ValidationErrorV2
 
-from langgraph.utils import RunnableCallable
+from langgraph.utils.runnable import RunnableCallable
 
 
 def _default_format_error(

libs/langgraph/langgraph/pregel/__init__.py

@@ -34,8 +34,6 @@ from langchain_core.runnables.config import (
     ensure_config,
     get_async_callback_manager_for_config,
     get_callback_manager_for_config,
-    merge_configs,
-    patch_config,
 )
 from langchain_core.runnables.utils import (
     ConfigurableFieldSpec,
@@ -76,7 +74,6 @@ from langgraph.pregel.algo import (
     local_write,
     prepare_next_tasks,
 )
-from langgraph.pregel.config import patch_checkpoint_map, patch_configurable
 from langgraph.pregel.debug import tasks_w_writes
 from langgraph.pregel.io import read_channels
 from langgraph.pregel.loop import AsyncPregelLoop, SyncPregelLoop
@@ -96,7 +93,13 @@ from langgraph.pregel.utils import (
 from langgraph.pregel.validate import validate_graph, validate_keys
 from langgraph.pregel.write import ChannelWrite, ChannelWriteEntry
 from langgraph.store.base import BaseStore
-from langgraph.utils import RunnableCallable
+from langgraph.utils.config import (
+    merge_configs,
+    patch_checkpoint_map,
+    patch_config,
+    patch_configurable,
+)
+from langgraph.utils.runnable import RunnableCallable
 
 WriteValue = Union[
     Runnable[Input, Output],

libs/langgraph/langgraph/pregel/algo.py

@@ -17,11 +17,7 @@ from typing import (
 from uuid import UUID, uuid5
 
 from langchain_core.callbacks.manager import AsyncParentRunManager, ParentRunManager
-from langchain_core.runnables.config import (
-    RunnableConfig,
-    merge_configs,
-    patch_config,
-)
+from langchain_core.runnables.config import RunnableConfig
 
 from langgraph.channels.base import BaseChannel
 from langgraph.checkpoint.base import (
@@ -51,6 +47,7 @@ from langgraph.pregel.log import logger
 from langgraph.pregel.manager import ChannelsManager
 from langgraph.pregel.read import PregelNode
 from langgraph.pregel.types import All, PregelExecutableTask, PregelTask
+from langgraph.utils.config import merge_configs, patch_config
 
 
 class WritesProtocol(Protocol):

libs/langgraph/langgraph/pregel/config.py

@@ -1,34 +0,0 @@
-from typing import Any, Optional
-
-from langchain_core.runnables import RunnableConfig
-
-from langgraph.checkpoint.base import CheckpointMetadata
-from langgraph.constants import CONFIG_KEY_CHECKPOINT_MAP
-
-
-def patch_configurable(
-    config: Optional[RunnableConfig], patch: dict[str, Any]
-) -> RunnableConfig:
-    if config is None:
-        return {"configurable": patch}
-    else:
-        return {**config, "configurable": {**config["configurable"], **patch}}
-
-
-def patch_checkpoint_map(
-    config: RunnableConfig, metadata: Optional[CheckpointMetadata]
-) -> RunnableConfig:
-    if parents := (metadata.get("parents") if metadata else None):
-        return patch_configurable(
-            config,
-            {
-                CONFIG_KEY_CHECKPOINT_MAP: {
-                    **parents,
-                    config["configurable"]["checkpoint_ns"]: config["configurable"][
-                        "checkpoint_id"
-                    ],
-                },
-            },
-        )
-    else:
-        return config

libs/langgraph/langgraph/pregel/loop.py

@@ -59,7 +59,6 @@ from langgraph.pregel.algo import (
     prepare_next_tasks,
     should_interrupt,
 )
-from langgraph.pregel.config import patch_configurable
 from langgraph.pregel.debug import (
     map_debug_checkpoint,
     map_debug_task_results,
@@ -86,6 +85,7 @@ from langgraph.pregel.types import PregelExecutableTask
 from langgraph.pregel.utils import get_new_channel_versions
 from langgraph.store.base import BaseStore
 from langgraph.store.batch import AsyncBatchedStore
+from langgraph.utils.config import patch_configurable
 
 V = TypeVar("V")
 INPUT_DONE = object()

libs/langgraph/langgraph/pregel/manager.py

@@ -2,7 +2,7 @@ import asyncio
 from contextlib import AsyncExitStack, ExitStack, asynccontextmanager, contextmanager
 from typing import AsyncIterator, Iterator, Mapping, Optional, Union
 
-from langchain_core.runnables import RunnableConfig, patch_config
+from langchain_core.runnables import RunnableConfig
 
 from langgraph.channels.base import BaseChannel
 from langgraph.checkpoint.base import Checkpoint
@@ -14,6 +14,7 @@ from langgraph.managed.base import (
 )
 from langgraph.managed.context import Context
 from langgraph.store.base import BaseStore
+from langgraph.utils.config import patch_configurable
 
 
 @contextmanager
@@ -26,7 +27,7 @@ def ChannelsManager(
     skip_context: bool = False,
 ) -> Iterator[tuple[Mapping[str, BaseChannel], ManagedValueMapping]]:
     """Manage channels for the lifetime of a Pregel invocation (multiple steps)."""
-    config_for_managed = patch_config(config, configurable={CONFIG_KEY_STORE: store})
+    config_for_managed = patch_configurable(config, {CONFIG_KEY_STORE: store})
     channel_specs: Mapping[str, BaseChannel] = {}
     managed_specs: Mapping[str, ManagedValueSpec] = {}
     for k, v in specs.items():
@@ -69,7 +70,7 @@ async def AsyncChannelsManager(
     skip_context: bool = False,
 ) -> AsyncIterator[Mapping[str, BaseChannel]]:
     """Manage channels for the lifetime of a Pregel invocation (multiple steps)."""
-    config_for_managed = patch_config(config, configurable={CONFIG_KEY_STORE: store})
+    config_for_managed = patch_configurable(config, {CONFIG_KEY_STORE: store})
     channel_specs: Mapping[str, BaseChannel] = {}
     managed_specs: Mapping[str, ManagedValueSpec] = {}
     for k, v in specs.items():

libs/langgraph/langgraph/pregel/read.py

@@ -19,13 +19,13 @@ from langchain_core.runnables import (
     RunnableSerializable,
 )
 from langchain_core.runnables.base import Input, Other, Output, coerce_to_runnable
-from langchain_core.runnables.config import merge_configs
 from langchain_core.runnables.utils import ConfigurableFieldSpec
 
 from langgraph.constants import CONFIG_KEY_READ
 from langgraph.pregel.retry import RetryPolicy
 from langgraph.pregel.write import ChannelWrite
-from langgraph.utils import RunnableCallable
+from langgraph.utils.config import merge_configs
+from langgraph.utils.runnable import RunnableCallable
 
 READ_TYPE = Callable[[str, bool], Union[Any, dict[str, Any]]]
 

libs/langgraph/langgraph/pregel/write.py

@@ -18,7 +18,7 @@ from langchain_core.runnables.utils import ConfigurableFieldSpec
 
 from langgraph.constants import CONFIG_KEY_SEND, TASKS, Send
 from langgraph.errors import InvalidUpdateError
-from langgraph.utils import RunnableCallable
+from langgraph.utils.runnable import RunnableCallable
 
 TYPE_SEND = Callable[[Sequence[tuple[str, Any]]], None]
 R = TypeVar("R", bound=Runnable)

libs/langgraph/langgraph/utils/config.py

@@ -0,0 +1,150 @@
+from typing import Any, Optional
+
+from langchain_core.callbacks import Callbacks
+from langchain_core.runnables import RunnableConfig
+from langchain_core.runnables.config import COPIABLE_KEYS, DEFAULT_RECURSION_LIMIT
+
+from langgraph.checkpoint.base import CheckpointMetadata
+from langgraph.constants import CONFIG_KEY_CHECKPOINT_MAP
+
+
+def patch_configurable(
+    config: Optional[RunnableConfig], patch: dict[str, Any]
+) -> RunnableConfig:
+    if config is None:
+        return {"configurable": patch}
+    else:
+        return {**config, "configurable": {**config["configurable"], **patch}}
+
+
+def patch_checkpoint_map(
+    config: RunnableConfig, metadata: Optional[CheckpointMetadata]
+) -> RunnableConfig:
+    if parents := (metadata.get("parents") if metadata else None):
+        return patch_configurable(
+            config,
+            {
+                CONFIG_KEY_CHECKPOINT_MAP: {
+                    **parents,
+                    config["configurable"]["checkpoint_ns"]: config["configurable"][
+                        "checkpoint_id"
+                    ],
+                },
+            },
+        )
+    else:
+        return config
+
+
+def merge_configs(*configs: Optional[RunnableConfig]) -> RunnableConfig:
+    """Merge multiple configs into one.
+
+    Args:
+        *configs (Optional[RunnableConfig]): The configs to merge.
+
+    Returns:
+        RunnableConfig: The merged config.
+    """
+    base: RunnableConfig = {}
+    # Even though the keys aren't literals, this is correct
+    # because both dicts are the same type
+    for config in configs:
+        if config is None:
+            continue
+        for key in config:
+            if key == "metadata":
+                base[key] = {  # type: ignore
+                    **base.get(key, {}),  # type: ignore
+                    **(config.get(key) or {}),  # type: ignore
+                }
+            elif key == "tags":
+                base[key] = sorted(  # type: ignore
+                    set(base.get(key, []) + (config.get(key) or [])),  # type: ignore
+                )
+            elif key == "configurable":
+                base[key] = {  # type: ignore
+                    **base.get(key, {}),  # type: ignore
+                    **(config.get(key) or {}),  # type: ignore
+                }
+            elif key == "callbacks":
+                base_callbacks = base.get("callbacks")
+                these_callbacks = config["callbacks"]
+                # callbacks can be either None, list[handler] or manager
+                # so merging two callbacks values has 6 cases
+                if isinstance(these_callbacks, list):
+                    if base_callbacks is None:
+                        base["callbacks"] = these_callbacks.copy()
+                    elif isinstance(base_callbacks, list):
+                        base["callbacks"] = base_callbacks + these_callbacks
+                    else:
+                        # base_callbacks is a manager
+                        mngr = base_callbacks.copy()
+                        for callback in these_callbacks:
+                            mngr.add_handler(callback, inherit=True)
+                        base["callbacks"] = mngr
+                elif these_callbacks is not None:
+                    # these_callbacks is a manager
+                    if base_callbacks is None:
+                        base["callbacks"] = these_callbacks.copy()
+                    elif isinstance(base_callbacks, list):
+                        mngr = these_callbacks.copy()
+                        for callback in base_callbacks:
+                            mngr.add_handler(callback, inherit=True)
+                        base["callbacks"] = mngr
+                    else:
+                        # base_callbacks is also a manager
+                        base["callbacks"] = base_callbacks.merge(these_callbacks)
+            elif key == "recursion_limit":
+                if config["recursion_limit"] != DEFAULT_RECURSION_LIMIT:
+                    base["recursion_limit"] = config["recursion_limit"]
+            elif key in COPIABLE_KEYS and config[key] is not None:  # type: ignore[literal-required]
+                base[key] = config[key].copy()  # type: ignore[literal-required]
+            else:
+                base[key] = config[key] or base.get(key)  # type: ignore
+    return base
+
+
+def patch_config(
+    config: Optional[RunnableConfig],
+    *,
+    callbacks: Optional[Callbacks] = None,
+    recursion_limit: Optional[int] = None,
+    max_concurrency: Optional[int] = None,
+    run_name: Optional[str] = None,
+    configurable: Optional[dict[str, Any]] = None,
+) -> RunnableConfig:
+    """Patch a config with new values.
+
+    Args:
+        config (Optional[RunnableConfig]): The config to patch.
+        callbacks (Optional[BaseCallbackManager], optional): The callbacks to set.
+          Defaults to None.
+        recursion_limit (Optional[int], optional): The recursion limit to set.
+          Defaults to None.
+        max_concurrency (Optional[int], optional): The max concurrency to set.
+          Defaults to None.
+        run_name (Optional[str], optional): The run name to set. Defaults to None.
+        configurable (Optional[Dict[str, Any]], optional): The configurable to set.
+          Defaults to None.
+
+    Returns:
+        RunnableConfig: The patched config.
+    """
+    config = config or {}
+    if callbacks is not None:
+        # If we're replacing callbacks, we need to unset run_name
+        # As that should apply only to the same run as the original callbacks
+        config["callbacks"] = callbacks
+        if "run_name" in config:
+            del config["run_name"]
+        if "run_id" in config:
+            del config["run_id"]
+    if recursion_limit is not None:
+        config["recursion_limit"] = recursion_limit
+    if max_concurrency is not None:
+        config["max_concurrency"] = max_concurrency
+    if run_name is not None:
+        config["run_name"] = run_name
+    if configurable is not None:
+        config["configurable"] = {**config.get("configurable", {}), **configurable}
+    return config

libs/langgraph/langgraph/utils/fields.py

@@ -0,0 +1,101 @@
+from typing import Any, Optional, Type, Union
+
+from typing_extensions import (
+    Annotated,
+    NotRequired,
+    ReadOnly,
+    Required,
+    get_origin,
+)
+
+
+def _is_optional_type(type_: Any) -> bool:
+    """Check if a type is Optional."""
+
+    if hasattr(type_, "__origin__") and hasattr(type_, "__args__"):
+        origin = get_origin(type_)
+        if origin is Optional:
+            return True
+        if origin is Union:
+            return any(
+                arg is type(None) or _is_optional_type(arg) for arg in type_.__args__
+            )
+        if origin is Annotated:
+            return _is_optional_type(type_.__args__[0])
+        return origin is None
+    if hasattr(type_, "__bound__") and type_.__bound__ is not None:
+        return _is_optional_type(type_.__bound__)
+    return type_ is None
+
+
+def _is_required_type(type_: Any) -> Optional[bool]:
+    """Check if an annotation is marked as Required/NotRequired.
+
+    Returns:
+        - True if required
+        - False if not required
+        - None if not annotated with either
+    """
+    origin = get_origin(type_)
+    if origin is Required:
+        return True
+    if origin is NotRequired:
+        return False
+    if origin is Annotated or getattr(origin, "__args__", None):
+        # See https://typing.readthedocs.io/en/latest/spec/typeddict.html#interaction-with-annotated
+        return _is_required_type(type_.__args__[0])
+    return None
+
+
+def _is_readonly_type(type_: Any) -> bool:
+    """Check if an annotation is marked as ReadOnly.
+
+    Returns:
+        - True if is read only
+        - False if not read only
+    """
+
+    # See: https://typing.readthedocs.io/en/latest/spec/typeddict.html#typing-readonly-type-qualifier
+    origin = get_origin(type_)
+    if origin is Annotated:
+        return _is_readonly_type(type_.__args__[0])
+    if origin is ReadOnly:
+        return True
+    return False
+
+
+_DEFAULT_KEYS = frozenset()
+
+
+def get_field_default(name: str, type_: Any, schema: Type[Any]) -> Any:
+    """Determine the default value for a field in a state schema.
+
+    This is based on:
+        If TypedDict:
+            - Required/NotRequired
+            - total=False -> everything optional
+        - Type annotation (Optional/Union[None])
+    """
+    optional_keys = getattr(schema, "__optional_keys__", _DEFAULT_KEYS)
+    irq = _is_required_type(type_)
+    if name in optional_keys:
+        # Either total=False or explicit NotRequired.
+        # No type annotation trumps this.
+        if irq:
+            # Unless it's earlier versions of python & explicit Required
+            return ...
+        return None
+    if irq is not None:
+        if irq:
+            # Handle Required[<type>]
+            # (we already handled NotRequired and total=False)
+            return ...
+        # Handle NotRequired[<type>] for earlier versions of python
+        return None
+    # Note, we ignore ReadOnly attributes,
+    # as they don't make much sense. (we don't care if you mutate the state in your node)
+    # and mutating state in your node has no effect on our graph state.
+    # Base case is the annotation
+    if _is_optional_type(type_):
+        return None
+    return ...

libs/langgraph/langgraph/utils/runnable.py

@@ -4,8 +4,9 @@ import inspect
 import sys
 from contextvars import copy_context
 from functools import partial, wraps
-from typing import Any, AsyncIterator, Awaitable, Callable, Optional, Type, Union
+from typing import Any, AsyncIterator, Awaitable, Callable, Optional
 
+from langchain_core.load.serializable import to_json_not_implemented
 from langchain_core.runnables.base import (
     Runnable,
     RunnableConfig,
@@ -14,19 +15,16 @@ from langchain_core.runnables.base import (
     RunnableParallel,
 )
 from langchain_core.runnables.config import (
-    merge_configs,
+    ensure_config,
+    get_async_callback_manager_for_config,
+    get_callback_manager_for_config,
     run_in_executor,
     var_child_runnable_config,
 )
-from langchain_core.runnables.utils import accepts_config
-from typing_extensions import (
-    Annotated,
-    NotRequired,
-    ReadOnly,
-    Required,
-    TypeGuard,
-    get_origin,
-)
+from langchain_core.runnables.utils import accepts_config, accepts_run_manager
+from typing_extensions import TypeGuard
+
+from langgraph.utils.config import merge_configs, patch_config
 
 try:
     from langchain_core.runnables.config import _set_config_context
@@ -42,6 +40,9 @@ class StrEnum(str, enum.Enum):
     """A string enum."""
 
 
+ASYNCIO_ACCEPTS_CONTEXT = sys.version_info >= (3, 11)
+
+
 class RunnableCallable(Runnable):
     """A much simpler version of RunnableLambda that requires sync and async functions."""
 
@@ -70,11 +71,18 @@ class RunnableCallable(Runnable):
             except AttributeError:
                 pass
         self.func = func
+        if func is not None:
+            self.func_accepts_config = accepts_config(func)
+            self.func_accepts_run_manager = accepts_run_manager(func)
         self.afunc = afunc
+        if afunc is not None:
+            self.afunc_accepts_config = accepts_config(afunc)
+            self.afunc_accepts_run_manager = accepts_run_manager(afunc)
         self.config: Optional[RunnableConfig] = {"tags": tags} if tags else None
         self.kwargs = kwargs
         self.trace = trace
         self.recurse = recurse
+        self.serialized = to_json_not_implemented(self)
 
     def __repr__(self) -> str:
         repr_args = {
@@ -94,15 +102,34 @@ class RunnableCallable(Runnable):
                 " via the async API (ainvoke, astream, etc.)"
             )
         kwargs = {**self.kwargs, **kwargs}
+        config = ensure_config(merge_configs(self.config, config))
+        context = copy_context()
         if self.trace:
-            ret = self._call_with_config(
-                self.func, input, merge_configs(self.config, config), **kwargs
+            config = ensure_config(config)
+            callback_manager = get_callback_manager_for_config(config)
+            run_manager = callback_manager.on_chain_start(
+                self.serialized,
+                input,
+                name=config.get("run_name") or self.get_name(),
+                run_id=config.pop("run_id", None),
             )
+            try:
+                child_config = patch_config(config, callbacks=run_manager.get_child())
+                context = copy_context()
+                context.run(_set_config_context, child_config)
+                if self.func_accepts_config:
+                    kwargs["config"] = config
+                if self.func_accepts_run_manager:
+                    kwargs["run_manager"] = run_manager
+                ret = context.run(self.func, input, **kwargs)
+            except BaseException as e:
+                run_manager.on_chain_error(e)
+                raise
+            else:
+                run_manager.on_chain_end(ret)
         else:
-            config = merge_configs(self.config, config)
-            context = copy_context()
             context.run(_set_config_context, config)
-            if accepts_config(self.func):
+            if self.func_accepts_config:
                 kwargs["config"] = config
             ret = context.run(self.func, input, **kwargs)
         if isinstance(ret, Runnable) and self.recurse:
@@ -115,17 +142,38 @@ class RunnableCallable(Runnable):
         if not self.afunc:
             return self.invoke(input, config)
         kwargs = {**self.kwargs, **kwargs}
+        config = ensure_config(merge_configs(self.config, config))
+        context = copy_context()
         if self.trace:
-            ret = await self._acall_with_config(
-                self.afunc, input, merge_configs(self.config, config), **kwargs
+            callback_manager = get_async_callback_manager_for_config(config)
+            run_manager = await callback_manager.on_chain_start(
+                self.serialized,
+                input,
+                name=config.get("run_name") or self.name,
+                run_id=config.pop("run_id", None),
             )
+            try:
+                child_config = patch_config(config, callbacks=run_manager.get_child())
+                context.run(_set_config_context, child_config)
+                if self.afunc_accepts_config:
+                    kwargs["config"] = config
+                if self.afunc_accepts_run_manager:
+                    kwargs["run_manager"] = run_manager
+                coro = self.afunc(input, **kwargs)
+                if ASYNCIO_ACCEPTS_CONTEXT:
+                    ret = await asyncio.create_task(coro, context=context)
+                else:
+                    ret = await coro
+            except BaseException as e:
+                await run_manager.on_chain_error(e)
+                raise
+            else:
+                await run_manager.on_chain_end(ret)
         else:
-            config = merge_configs(self.config, config)
-            context = copy_context()
             context.run(_set_config_context, config)
-            if accepts_config(self.afunc):
+            if self.afunc_accepts_config:
                 kwargs["config"] = config
-            if sys.version_info >= (3, 11):
+            if ASYNCIO_ACCEPTS_CONTEXT:
                 ret = await asyncio.create_task(
                     self.afunc(input, **kwargs), context=context
                 )
@@ -188,95 +236,3 @@ def coerce_to_runnable(thing: RunnableLike, *, name: str, trace: bool) -> Runnab
             f"Expected a Runnable, callable or dict."
             f"Instead got an unsupported type: {type(thing)}"
         )
-
-
-def _is_optional_type(type_: Any) -> bool:
-    """Check if a type is Optional."""
-
-    if hasattr(type_, "__origin__") and hasattr(type_, "__args__"):
-        origin = get_origin(type_)
-        if origin is Optional:
-            return True
-        if origin is Union:
-            return any(
-                arg is type(None) or _is_optional_type(arg) for arg in type_.__args__
-            )
-        if origin is Annotated:
-            return _is_optional_type(type_.__args__[0])
-        return origin is None
-    if hasattr(type_, "__bound__") and type_.__bound__ is not None:
-        return _is_optional_type(type_.__bound__)
-    return type_ is None
-
-
-def _is_required_type(type_: Any) -> Optional[bool]:
-    """Check if an annotation is marked as Required/NotRequired.
-
-    Returns:
-        - True if required
-        - False if not required
-        - None if not annotated with either
-    """
-    origin = get_origin(type_)
-    if origin is Required:
-        return True
-    if origin is NotRequired:
-        return False
-    if origin is Annotated or getattr(origin, "__args__", None):
-        # See https://typing.readthedocs.io/en/latest/spec/typeddict.html#interaction-with-annotated
-        return _is_required_type(type_.__args__[0])
-    return None
-
-
-def _is_readonly_type(type_: Any) -> bool:
-    """Check if an annotation is marked as ReadOnly.
-
-    Returns:
-        - True if is read only
-        - False if not read only
-    """
-
-    # See: https://typing.readthedocs.io/en/latest/spec/typeddict.html#typing-readonly-type-qualifier
-    origin = get_origin(type_)
-    if origin is Annotated:
-        return _is_readonly_type(type_.__args__[0])
-    if origin is ReadOnly:
-        return True
-    return False
-
-
-_DEFAULT_KEYS = frozenset()
-
-
-def get_field_default(name: str, type_: Any, schema: Type[Any]) -> Any:
-    """Determine the default value for a field in a state schema.
-
-    This is based on:
-        If TypedDict:
-            - Required/NotRequired
-            - total=False -> everything optional
-        - Type annotation (Optional/Union[None])
-    """
-    optional_keys = getattr(schema, "__optional_keys__", _DEFAULT_KEYS)
-    irq = _is_required_type(type_)
-    if name in optional_keys:
-        # Either total=False or explicit NotRequired.
-        # No type annotation trumps this.
-        if irq:
-            # Unless it's earlier versions of python & explicit Required
-            return ...
-        return None
-    if irq is not None:
-        if irq:
-            # Handle Required[<type>]
-            # (we already handled NotRequired and total=False)
-            return ...
-        # Handle NotRequired[<type>] for earlier versions of python
-        return None
-    # Note, we ignore ReadOnly attributes,
-    # as they don't make much sense. (we don't care if you mutate the state in your node)
-    # and mutating state in your node has no effect on our graph state.
-    # Base case is the annotation
-    if _is_optional_type(type_):
-        return None
-    return ...

libs/langgraph/tests/test_utils.py

@@ -21,12 +21,8 @@ from typing_extensions import Annotated, NotRequired, Required
 
 from langgraph.graph import END, StateGraph
 from langgraph.graph.graph import CompiledGraph
-from langgraph.utils import (
-    _is_optional_type,
-    get_field_default,
-    is_async_callable,
-    is_async_generator,
-)
+from langgraph.utils.fields import _is_optional_type, get_field_default
+from langgraph.utils.runnable import is_async_callable, is_async_generator
 
 pytestmark = pytest.mark.anyio