test_chain_ordering() — langchain Function Reference

Architecture documentation for the test_chain_ordering() function in test_runnable_events_v2.py from the langchain codebase.

Function python LangChainCore Runnables calls 1

Entity Profile

LangChainCore→ Runnables→ test_chain_ordering() — langchain Function Reference

Dependency Diagram

graph TD
  66b2db10_fe7c_963f_a210_927bdc661000["test_chain_ordering()"]
  33c02978_2077_5819_7048_bc2a81e80625["test_runnable_events_v2.py"]
  66b2db10_fe7c_963f_a210_927bdc661000 -->|defined in| 33c02978_2077_5819_7048_bc2a81e80625
  8e5f2701_6ca3_f1d9_98ad_ec0c5f691bf1["_with_nulled_run_id()"]
  66b2db10_fe7c_963f_a210_927bdc661000 -->|calls| 8e5f2701_6ca3_f1d9_98ad_ec0c5f691bf1
  style 66b2db10_fe7c_963f_a210_927bdc661000 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py lines 1487–1597

async def test_chain_ordering() -> None:
    """Test the event stream with a tool."""

    def foo(a: str) -> str:
        return a

    def bar(a: str) -> str:
        return a

    chain = RunnableLambda(foo) | RunnableLambda(bar)
    iterable = chain.astream_events("q", version="v2")

    events = []

    try:
        for _ in range(10):
            next_chunk = await anext(iterable)
            events.append(next_chunk)
    except Exception:
        pass

    events = _with_nulled_run_id(events)
    for event in events:
        event["tags"] = sorted(event["tags"])

    _assert_events_equal_allow_superset_metadata(
        events,
        [
            {
                "data": {"input": "q"},
                "event": "on_chain_start",
                "metadata": {},
                "name": "RunnableSequence",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
            {
                "data": {},
                "event": "on_chain_start",
                "metadata": {},
                "name": "foo",
                "run_id": "",
                "parent_ids": [],
                "tags": ["seq:step:1"],
            },
            {
                "data": {"chunk": "q"},
                "event": "on_chain_stream",
                "metadata": {},
                "name": "foo",
                "run_id": "",
                "parent_ids": [],
                "tags": ["seq:step:1"],
            },
            {
                "data": {"input": "q", "output": "q"},
                "event": "on_chain_end",
                "metadata": {},
                "name": "foo",
                "run_id": "",
                "parent_ids": [],
                "tags": ["seq:step:1"],
            },
            {
                "data": {},
                "event": "on_chain_start",
                "metadata": {},
                "name": "bar",
                "run_id": "",
                "parent_ids": [],
                "tags": ["seq:step:2"],
            },
            {
                "data": {"chunk": "q"},
                "event": "on_chain_stream",
                "metadata": {},
                "name": "bar",
                "run_id": "",
                "parent_ids": [],
                "tags": ["seq:step:2"],