Home / Function/ test_chain_ordering() — langchain Function Reference

test_chain_ordering() — langchain Function Reference

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

Entity Profile

Dependency Diagram

graph TD
  fa4fc5d4_750c_9dac_3690_f69ec02f9886["test_chain_ordering()"]
  8ff41f3c_f250_f8de_8094_4f24860a10e0["test_runnable_events_v1.py"]
  fa4fc5d4_750c_9dac_3690_f69ec02f9886 -->|defined in| 8ff41f3c_f250_f8de_8094_4f24860a10e0
  a14aef27_93bc_0ce1_f208_f5841ae9507f["_with_nulled_run_id()"]
  fa4fc5d4_750c_9dac_3690_f69ec02f9886 -->|calls| a14aef27_93bc_0ce1_f208_f5841ae9507f
  6ebe7fde_6e8f_dae5_d42f_9cea181617f5["_assert_events_equal_allow_superset_metadata()"]
  fa4fc5d4_750c_9dac_3690_f69ec02f9886 -->|calls| 6ebe7fde_6e8f_dae5_d42f_9cea181617f5
  style fa4fc5d4_750c_9dac_3690_f69ec02f9886 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v1.py lines 1528–1638

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="v1")

    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"],

Domain

Subdomains

Frequently Asked Questions

What does test_chain_ordering() do?
test_chain_ordering() is a function in the langchain codebase, defined in libs/core/tests/unit_tests/runnables/test_runnable_events_v1.py.
Where is test_chain_ordering() defined?
test_chain_ordering() is defined in libs/core/tests/unit_tests/runnables/test_runnable_events_v1.py at line 1528.
What does test_chain_ordering() call?
test_chain_ordering() calls 2 function(s): _assert_events_equal_allow_superset_metadata, _with_nulled_run_id.

Analyze Your Own Codebase

Get architecture documentation, dependency graphs, and domain analysis for your codebase in minutes.

Try Supermodel Free