langchain-ai--langgraph/libs/langgraph/tests/test_tracing_interops.py

import json
import sys
import time
from collections.abc import Callable
from typing import Any, TypeVar
from unittest.mock import MagicMock

import langsmith as ls
import pytest
from langchain_core.runnables import RunnableConfig
from langchain_core.tracers import LangChainTracer
from typing_extensions import TypedDict

from langgraph.graph import StateGraph

pytestmark = pytest.mark.anyio


def _get_mock_client(**kwargs: Any) -> ls.Client:
    mock_session = MagicMock()
    return ls.Client(session=mock_session, api_key="test", **kwargs)


def _get_calls(
    mock_client: Any,
    verbs: set[str] = {"POST"},
) -> list:
    return [
        c
        for c in mock_client.session.request.mock_calls
        if c.args and c.args[0] in verbs
    ]


T = TypeVar("T")


def wait_for(
    condition: Callable[[], tuple[T, bool]],
    max_sleep_time: int = 10,
    sleep_time: int = 3,
) -> T:
    """Wait for a condition to be true."""
    start_time = time.time()
    last_e = None
    while time.time() - start_time < max_sleep_time:
        try:
            res, cond = condition()
            if cond:
                return res
        except Exception as e:
            last_e = e
            time.sleep(sleep_time)
    total_time = time.time() - start_time
    if last_e is not None:
        raise last_e
    raise ValueError(f"Callable did not return within {total_time}")


@pytest.mark.skip("This test times out in CI")
async def test_nested_tracing():
    lt_py_311 = sys.version_info < (3, 11)
    mock_client = _get_mock_client()

    class State(TypedDict):
        value: str

    @ls.traceable
    async def some_traceable(content: State):
        return await child_graph.ainvoke(content)

    async def parent_node(state: State, config: RunnableConfig) -> State:
        if lt_py_311:
            result = await some_traceable(state, langsmith_extra={"config": config})
        else:
            result = await some_traceable(state)
        return {"value": f"parent_{result['value']}"}

    async def child_node(state: State) -> State:
        return {"value": f"child_{state['value']}"}

    child_builder = StateGraph(State)
    child_builder.add_node(child_node)
    child_builder.add_edge("__start__", "child_node")
    child_graph = child_builder.compile().with_config(run_name="child_graph")

    parent_builder = StateGraph(State)
    parent_builder.add_node(parent_node)
    parent_builder.add_edge("__start__", "parent_node")
    parent_graph = parent_builder.compile()

    tracer = LangChainTracer(client=mock_client)
    result = await parent_graph.ainvoke({"value": "input"}, {"callbacks": [tracer]})

    assert result == {"value": "parent_child_input"}

    def get_posts():
        post_calls = _get_calls(mock_client, verbs={"POST"})

        posts = [p for c in post_calls for p in json.loads(c.kwargs["data"])["post"]]
        names = [p.get("name") for p in posts]
        if "child_node" in names:
            return posts, True
        return None, False

    posts = wait_for(get_posts)
    # If the callbacks weren't propagated correctly, we'd
    # end up with broken dotted_orders
    parent_run = next(data for data in posts if data["name"] == "parent_node")
    child_run = next(data for data in posts if data["name"] == "child_graph")
    traceable_run = next(data for data in posts if data["name"] == "some_traceable")

    assert child_run["dotted_order"].startswith(traceable_run["dotted_order"])
    assert traceable_run["dotted_order"].startswith(parent_run["dotted_order"])

    assert child_run["parent_run_id"] == traceable_run["id"]
    assert traceable_run["parent_run_id"] == parent_run["id"]
    assert parent_run["trace_id"] == child_run["trace_id"] == traceable_run["trace_id"]