test_events_astream_config() — langchain Function Reference

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

Function python LangChainCore Runnables calls 2

Entity Profile

LangChainCore→ Runnables→ test_events_astream_config() — langchain Function Reference

Dependency Diagram

graph TD
  b5fc02db_40d4_9b8c_c881_ac0db3a7512a["test_events_astream_config()"]
  33c02978_2077_5819_7048_bc2a81e80625["test_runnable_events_v2.py"]
  b5fc02db_40d4_9b8c_c881_ac0db3a7512a -->|defined in| 33c02978_2077_5819_7048_bc2a81e80625
  44779688_f3fe_6e36_2d50_baa2a78e6a2d["invoke()"]
  b5fc02db_40d4_9b8c_c881_ac0db3a7512a -->|calls| 44779688_f3fe_6e36_2d50_baa2a78e6a2d
  716d2a5e_dc8e_3cae_e044_b56b06bee655["_collect_events()"]
  b5fc02db_40d4_9b8c_c881_ac0db3a7512a -->|calls| 716d2a5e_dc8e_3cae_e044_b56b06bee655
  style b5fc02db_40d4_9b8c_c881_ac0db3a7512a fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py lines 1824–1902

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="v2"))
    _assert_events_equal_allow_superset_metadata(
        events,
        [
            {
                "data": {"input": "hello"},
                "event": "on_chat_model_start",
                "metadata": {"ls_model_type": "chat"},
                "name": "GenericFakeChatModel",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
            {
                "data": {
                    "chunk": AIMessageChunk(
                        content="Goodbye",
                        id="ai2",
                    )
                },
                "event": "on_chat_model_stream",
                "metadata": {"ls_model_type": "chat"},
                "name": "GenericFakeChatModel",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
            {
                "data": {"chunk": AIMessageChunk(content=" ", id="ai2")},
                "event": "on_chat_model_stream",
                "metadata": {"ls_model_type": "chat"},
                "name": "GenericFakeChatModel",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
            {
                "data": {
                    "chunk": AIMessageChunk(
                        content="world", id="ai2", chunk_position="last"
                    )
                },
                "event": "on_chat_model_stream",
                "metadata": {"ls_model_type": "chat"},
                "name": "GenericFakeChatModel",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
            {
                "data": {
                    "output": AIMessageChunk(
                        content="Goodbye world", id="ai2", chunk_position="last"
                    ),
                },
                "event": "on_chat_model_end",
                "metadata": {"ls_model_type": "chat"},
                "name": "GenericFakeChatModel",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
        ],
    )