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

test_astream_with_model_in_chain() — langchain Function Reference

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

Function python LangChainCore Runnables calls 2

Entity Profile

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py lines 606–852

async def test_astream_with_model_in_chain() -> None:
    """Scenarios with model when it is not the only runnable in the chain."""
    infinite_cycle = cycle([AIMessage(content="hello world!")])
    # 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>")
    )

    @RunnableLambda
    def i_dont_stream(value: Any, config: RunnableConfig) -> Any:
        if sys.version_info >= (3, 11):
            return model.invoke(value)
        return model.invoke(value, config)

    events = await _collect_events(i_dont_stream.astream_events("hello", version="v2"))
    _assert_events_equal_allow_superset_metadata(
        events,
        [
            {
                "data": {"input": "hello"},
                "event": "on_chain_start",
                "metadata": {},
                "name": "i_dont_stream",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
            {
                "data": {"input": {"messages": [[HumanMessage(content="hello")]]}},
                "event": "on_chat_model_start",
                "metadata": {
                    "a": "b",
                    "ls_model_type": "chat",
                    "ls_stop": "<stop_token>",
                },
                "name": "my_model",
                "run_id": "",
                "parent_ids": [],
                "tags": ["my_model"],
            },
            {
                "data": {
                    "chunk": _any_id_ai_message_chunk(
                        content="hello",
                    )
                },
                "event": "on_chat_model_stream",
                "metadata": {
                    "a": "b",
                    "ls_model_type": "chat",
                    "ls_stop": "<stop_token>",
                },
                "name": "my_model",
                "run_id": "",
                "parent_ids": [],
                "tags": ["my_model"],
            },
            {
                "data": {"chunk": _any_id_ai_message_chunk(content=" ")},
                "event": "on_chat_model_stream",
                "metadata": {
                    "a": "b",
                    "ls_model_type": "chat",
                    "ls_stop": "<stop_token>",
                },
                "name": "my_model",
                "run_id": "",
                "parent_ids": [],
                "tags": ["my_model"],
            },
            {
                "data": {
                    "chunk": _any_id_ai_message_chunk(

Domain

Subdomains

Defined In

Calls

Source

Frequently Asked Questions

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

Analyze Your Own Codebase

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

Try Supermodel Free