test_tree_is_constructed() — langchain Function Reference

Architecture documentation for the test_tree_is_constructed() function in test_tracing_interops.py from the langchain codebase.

Function python LangChainCore Runnables

Entity Profile

LangChainCore→ Runnables→ test_tree_is_constructed() — langchain Function Reference

Dependency Diagram

graph TD
  cb68ce75_b00f_ceca_ba75_695037791990["test_tree_is_constructed()"]
  0944f1ae_0eea_e3a0_544b_8617adb53dbc["test_tracing_interops.py"]
  cb68ce75_b00f_ceca_ba75_695037791990 -->|defined in| 0944f1ae_0eea_e3a0_544b_8617adb53dbc
  style cb68ce75_b00f_ceca_ba75_695037791990 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_tracing_interops.py lines 436–510

def test_tree_is_constructed(parent_type: Literal["ls", "lc"]) -> None:
    mock_session = MagicMock()
    mock_client_ = Client(
        session=mock_session, api_key="test", auto_batch_tracing=False
    )
    grandchild_run = None
    kitten_run = None

    @traceable
    def kitten(x: str) -> str:
        nonlocal kitten_run
        kitten_run = get_current_run_tree()
        return x

    @RunnableLambda
    def grandchild(x: str) -> str:
        nonlocal grandchild_run
        grandchild_run = get_current_run_tree()
        return kitten(x)

    @RunnableLambda
    def child(x: str) -> str:
        return grandchild.invoke(x)

    rid = uuid.uuid4()
    with tracing_context(
        client=mock_client_,
        enabled=True,
        metadata={"some_foo": "some_bar"},
        tags=["afoo"],
    ):
        collected: dict[str, RunTree] = {}

        def collect_run(run: RunTree) -> None:
            collected[str(run.id)] = run

        if parent_type == "ls":

            @traceable
            def parent() -> str:
                return child.invoke("foo")

            assert (
                parent(langsmith_extra={"on_end": collect_run, "run_id": rid}) == "foo"
            )
            assert collected

        else:

            @RunnableLambda
            def parent(_: Any) -> str:
                return child.invoke("foo")

            tracer = LangChainTracer()
            tracer._persist_run = collect_run  # type: ignore[method-assign]

            assert parent.invoke(..., {"run_id": rid, "callbacks": [tracer]}) == "foo"  # type: ignore[attr-defined]
    run = collected.get(str(rid))

    assert run is not None
    assert run.name == "parent"
    assert run.child_runs
    child_run = run.child_runs[0]
    assert child_run.name == "child"
    assert isinstance(grandchild_run, RunTree)
    assert grandchild_run.name == "grandchild"
    assert grandchild_run.metadata.get("some_foo") == "some_bar"
    assert "afoo" in grandchild_run.tags  # type: ignore[operator]
    assert isinstance(kitten_run, RunTree)
    assert kitten_run.name == "kitten"
    assert not kitten_run.child_runs
    assert kitten_run.metadata.get("some_foo") == "some_bar"
    assert "afoo" in kitten_run.tags  # type: ignore[operator]
    assert grandchild_run is not None
    assert kitten_run.dotted_order.startswith(grandchild_run.dotted_order)