test_runnable_typed_dict_schema() — langchain Function Reference

Architecture documentation for the test_runnable_typed_dict_schema() function in test_runnable.py from the langchain codebase.

Function python LangChainCore Runnables

Entity Profile

LangChainCore→ Runnables→ test_runnable_typed_dict_schema() — langchain Function Reference

Dependency Diagram

graph TD
  512dc34f_b59b_40a5_a0bd_138144875864["test_runnable_typed_dict_schema()"]
  26df6ad8_0189_51d0_c3c1_6c3248893ff5["test_runnable.py"]
  512dc34f_b59b_40a5_a0bd_138144875864 -->|defined in| 26df6ad8_0189_51d0_c3c1_6c3248893ff5
  style 512dc34f_b59b_40a5_a0bd_138144875864 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable.py lines 5744–5769

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.
    """

    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.model_validate({"foo": "Y", "bar": "Z"}))
        == "RunnableParallel<foo,other>Input(root={'foo': 'Y', 'bar': 'Z'})"
    )