langchain Architecture Docs
Home / Function/ test_astream_events_from_custom_runnable() — langchain Function Reference

test_astream_events_from_custom_runnable() — langchain Function Reference

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

Function python LangChainCore Runnables calls 2

Entity Profile

Dependency Diagram

graph TD
  20a2cdd2_25f8_e563_51d8_eecf2fc98b01["test_astream_events_from_custom_runnable()"]
  33c02978_2077_5819_7048_bc2a81e80625["test_runnable_events_v2.py"]
  20a2cdd2_25f8_e563_51d8_eecf2fc98b01 -->|defined in| 33c02978_2077_5819_7048_bc2a81e80625
  858b85ba_6d9e_d209_9eb5_2469bad0adca["astream()"]
  20a2cdd2_25f8_e563_51d8_eecf2fc98b01 -->|calls| 858b85ba_6d9e_d209_9eb5_2469bad0adca
  716d2a5e_dc8e_3cae_e044_b56b06bee655["_collect_events()"]
  20a2cdd2_25f8_e563_51d8_eecf2fc98b01 -->|calls| 716d2a5e_dc8e_3cae_e044_b56b06bee655
  style 20a2cdd2_25f8_e563_51d8_eecf2fc98b01 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py lines 2185–2241

async def test_astream_events_from_custom_runnable() -> None:
    """Test astream events from a custom runnable."""
    iterator = ["1", "2", "3"]
    runnable = StreamingRunnable(iterator)
    chunks = [chunk async for chunk in runnable.astream(1, version="v2")]
    assert chunks == ["1", "2", "3"]
    events = await _collect_events(runnable.astream_events(1, version="v2"))
    _assert_events_equal_allow_superset_metadata(
        events,
        [
            {
                "data": {"input": 1},
                "event": "on_chain_start",
                "metadata": {},
                "name": "StreamingRunnable",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
            {
                "data": {"chunk": "1"},
                "event": "on_chain_stream",
                "metadata": {},
                "name": "StreamingRunnable",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
            {
                "data": {"chunk": "2"},
                "event": "on_chain_stream",
                "metadata": {},
                "name": "StreamingRunnable",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
            {
                "data": {"chunk": "3"},
                "event": "on_chain_stream",
                "metadata": {},
                "name": "StreamingRunnable",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
            {
                "data": {"output": "123"},
                "event": "on_chain_end",
                "metadata": {},
                "name": "StreamingRunnable",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
        ],
    )

Domain

Subdomains

Defined In

Calls

Source

Frequently Asked Questions

What does test_astream_events_from_custom_runnable() do?
test_astream_events_from_custom_runnable() is a function in the langchain codebase, defined in libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py.
Where is test_astream_events_from_custom_runnable() defined?
test_astream_events_from_custom_runnable() is defined in libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py at line 2185.
What does test_astream_events_from_custom_runnable() call?
test_astream_events_from_custom_runnable() calls 2 function(s): _collect_events, astream.

Analyze Your Own Codebase

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

Try Supermodel Free