From b4f11929f80d3ffa8e81b583fed4197159cd30f8 Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Tue, 10 Dec 2024 15:24:56 +0100 Subject: [PATCH 1/4] fix(config): extract default values, description from pydantic models, typeddict and dataclass --- libs/langgraph/langgraph/pregel/__init__.py | 50 ++++++++++++++++++++- 1 file changed, 48 insertions(+), 2 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index e714afe21..2cd7f0984 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -11,6 +11,7 @@ from typing import ( AsyncIterator, Callable, Dict, + Generator, Iterator, Mapping, Optional, @@ -308,6 +309,44 @@ class Pregel(PregelProtocol): @property def config_specs(self) -> list[ConfigurableFieldSpec]: + # TODO: shouldn't this be in langchain_core? + def get_enhanced_type_hints( + type: Type[Any], + ) -> Generator[tuple[str, Any, Any, Optional[str]], None]: + """Attempt to extract default values and descriptions from provided config spec""" + for name, typ in get_type_hints(type).items(): + default = None + description = None + + # Pydantic models + try: + if hasattr(type, "__fields__") and name in type.__fields__: + field = type.__fields__[name] + + if ( + hasattr(field, "description") + and field.description is not None + ): + description = field.description + + if hasattr(field, "default") and field.default is not None: + default = field.default + + except (AttributeError, KeyError, TypeError): + pass + + # TypedDict, dataclass + try: + if hasattr(type, "__dict__"): + type_dict = getattr(type, "__dict__") + + if name in type_dict: + default = type_dict[name] + except (AttributeError, KeyError, TypeError): + pass + + yield name, typ, default, description + return [ spec for spec in get_unique_config_specs( @@ -319,8 +358,15 @@ class Pregel(PregelProtocol): ) + ( [ - ConfigurableFieldSpec(id=name, annotation=typ) - for name, typ in get_type_hints(self.config_type).items() + ConfigurableFieldSpec( + id=name, + annotation=typ, + default=default, + description=description, + ) + for name, typ, default, description in get_enhanced_type_hints( + self.config_type + ) ] if self.config_type is not None else [] From 1f68bd0d83b77d715f360daacb245e5b7c3506aa Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Tue, 10 Dec 2024 16:49:05 +0100 Subject: [PATCH 2/4] Move to langgraph.utils.fields --- libs/langgraph/langgraph/pregel/__init__.py | 41 +-------------------- libs/langgraph/langgraph/pregel/utils.py | 2 +- libs/langgraph/langgraph/utils/fields.py | 37 ++++++++++++++++++- 3 files changed, 38 insertions(+), 42 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index 2cd7f0984..ba4533830 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -11,7 +11,6 @@ from typing import ( AsyncIterator, Callable, Dict, - Generator, Iterator, Mapping, Optional, @@ -19,7 +18,6 @@ from typing import ( Type, Union, cast, - get_type_hints, overload, ) from uuid import UUID, uuid5 @@ -118,6 +116,7 @@ from langgraph.utils.config import ( patch_config, patch_configurable, ) +from langgraph.utils.fields import get_enhanced_type_hints from langgraph.utils.pydantic import create_model from langgraph.utils.queue import AsyncQueue, SyncQueue # type: ignore[attr-defined] @@ -309,44 +308,6 @@ class Pregel(PregelProtocol): @property def config_specs(self) -> list[ConfigurableFieldSpec]: - # TODO: shouldn't this be in langchain_core? - def get_enhanced_type_hints( - type: Type[Any], - ) -> Generator[tuple[str, Any, Any, Optional[str]], None]: - """Attempt to extract default values and descriptions from provided config spec""" - for name, typ in get_type_hints(type).items(): - default = None - description = None - - # Pydantic models - try: - if hasattr(type, "__fields__") and name in type.__fields__: - field = type.__fields__[name] - - if ( - hasattr(field, "description") - and field.description is not None - ): - description = field.description - - if hasattr(field, "default") and field.default is not None: - default = field.default - - except (AttributeError, KeyError, TypeError): - pass - - # TypedDict, dataclass - try: - if hasattr(type, "__dict__"): - type_dict = getattr(type, "__dict__") - - if name in type_dict: - default = type_dict[name] - except (AttributeError, KeyError, TypeError): - pass - - yield name, typ, default, description - return [ spec for spec in get_unique_config_specs( diff --git a/libs/langgraph/langgraph/pregel/utils.py b/libs/langgraph/langgraph/pregel/utils.py index 66464ef9a..0c7030bb0 100644 --- a/libs/langgraph/langgraph/pregel/utils.py +++ b/libs/langgraph/langgraph/pregel/utils.py @@ -1,4 +1,4 @@ -from typing import Optional +from typing import Any, Generator, Optional, Type, get_type_hints from langchain_core.runnables import RunnableLambda, RunnableSequence from langchain_core.runnables.utils import get_function_nonlocals diff --git a/libs/langgraph/langgraph/utils/fields.py b/libs/langgraph/langgraph/utils/fields.py index f4786cb34..503d4c2d4 100644 --- a/libs/langgraph/langgraph/utils/fields.py +++ b/libs/langgraph/langgraph/utils/fields.py @@ -1,5 +1,5 @@ import dataclasses -from typing import Any, Optional, Type, Union +from typing import Any, Generator, Optional, Type, Union, get_type_hints from typing_extensions import Annotated, NotRequired, ReadOnly, Required, get_origin @@ -106,3 +106,38 @@ def get_field_default(name: str, type_: Any, schema: Type[Any]) -> Any: if _is_optional_type(type_): return None return ... + + +def get_enhanced_type_hints( + type: Type[Any], +) -> Generator[tuple[str, Any, Any, Optional[str]], None]: + """Attempt to extract default values and descriptions from provided type, used for config schema.""" + for name, typ in get_type_hints(type).items(): + default = None + description = None + + # Pydantic models + try: + if hasattr(type, "__fields__") and name in type.__fields__: + field = type.__fields__[name] + + if hasattr(field, "description") and field.description is not None: + description = field.description + + if hasattr(field, "default") and field.default is not None: + default = field.default + + except (AttributeError, KeyError, TypeError): + pass + + # TypedDict, dataclass + try: + if hasattr(type, "__dict__"): + type_dict = getattr(type, "__dict__") + + if name in type_dict: + default = type_dict[name] + except (AttributeError, KeyError, TypeError): + pass + + yield name, typ, default, description From 5a30fc6a871da2a4c5b16fb1f6fb8b7d3aa7eb73 Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Tue, 10 Dec 2024 17:06:37 +0100 Subject: [PATCH 3/4] Handle PydanticUndefined, add tests --- libs/langgraph/langgraph/utils/fields.py | 8 +++- libs/langgraph/tests/test_utils.py | 60 +++++++++++++++++++++++- 2 files changed, 66 insertions(+), 2 deletions(-) diff --git a/libs/langgraph/langgraph/utils/fields.py b/libs/langgraph/langgraph/utils/fields.py index 503d4c2d4..009e5aee1 100644 --- a/libs/langgraph/langgraph/utils/fields.py +++ b/libs/langgraph/langgraph/utils/fields.py @@ -110,7 +110,7 @@ def get_field_default(name: str, type_: Any, schema: Type[Any]) -> Any: def get_enhanced_type_hints( type: Type[Any], -) -> Generator[tuple[str, Any, Any, Optional[str]], None]: +) -> Generator[tuple[str, Any, Any, Optional[str]], None, None]: """Attempt to extract default values and descriptions from provided type, used for config schema.""" for name, typ in get_type_hints(type).items(): default = None @@ -126,6 +126,12 @@ def get_enhanced_type_hints( if hasattr(field, "default") and field.default is not None: default = field.default + if ( + hasattr(default, "__class__") + and getattr(default.__class__, "__name__", "") + == "PydanticUndefinedType" + ): + default = None except (AttributeError, KeyError, TypeError): pass diff --git a/libs/langgraph/tests/test_utils.py b/libs/langgraph/tests/test_utils.py index e8ea94fff..616f1a78f 100644 --- a/libs/langgraph/tests/test_utils.py +++ b/libs/langgraph/tests/test_utils.py @@ -21,7 +21,11 @@ from typing_extensions import Annotated, NotRequired, Required from langgraph.graph import END, StateGraph from langgraph.graph.graph import CompiledGraph -from langgraph.utils.fields import _is_optional_type, get_field_default +from langgraph.utils.fields import ( + _is_optional_type, + get_enhanced_type_hints, + get_field_default, +) from langgraph.utils.runnable import is_async_callable, is_async_generator pytestmark = pytest.mark.anyio @@ -227,3 +231,57 @@ def test_is_required(): assert get_field_default("val_12", gcannos["val_12"], MyGrandChildDict) is None assert get_field_default("val_9", gcannos["val_9"], MyGrandChildDict) is None assert get_field_default("val_13", gcannos["val_13"], MyGrandChildDict) == ... + + +def test_enhanced_type_hints() -> None: + from dataclasses import dataclass + from typing import Annotated + + from pydantic import BaseModel, Field + + class MyTypedDict(TypedDict): + val_1: str + val_2: int = 42 + val_3: str = "default" + + hints = list(get_enhanced_type_hints(MyTypedDict)) + assert len(hints) == 3 + assert hints[0] == ("val_1", str, None, None) + assert hints[1] == ("val_2", int, 42, None) + assert hints[2] == ("val_3", str, "default", None) + + @dataclass + class MyDataclass: + val_1: str + val_2: int = 42 + val_3: str = "default" + + hints = list(get_enhanced_type_hints(MyDataclass)) + assert len(hints) == 3 + assert hints[0] == ("val_1", str, None, None) + assert hints[1] == ("val_2", int, 42, None) + assert hints[2] == ("val_3", str, "default", None) + + class MyPydanticModel(BaseModel): + val_1: str + val_2: int = 42 + val_3: str = Field(default="default", description="A description") + + hints = list(get_enhanced_type_hints(MyPydanticModel)) + assert len(hints) == 3 + assert hints[0] == ("val_1", str, None, None) + assert hints[1] == ("val_2", int, 42, None) + assert hints[2] == ("val_3", str, "default", "A description") + + class MyPydanticModelWithAnnotated(BaseModel): + val_1: Annotated[str, Field(description="A description")] + val_2: Annotated[int, Field(default=42)] + val_3: Annotated[ + str, Field(default="default", description="Another description") + ] + + hints = list(get_enhanced_type_hints(MyPydanticModelWithAnnotated)) + assert len(hints) == 3 + assert hints[0] == ("val_1", str, None, "A description") + assert hints[1] == ("val_2", int, 42, None) + assert hints[2] == ("val_3", str, "default", "Another description") From 17c1a8db46eddcc8637c5ac9fb4367e479d72f41 Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Tue, 10 Dec 2024 17:42:01 +0100 Subject: [PATCH 4/4] Fix lint --- libs/langgraph/langgraph/pregel/utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libs/langgraph/langgraph/pregel/utils.py b/libs/langgraph/langgraph/pregel/utils.py index 0c7030bb0..66464ef9a 100644 --- a/libs/langgraph/langgraph/pregel/utils.py +++ b/libs/langgraph/langgraph/pregel/utils.py @@ -1,4 +1,4 @@ -from typing import Any, Generator, Optional, Type, get_type_hints +from typing import Optional from langchain_core.runnables import RunnableLambda, RunnableSequence from langchain_core.runnables.utils import get_function_nonlocals