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

test_events_astream_config() — langchain Function Reference

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

Function python LangChainCore Runnables calls 2

Entity Profile

Dependency Diagram

graph TD
  3d788f98_8b32_89a7_b636_aa44f3c4900d["test_events_astream_config()"]
  8ff41f3c_f250_f8de_8094_4f24860a10e0["test_runnable_events_v1.py"]
  3d788f98_8b32_89a7_b636_aa44f3c4900d -->|defined in| 8ff41f3c_f250_f8de_8094_4f24860a10e0
  aac0453e_34cd_4dce_c7fa_f176ab20140b["_collect_events()"]
  3d788f98_8b32_89a7_b636_aa44f3c4900d -->|calls| aac0453e_34cd_4dce_c7fa_f176ab20140b
  6ebe7fde_6e8f_dae5_d42f_9cea181617f5["_assert_events_equal_allow_superset_metadata()"]
  3d788f98_8b32_89a7_b636_aa44f3c4900d -->|calls| 6ebe7fde_6e8f_dae5_d42f_9cea181617f5
  style 3d788f98_8b32_89a7_b636_aa44f3c4900d fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v1.py lines 1876–1954

async def test_events_astream_config() -> None:
    """Test that astream events support accepting config."""
    infinite_cycle = cycle([AIMessage(content="hello world!", id="ai1")])
    good_world_on_repeat = cycle([AIMessage(content="Goodbye world", id="ai2")])
    model = GenericFakeChatModel(messages=infinite_cycle).configurable_fields(
        messages=ConfigurableField(
            id="messages",
            name="Messages",
            description="Messages return by the LLM",
        )
    )

    model_02 = model.with_config({"configurable": {"messages": good_world_on_repeat}})
    assert model_02.invoke("hello") == AIMessage(content="Goodbye world", id="ai2")

    events = await _collect_events(model_02.astream_events("hello", version="v1"))
    _assert_events_equal_allow_superset_metadata(
        events,
        [
            {
                "data": {"input": "hello"},
                "event": "on_chat_model_start",
                "metadata": {},
                "name": "RunnableConfigurableFields",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
            {
                "data": {
                    "chunk": AIMessageChunk(
                        content="Goodbye",
                        id="ai2",
                    )
                },
                "event": "on_chat_model_stream",
                "metadata": {},
                "name": "RunnableConfigurableFields",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
            {
                "data": {"chunk": AIMessageChunk(content=" ", id="ai2")},
                "event": "on_chat_model_stream",
                "metadata": {},
                "name": "RunnableConfigurableFields",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
            {
                "data": {
                    "chunk": AIMessageChunk(
                        content="world", id="ai2", chunk_position="last"
                    )
                },
                "event": "on_chat_model_stream",
                "metadata": {},
                "name": "RunnableConfigurableFields",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
            {
                "data": {
                    "output": AIMessageChunk(
                        content="Goodbye world", id="ai2", chunk_position="last"
                    )
                },
                "event": "on_chat_model_end",
                "metadata": {},
                "name": "RunnableConfigurableFields",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
        ],
    )

Domain

Subdomains

Defined In

Calls

Source

Frequently Asked Questions

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

Analyze Your Own Codebase

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

Try Supermodel Free