test_listeners_async() — langchain Function Reference

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

Function python LangChainCore Runnables

Entity Profile

LangChainCore→ Runnables→ test_listeners_async() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable.py lines 5605–5630

async def test_listeners_async() -> 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"}]
    await chain.ainvoke(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."