test_runnable_iter_context_config() — langchain Function Reference

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

Function python LangChainCore Runnables calls 2

Entity Profile

LangChainCore→ Runnables→ test_runnable_iter_context_config() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable.py lines 5062–5115

def test_runnable_iter_context_config() -> None:
    """Test generator runnable config propagation.

    Test that a generator can call other runnables with config
    propagated from the context.
    """
    fake = RunnableLambda(len)

    @chain
    def gen(value: str) -> Iterator[int]:
        yield fake.invoke(value)
        yield fake.invoke(value * 2)
        yield fake.invoke(value * 3)

    assert gen.get_input_jsonschema() == {
        "title": "gen_input",
        "type": "string",
    }
    assert gen.get_output_jsonschema() == {
        "title": "gen_output",
        "type": "integer",
    }

    tracer = FakeTracer()
    assert gen.invoke("a", {"callbacks": [tracer]}) == 6
    assert len(tracer.runs) == 1
    assert tracer.runs[0].outputs == {"output": 6}
    assert len(tracer.runs[0].child_runs) == 3
    assert [r.inputs["input"] for r in tracer.runs[0].child_runs] == ["a", "aa", "aaa"]
    assert [(r.outputs or {})["output"] for r in tracer.runs[0].child_runs] == [1, 2, 3]
    tracer.runs.clear()

    assert list(gen.stream("a")) == [1, 2, 3]
    assert len(tracer.runs) == 0, "callbacks doesn't persist from previous call"

    tracer = FakeTracer()
    assert list(gen.stream("a", {"callbacks": [tracer]})) == [1, 2, 3]
    assert len(tracer.runs) == 1
    assert tracer.runs[0].outputs == {"output": 6}
    assert len(tracer.runs[0].child_runs) == 3
    assert [r.inputs["input"] for r in tracer.runs[0].child_runs] == ["a", "aa", "aaa"]
    assert [(r.outputs or {})["output"] for r in tracer.runs[0].child_runs] == [1, 2, 3]

    tracer = FakeTracer()
    assert gen.batch(["a", "a"], {"callbacks": [tracer]}) == [6, 6]
    assert len(tracer.runs) == 2
    assert tracer.runs[0].outputs == {"output": 6}
    assert tracer.runs[1].outputs == {"output": 6}
    assert len(tracer.runs[0].child_runs) == 3
    assert [r.inputs["input"] for r in tracer.runs[0].child_runs] == ["a", "aa", "aaa"]
    assert [(r.outputs or {})["output"] for r in tracer.runs[0].child_runs] == [1, 2, 3]
    assert len(tracer.runs[1].child_runs) == 3
    assert [r.inputs["input"] for r in tracer.runs[1].child_runs] == ["a", "aa", "aaa"]
    assert [(r.outputs or {})["output"] for r in tracer.runs[1].child_runs] == [1, 2, 3]