mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-09-09 11:17:53 +02:00
Cache
This commit is contained in:
William Fu-Hinthorn
@@ -945,139 +945,149 @@ def _pick_mapper(
|
||||
if isclass(schema):
|
||||
if issubclass(schema, dict):
|
||||
return None
|
||||
if issubclass(schema, BaseModel):
|
||||
return partial(_coerce_state_pydantic, schema)
|
||||
if issubclass(schema, BaseModelV1):
|
||||
return partial(_coerce_state_pydantic_v1, schema)
|
||||
if issubclass(schema, (BaseModel, BaseModelV1)):
|
||||
return _SchemaCoercionMapper(schema)
|
||||
return partial(_coerce_state, schema)
|
||||
|
||||
|
||||
def _coerce_state_pydantic(
|
||||
schema: Type[Any], input_data: dict[str, Any], *, __depth__: int = 5
|
||||
) -> Any:
|
||||
if not isinstance(input_data, dict) or __depth__ <= 0:
|
||||
return input_data
|
||||
class _SchemaCoercionMapper:
|
||||
_cache: dict[tuple[Type[Any], int], "_SchemaCoercionMapper"] = {}
|
||||
|
||||
processed_input = {}
|
||||
for field_name, field_value in input_data.items():
|
||||
if field_name not in schema.model_fields:
|
||||
processed_input[field_name] = field_value
|
||||
continue
|
||||
def __new__(cls, schema: Type[Any], max_depth: int = 5) -> "_SchemaCoercionMapper":
|
||||
key = (schema, max_depth)
|
||||
if key in cls._cache:
|
||||
return cls._cache[key]
|
||||
inst = super().__new__(cls)
|
||||
cls._cache[key] = inst
|
||||
return inst
|
||||
|
||||
field_info = schema.model_fields[field_name]
|
||||
field_type = field_info.annotation
|
||||
processed_input[field_name] = _process_field_value(
|
||||
field_type, field_value, __depth__ - 1
|
||||
)
|
||||
def __init__(self, schema: Type[Any], max_depth: int = 5):
|
||||
if hasattr(self, "_inited"):
|
||||
return
|
||||
self._inited = True
|
||||
self.schema = schema
|
||||
self.max_depth = max_depth
|
||||
if hasattr(schema, "model_fields") and hasattr(schema, "model_construct"):
|
||||
self._fields = {n: f.annotation for n, f in schema.model_fields.items()}
|
||||
self._construct = schema.model_construct
|
||||
elif hasattr(schema, "__fields__") and callable(
|
||||
getattr(schema, "construct", None)
|
||||
):
|
||||
self._fields = {n: f.annotation for n, f in schema.__fields__.items()}
|
||||
self._construct = schema.construct
|
||||
else:
|
||||
raise TypeError("Schema is neither valid Pydantic v1 nor v2 model.")
|
||||
self._field_coercers: Optional[dict[str, Callable[[Any, Any], Any]]] = None
|
||||
|
||||
return schema.model_construct(**processed_input)
|
||||
def __call__(self, input_data: Any, depth: Optional[int] = None) -> Any:
|
||||
return self.coerce(input_data, depth)
|
||||
|
||||
def coerce(self, input_data: Any, depth: Optional[int] = None) -> Any:
|
||||
if depth is None:
|
||||
depth = self.max_depth
|
||||
if not isinstance(input_data, dict) or depth <= 0:
|
||||
return input_data
|
||||
processed = {}
|
||||
if self._field_coercers is None:
|
||||
self._field_coercers = {
|
||||
n: self._build_coercer(t) for n, t in self._fields.items()
|
||||
}
|
||||
for k, v in input_data.items():
|
||||
fn = self._field_coercers.get(k)
|
||||
processed[k] = fn(v, depth - 1) if fn else v
|
||||
return self._construct(**processed)
|
||||
|
||||
def _coerce_state_pydantic_v1(
|
||||
schema: Type[Any], input_data: dict[str, Any], *, __depth__: int = 5
|
||||
) -> Any:
|
||||
if not isinstance(input_data, dict) or __depth__ <= 0:
|
||||
return input_data
|
||||
|
||||
processed_input = {}
|
||||
for field_name, field_value in input_data.items():
|
||||
if field_name not in schema.__fields__:
|
||||
processed_input[field_name] = field_value
|
||||
continue
|
||||
|
||||
field_info = schema.__fields__[field_name]
|
||||
field_type = field_info.annotation
|
||||
processed_input[field_name] = _process_field_value(
|
||||
field_type, field_value, __depth__ - 1
|
||||
)
|
||||
|
||||
return schema.construct(**processed_input)
|
||||
|
||||
|
||||
def _process_field_value(
|
||||
field_type: Type[Any], field_value: Any, __depth__: int
|
||||
) -> Any:
|
||||
if __depth__ <= 0 or field_value is None:
|
||||
return field_value
|
||||
origin = get_origin(field_type)
|
||||
|
||||
if origin is Annotated:
|
||||
real_type, *_ = get_args(field_type)
|
||||
res = _process_field_value(real_type, field_value, __depth__)
|
||||
return res
|
||||
|
||||
if isclass(field_type):
|
||||
is_class_ = True
|
||||
try:
|
||||
is_model = issubclass(field_type, BaseModel)
|
||||
except TypeError:
|
||||
is_class_ = False
|
||||
is_model = False
|
||||
if is_model:
|
||||
if isinstance(field_value, dict):
|
||||
return _coerce_state_pydantic(
|
||||
field_type, field_value, __depth__=__depth__
|
||||
)
|
||||
return field_value
|
||||
if is_class_ and issubclass(field_type, BaseModelV1):
|
||||
if isinstance(field_value, dict):
|
||||
return _coerce_state_pydantic_v1(
|
||||
field_type, field_value, __depth__=__depth__
|
||||
)
|
||||
return field_value
|
||||
|
||||
if origin is list or field_type is list:
|
||||
if not isinstance(field_value, (list, tuple)):
|
||||
raise TypeError(
|
||||
f"Expected a list/tuple for {field_type}, got {type(field_value)}."
|
||||
)
|
||||
(item_type,) = get_args(field_type)
|
||||
return [
|
||||
_process_field_value(item_type, item, __depth__ - 1) for item in field_value
|
||||
]
|
||||
|
||||
if origin is dict or field_type is dict:
|
||||
if not isinstance(field_value, dict):
|
||||
raise TypeError(
|
||||
f"Expected a dict for {field_type}, got {type(field_value)}."
|
||||
)
|
||||
key_type, val_type = get_args(field_type)
|
||||
return {
|
||||
_process_field_value(key_type, k, __depth__ - 1): _process_field_value(
|
||||
val_type, v, __depth__ - 1
|
||||
)
|
||||
for k, v in field_value.items()
|
||||
}
|
||||
|
||||
if origin is tuple:
|
||||
if not isinstance(field_value, (list, tuple)):
|
||||
raise TypeError(
|
||||
f"Expected a tuple/list for {field_type}, got {type(field_value)}."
|
||||
)
|
||||
args = get_args(field_type)
|
||||
# Handle tuple[type1, type2, ...] with fixed length and different types
|
||||
result = []
|
||||
for i, arg in enumerate(args):
|
||||
if i < len(field_value):
|
||||
result.append(_process_field_value(arg, field_value[i], __depth__ - 1))
|
||||
else:
|
||||
# If field_value is shorter than expected, use None for remaining positions
|
||||
result.append(None)
|
||||
# If field_value is longer than expected, truncate it
|
||||
return tuple(result)
|
||||
|
||||
if origin is Union:
|
||||
for arg in get_args(field_type):
|
||||
if arg is type(None):
|
||||
# e.g. Optional
|
||||
continue
|
||||
def _build_coercer(self, field_type: Any) -> Callable[[Any, Any], Any]:
|
||||
origin = get_origin(field_type)
|
||||
if origin is Annotated:
|
||||
real_type, *_ = get_args(field_type)
|
||||
sub = self._build_coercer(real_type)
|
||||
return lambda v, d: sub(v, d)
|
||||
if isclass(field_type):
|
||||
is_class_ = True
|
||||
try:
|
||||
result = _process_field_value(arg, field_value, __depth__ - 1)
|
||||
return result
|
||||
except Exception:
|
||||
pass # Fall back to the next union argument
|
||||
is_base_model = issubclass(field_type, BaseModel)
|
||||
except TypeError:
|
||||
is_class_ = False
|
||||
is_base_model = False
|
||||
|
||||
return field_value
|
||||
if is_base_model:
|
||||
mapper = _SchemaCoercionMapper(field_type, self.max_depth)
|
||||
return lambda v, d: mapper.coerce(v, d) if isinstance(v, dict) else v
|
||||
if is_class_ and issubclass(field_type, BaseModelV1):
|
||||
mapper = _SchemaCoercionMapper(field_type, self.max_depth)
|
||||
return lambda v, d: mapper.coerce(v, d) if isinstance(v, dict) else v
|
||||
if origin is list or field_type is list:
|
||||
args = get_args(field_type)
|
||||
if len(args) != 1:
|
||||
return lambda v, d: v
|
||||
sub = self._build_coercer(args[0])
|
||||
|
||||
def list_coercer(v: Any, d: Any) -> Any:
|
||||
if not isinstance(v, (list, tuple)):
|
||||
raise TypeError(f"Expected list, got {type(v).__name__}")
|
||||
return [sub(x, d - 1) for x in v]
|
||||
|
||||
return list_coercer
|
||||
if origin is dict or field_type is dict:
|
||||
args = get_args(field_type)
|
||||
if len(args) != 2:
|
||||
|
||||
def plain_dict_coercer(v: Any, d: Any) -> Any:
|
||||
if not isinstance(v, dict):
|
||||
raise TypeError(f"Expected dict, got {type(v).__name__}")
|
||||
return v
|
||||
|
||||
return plain_dict_coercer
|
||||
k_sub = self._build_coercer(args[0])
|
||||
v_sub = self._build_coercer(args[1])
|
||||
|
||||
def dict_coercer(v: Any, d: Any) -> Any:
|
||||
if not isinstance(v, dict):
|
||||
raise TypeError(f"Expected dict, got {type(v).__name__}")
|
||||
return {k_sub(k, d - 1): v_sub(val, d - 1) for k, val in v.items()}
|
||||
|
||||
return dict_coercer
|
||||
|
||||
if origin is tuple:
|
||||
targs = get_args(field_type)
|
||||
if not targs:
|
||||
return lambda v, d: v
|
||||
subs = [self._build_coercer(a) for a in targs]
|
||||
|
||||
def tuple_coercer(v: Any, d: Any) -> Any:
|
||||
if not isinstance(v, (list, tuple)):
|
||||
raise TypeError(f"Expected tuple-like, got {type(v).__name__}")
|
||||
out = []
|
||||
for i, sp in enumerate(subs):
|
||||
out.append(sp(v[i] if i < len(v) else None, d - 1))
|
||||
return tuple(out)
|
||||
|
||||
return tuple_coercer
|
||||
if origin is Union:
|
||||
uargs = get_args(field_type)
|
||||
subs, none_in_union = [], False
|
||||
for arg in uargs:
|
||||
if arg is type(None):
|
||||
none_in_union = True
|
||||
else:
|
||||
subs.append(self._build_coercer(arg))
|
||||
|
||||
def union_coercer(v: Any, d: Any) -> Any:
|
||||
if v is None and none_in_union:
|
||||
return None
|
||||
err = None
|
||||
for sp in subs:
|
||||
try:
|
||||
return sp(v, d - 1)
|
||||
except Exception as e:
|
||||
err = e
|
||||
if err:
|
||||
raise err
|
||||
return v
|
||||
|
||||
return union_coercer
|
||||
return lambda v, d: v
|
||||
|
||||
|
||||
def _coerce_state(schema: Type[Any], input: dict[str, Any]) -> dict[str, Any]:
|
||||
|
||||
@@ -3069,7 +3069,7 @@ def test_nested_pydantic_models(version: str) -> None:
|
||||
# Basic nested model tests
|
||||
top_level: str
|
||||
nested: NestedModel
|
||||
optional_nested: Optional[NestedModel] = None
|
||||
optional_nested: Annotated[Optional[NestedModel], lambda x, y: y, "Foo"]
|
||||
dict_nested: dict[str, NestedModel]
|
||||
list_nested: Annotated[
|
||||
Union[dict, list[dict[str, NestedModel]]], lambda x, y: (x or []) + [y]
|
||||
@@ -3126,8 +3126,10 @@ def test_nested_pydantic_models(version: str) -> None:
|
||||
|
||||
update = {"top_level": "updated", "nested": {"value": 100, "name": "updated"}}
|
||||
|
||||
expected = State(**inputs)
|
||||
|
||||
def node_fn(state: State) -> dict:
|
||||
assert state == State(**inputs)
|
||||
assert state == expected
|
||||
return update
|
||||
|
||||
builder = StateGraph(State)
|
||||
@@ -3140,6 +3142,11 @@ def test_nested_pydantic_models(version: str) -> None:
|
||||
|
||||
assert result == {**inputs, **update}
|
||||
|
||||
new_inputs = inputs.copy()
|
||||
new_inputs["list_nested"] = {"foo": "bar"}
|
||||
expected = State(**new_inputs)
|
||||
assert {**new_inputs, **update} == graph.invoke(new_inputs.copy())
|
||||
|
||||
|
||||
@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
|
||||
def test_in_one_fan_out_state_graph_waiting_edge_plus_regular(
|
||||
|
||||
Reference in New Issue
Block a user