core: fix issue with runnable parallel schema being empty when children runnable input schemas use TypedDict's

libs/core/langchain_core/runnables/base.py

@@ -3593,6 +3593,11 @@ def get_input_schema(
             == "object"
             for s in self.steps__.values()
         ):
+            for step in self.steps__.values():
+                for k, v in step.get_input_schema(config).model_fields.items():
+                    if v.annotation != Any and k == "root":
+                        return super().get_input_schema(config)
+
             # This is correct, but pydantic typings/mypy don't think so.
             return create_model_v2(  # type: ignore[call-overload]
                 self.get_name("Input"),

libs/core/tests/unit_tests/runnables/test_runnable.py

@@ -5465,3 +5465,40 @@ def add_ten(x: dict[str, int]) -> dict[str, int]:
 
     result = runnable_assign.invoke({"input": 5})
     assert result == {"input": 5, "add_step": {"added": 15}}
+
+
+def test_runnable_typed_dict_schema() -> None:
+    """Testing that the schema is generated properly(not empty) when using TypedDict
+
+    subclasses to annotate the arguments of a RunnableParallel children.
+    """
+    from typing_extensions import TypedDict
+
+    from langchain_core.runnables import RunnableLambda, RunnableParallel
+
+    class Foo(TypedDict):
+        foo: str
+
+    class InputData(Foo):
+        bar: str
+
+    def forward_foo(input_data: InputData) -> str:
+        return input_data["foo"]
+
+    def transform_input(input_data: InputData) -> dict[str, str]:
+        foo = input_data["foo"]
+        bar = input_data["bar"]
+
+        return {"transformed": foo + bar}
+
+    foo_runnable = RunnableLambda(forward_foo)
+    other_runnable = RunnableLambda(transform_input)
+
+    parallel = RunnableParallel(
+        foo=foo_runnable,
+        other=other_runnable,
+    )
+    assert (
+        repr(parallel.input_schema.validate({"foo": "Y", "bar": "Z"}))
+        == "RunnableParallel<foo,other>Input(root={'foo': 'Y', 'bar': 'Z'})"
+    )