Home / Function/ test_event_streaming_with_tools() — langchain Function Reference

test_event_streaming_with_tools() — langchain Function Reference

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

Entity Profile

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py lines 1085–1210

async def test_event_streaming_with_tools() -> None:
    """Test streaming events with different tool definitions."""

    @tool
    def parameterless() -> str:
        """A tool that does nothing."""
        return "hello"

    @tool
    def with_callbacks(callbacks: Callbacks) -> str:
        """A tool that does nothing."""
        _ = callbacks
        return "world"

    @tool
    def with_parameters(x: int, y: str) -> dict:
        """A tool that does nothing."""
        return {"x": x, "y": y}

    @tool
    def with_parameters_and_callbacks(x: int, y: str, callbacks: Callbacks) -> dict:
        """A tool that does nothing."""
        _ = callbacks
        return {"x": x, "y": y}

    events = await _collect_events(parameterless.astream_events({}, version="v2"))
    _assert_events_equal_allow_superset_metadata(
        events,
        [
            {
                "data": {"input": {}},
                "event": "on_tool_start",
                "metadata": {},
                "name": "parameterless",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
            {
                "data": {"output": "hello"},
                "event": "on_tool_end",
                "metadata": {},
                "name": "parameterless",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
        ],
    )
    events = await _collect_events(with_callbacks.astream_events({}, version="v2"))
    _assert_events_equal_allow_superset_metadata(
        events,
        [
            {
                "data": {"input": {}},
                "event": "on_tool_start",
                "metadata": {},
                "name": "with_callbacks",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
            {
                "data": {"output": "world"},
                "event": "on_tool_end",
                "metadata": {},
                "name": "with_callbacks",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
        ],
    )
    events = await _collect_events(
        with_parameters.astream_events({"x": 1, "y": "2"}, version="v2")
    )
    _assert_events_equal_allow_superset_metadata(
        events,
        [
            {
                "data": {"input": {"x": 1, "y": "2"}},

Domain

Subdomains

Frequently Asked Questions

What does test_event_streaming_with_tools() do?
test_event_streaming_with_tools() is a function in the langchain codebase, defined in libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py.
Where is test_event_streaming_with_tools() defined?
test_event_streaming_with_tools() is defined in libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py at line 1085.
What does test_event_streaming_with_tools() call?
test_event_streaming_with_tools() 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