factor out util

libs/langgraph/langgraph/graph/state.py

@@ -77,7 +77,7 @@ from langgraph.pregel.write import (
 from langgraph.store.base import BaseStore
 from langgraph.types import All, Checkpointer, Command, RetryPolicy
 from langgraph.utils.fields import get_field_default
-from langgraph.utils.pydantic import create_model
+from langgraph.utils.pydantic import create_model, get_update_as_tuples
 from langgraph.utils.runnable import RunnableCallable, RunnableLike, coerce_to_runnable
 
 logger = logging.getLogger(__name__)
@@ -764,32 +764,7 @@ class CompiledStateGraph(CompiledGraph):
                         updates.extend(_get_updates(i) or ())
                 return updates
             elif (t := type(input)) and get_type_hints(t):
-                # Pydantic v1
-                if isinstance(input, BaseModelV1):
-                    keep: Optional[set[str]] = input.__fields_set__
-                    defaults = {k: v.default for k, v in t.__fields__.items()}
-                # Pydantic v2
-                elif isinstance(input, BaseModel):
-                    keep = input.model_fields_set
-                    defaults = {k: v.default for k, v in input.model_fields.items()}
-                else:
-                    keep = None
-                    defaults = {}
-
-                # NOTE: This behavior for Pydantic is somewhat inelegant,
-                # but we keep around for backwards compatibility
-                # if input is a Pydantic model, only update values
-                # that are different from the default values or in the keep set
-                return [
-                    (k, value)
-                    for k in output_keys
-                    if (value := getattr(input, k, MISSING)) is not MISSING
-                    and (
-                        value is not None
-                        or defaults.get(k, MISSING) is not None
-                        or (keep is not None and k in keep)
-                    )
-                ]
+                return get_update_as_tuples(input, output_keys)
             else:
                 msg = create_error_message(
                     message=f"Expected dict, got {input}",

libs/langgraph/langgraph/types.py

@@ -20,11 +20,10 @@ from typing import (
 )
 
 from langchain_core.runnables import Runnable, RunnableConfig
-from pydantic import BaseModel
-from pydantic.v1 import BaseModel as BaseModelV1
 from typing_extensions import Self
 
 from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointMetadata
+from langgraph.utils.pydantic import get_update_as_tuples
 
 if TYPE_CHECKING:
     from langgraph.pregel.protocol import PregelProtocol
@@ -71,11 +70,6 @@ else:
     _DC_KWARGS = {"frozen": True}
 
 
-# NOTE: this is redefined here separately from langgraph.constants
-# to avoid a circular import
-MISSING = object()
-
-
 def default_retry_on(exc: Exception) -> bool:
     import httpx
     import requests
@@ -325,28 +319,7 @@ class Command(Generic[N], ToolOutputMixin):
         ):
             return self.update
         elif hints := get_type_hints(type(self.update)):
-            # Pydantic v1
-            if isinstance(self.update, BaseModelV1):
-                keep: Optional[set[str]] = self.update.__fields_set__
-                defaults = {k: v.default for k, v in self.update.__fields__.items()}
-            # Pydantic v2
-            elif isinstance(self.update, BaseModel):
-                keep = self.update.model_fields_set
-                defaults = {k: v.default for k, v in self.update.model_fields.items()}
-            else:
-                keep = None
-                defaults = {}
-
-            return [
-                (k, value)
-                for k in hints
-                if (value := getattr(self.update, k, MISSING)) is not MISSING
-                and (
-                    value is not None
-                    or defaults.get(k, MISSING) is not None
-                    or (keep is not None and k in keep)
-                )
-            ]
+            return get_update_as_tuples(self.update, tuple(hints.keys()))
         elif self.update is not None:
             return [("__root__", self.update)]
         else:

libs/langgraph/langgraph/utils/pydantic.py

@@ -1,12 +1,16 @@
 import sys
 import typing
 from dataclasses import is_dataclass
-from typing import Any, Dict, Optional, Union
+from typing import Any, Dict, Optional, Sequence, Union
 
 import typing_extensions
 from pydantic import BaseModel
 from pydantic.v1 import BaseModel as BaseModelV1
 
+# NOTE: this is redefined here separately from langgraph.constants
+# to avoid a circular import
+MISSING = object()
+
 
 def create_model(
     model_name: str,
@@ -41,6 +45,36 @@ def create_model(
         return create_model(model_name, **v1_kwargs, **(field_definitions or {}))
 
 
+def get_update_as_tuples(input: Any, keys: Sequence[str]) -> list[tuple[str, Any]]:
+    """Get Pydantic state update as a list of (key, value) tuples."""
+    # Pydantic v1
+    if isinstance(input, BaseModelV1):
+        keep: Optional[set[str]] = input.__fields_set__
+        defaults = {k: v.default for k, v in input.__fields__.items()}
+    # Pydantic v2
+    elif isinstance(input, BaseModel):
+        keep = input.model_fields_set
+        defaults = {k: v.default for k, v in input.model_fields.items()}
+    else:
+        keep = None
+        defaults = {}
+
+    # NOTE: This behavior for Pydantic is somewhat inelegant,
+    # but we keep around for backwards compatibility
+    # if input is a Pydantic model, only update values
+    # that are different from the default values or in the keep set
+    return [
+        (k, value)
+        for k in keys
+        if (value := getattr(input, k, MISSING)) is not MISSING
+        and (
+            value is not None
+            or defaults.get(k, MISSING) is not None
+            or (keep is not None and k in keep)
+        )
+    ]
+
+
 def is_supported_by_pydantic(type_: Any) -> bool:
     """Check if a given "complex" type is supported by pydantic.