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

test_event_stream_with_triple_lambda_test_filtering() — langchain Function Reference

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

Function python LangChainCore Runnables calls 1

Entity Profile

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py lines 376–462

async def test_event_stream_with_triple_lambda_test_filtering() -> None:
    """Test filtering based on tags / names."""

    def reverse(s: str) -> str:
        """Reverse a string."""
        return s[::-1]

    r = RunnableLambda(func=reverse)

    chain = (
        r.with_config({"run_name": "1"})
        | r.with_config({"run_name": "2", "tags": ["my_tag"]})
        | r.with_config({"run_name": "3", "tags": ["my_tag"]})
    )
    events = await _collect_events(
        chain.astream_events("hello", include_names=["1"], version="v2")
    )
    _assert_events_equal_allow_superset_metadata(
        events,
        [
            {
                "data": {"input": "hello"},
                "event": "on_chain_start",
                "metadata": {},
                "name": "1",
                "run_id": "",
                "parent_ids": [],
                "tags": ["seq:step:1"],
            },
            {
                "data": {"chunk": "olleh"},
                "event": "on_chain_stream",
                "metadata": {},
                "name": "1",
                "run_id": "",
                "parent_ids": [],
                "tags": ["seq:step:1"],
            },
            {
                "data": {"output": "olleh"},
                "event": "on_chain_end",
                "metadata": {},
                "name": "1",
                "run_id": "",
                "parent_ids": [],
                "tags": ["seq:step:1"],
            },
        ],
    )

    events = await _collect_events(
        chain.astream_events(
            "hello", include_tags=["my_tag"], exclude_names=["2"], version="v2"
        )
    )
    _assert_events_equal_allow_superset_metadata(
        events,
        [
            {
                "data": {"input": "hello"},
                "event": "on_chain_start",
                "metadata": {},
                "name": "3",
                "run_id": "",
                "parent_ids": [],
                "tags": ["my_tag", "seq:step:3"],
            },
            {
                "data": {"chunk": "olleh"},
                "event": "on_chain_stream",
                "metadata": {},
                "name": "3",
                "run_id": "",
                "parent_ids": [],
                "tags": ["my_tag", "seq:step:3"],
            },
            {
                "data": {"output": "olleh"},
                "event": "on_chain_end",
                "metadata": {},
                "name": "3",

Domain

Subdomains

Defined In

Calls

Source

Frequently Asked Questions

What does test_event_stream_with_triple_lambda_test_filtering() do?
test_event_stream_with_triple_lambda_test_filtering() 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_triple_lambda_test_filtering() defined?
test_event_stream_with_triple_lambda_test_filtering() is defined in libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py at line 376.
What does test_event_stream_with_triple_lambda_test_filtering() call?
test_event_stream_with_triple_lambda_test_filtering() 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