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

test_event_stream_with_simple_chain() — langchain Function Reference

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

Function python LangChainCore Runnables calls 1

Entity Profile

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py lines 855–1082

async def test_event_stream_with_simple_chain() -> None:
    """Test as event stream."""
    template = ChatPromptTemplate.from_messages(
        [
            ("system", "You are Cat Agent 007"),
            ("human", "{question}"),
        ]
    ).with_config({"run_name": "my_template", "tags": ["my_template"]})

    infinite_cycle = cycle(
        [
            AIMessage(content="hello world!", id="ai1"),
            AIMessage(content="goodbye world!", id="ai2"),
        ]
    )
    # When streaming GenericFakeChatModel breaks AIMessage into chunks based on spaces
    model = (
        GenericFakeChatModel(messages=infinite_cycle)
        .with_config(
            {
                "metadata": {"a": "b"},
                "tags": ["my_model"],
                "run_name": "my_model",
            }
        )
        .bind(stop="<stop_token>")
    )

    chain = (template | model).with_config(
        {
            "metadata": {"foo": "bar"},
            "tags": ["my_chain"],
            "run_name": "my_chain",
        }
    )

    events = await _collect_events(
        chain.astream_events({"question": "hello"}, version="v2")
    )
    _assert_events_equal_allow_superset_metadata(
        events,
        [
            {
                "data": {"input": {"question": "hello"}},
                "event": "on_chain_start",
                "metadata": {"foo": "bar"},
                "name": "my_chain",
                "run_id": "",
                "parent_ids": [],
                "tags": ["my_chain"],
            },
            {
                "data": {"input": {"question": "hello"}},
                "event": "on_prompt_start",
                "metadata": {"foo": "bar"},
                "name": "my_template",
                "run_id": "",
                "parent_ids": [],
                "tags": ["my_chain", "my_template", "seq:step:1"],
            },
            {
                "data": {
                    "input": {"question": "hello"},
                    "output": ChatPromptValue(
                        messages=[
                            SystemMessage(content="You are Cat Agent 007"),
                            HumanMessage(content="hello"),
                        ]
                    ),
                },
                "event": "on_prompt_end",
                "metadata": {"foo": "bar"},
                "name": "my_template",
                "run_id": "",
                "parent_ids": [],
                "tags": ["my_chain", "my_template", "seq:step:1"],
            },
            {
                "data": {
                    "input": {
                        "messages": [

Domain

Subdomains

Defined In

Calls

Source

Frequently Asked Questions

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

Analyze Your Own Codebase

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

Try Supermodel Free