test_listeners() — langchain Function Reference

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

Function python LangChainCore Runnables calls 1

Entity Profile

LangChainCore→ Runnables→ test_listeners() — langchain Function Reference

Dependency Diagram

graph TD
  033e4cdf_ae1b_97c1_e1ac_50d22d659021["test_listeners()"]
  26df6ad8_0189_51d0_c3c1_6c3248893ff5["test_runnable.py"]
  033e4cdf_ae1b_97c1_e1ac_50d22d659021 -->|defined in| 26df6ad8_0189_51d0_c3c1_6c3248893ff5
  fb618d44_c03b_ea8b_385b_2278dfb173d4["invoke()"]
  033e4cdf_ae1b_97c1_e1ac_50d22d659021 -->|calls| fb618d44_c03b_ea8b_385b_2278dfb173d4
  style 033e4cdf_ae1b_97c1_e1ac_50d22d659021 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable.py lines 5578–5602

def test_listeners() -> None:
    def fake_chain(inputs: dict[str, str]) -> dict[str, str]:
        return {**inputs, "key": "extra"}

    shared_state = {}
    value1 = {"inputs": {"name": "one"}, "outputs": {"name": "one"}}
    value2 = {"inputs": {"name": "two"}, "outputs": {"name": "two"}}

    def on_start(run: Run) -> None:
        shared_state[run.id] = {"inputs": run.inputs}

    def on_end(run: Run) -> None:
        shared_state[run.id]["outputs"] = run.inputs

    chain = (
        RunnableLambda(fake_chain)
        .with_listeners(on_end=on_end, on_start=on_start)
        .map()
    )

    data = [{"name": "one"}, {"name": "two"}]
    chain.invoke(data, config={"max_concurrency": 1})
    assert len(shared_state) == 2
    assert value1 in shared_state.values(), "Value not found in the dictionary."
    assert value2 in shared_state.values(), "Value not found in the dictionary."