test_router_runnable() — langchain Function Reference

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

Function python LangChainCore Runnables calls 2

Entity Profile

LangChainCore→ Runnables→ test_router_runnable() — langchain Function Reference

Dependency Diagram

graph TD
  eb2f915f_755e_dd2c_1ba1_afcfb20dd5be["test_router_runnable()"]
  26df6ad8_0189_51d0_c3c1_6c3248893ff5["test_runnable.py"]
  eb2f915f_755e_dd2c_1ba1_afcfb20dd5be -->|defined in| 26df6ad8_0189_51d0_c3c1_6c3248893ff5
  593bb72a_4695_6c93_b95c_e277aca006ae["batch()"]
  eb2f915f_755e_dd2c_1ba1_afcfb20dd5be -->|calls| 593bb72a_4695_6c93_b95c_e277aca006ae
  a3895c50_dab4_0c36_2d24_f63121e198a0["invoke()"]
  eb2f915f_755e_dd2c_1ba1_afcfb20dd5be -->|calls| a3895c50_dab4_0c36_2d24_f63121e198a0
  style eb2f915f_755e_dd2c_1ba1_afcfb20dd5be fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable.py lines 2865–2906

def test_router_runnable(mocker: MockerFixture, snapshot: SnapshotAssertion) -> None:
    chain1 = ChatPromptTemplate.from_template(
        "You are a math genius. Answer the question: {question}"
    ) | FakeListLLM(responses=["4"])
    chain2 = ChatPromptTemplate.from_template(
        "You are an english major. Answer the question: {question}"
    ) | FakeListLLM(responses=["2"])
    router = RouterRunnable({"math": chain1, "english": chain2})
    chain: Runnable = {
        "key": lambda x: x["key"],
        "input": {"question": lambda x: x["question"]},
    } | router
    assert dumps(chain, pretty=True) == snapshot

    result = chain.invoke({"key": "math", "question": "2 + 2"})
    assert result == "4"

    result2 = chain.batch(
        [
            {"key": "math", "question": "2 + 2"},
            {"key": "english", "question": "2 + 2"},
        ]
    )
    assert result2 == ["4", "2"]

    # Test invoke
    router_spy = mocker.spy(router.__class__, "invoke")
    tracer = FakeTracer()
    assert (
        chain.invoke({"key": "math", "question": "2 + 2"}, {"callbacks": [tracer]})
        == "4"
    )
    assert router_spy.call_args.args[1] == {
        "key": "math",
        "input": {"question": "2 + 2"},
    }
    assert len([r for r in tracer.runs if r.parent_run_id is None]) == 1
    parent_run = next(r for r in tracer.runs if r.parent_run_id is None)
    assert len(parent_run.child_runs) == 2
    router_run = parent_run.child_runs[1]
    assert router_run.name == "RunnableSequence"  # TODO: should be RunnableRouter
    assert len(router_run.child_runs) == 2