test_event_streaming_with_tools() — langchain Function Reference
Architecture documentation for the test_event_streaming_with_tools() function in test_runnable_events_v1.py from the langchain codebase.
Entity Profile
Dependency Diagram
graph TD 866e59ec_9b04_395f_5ae3_1a283a31e020["test_event_streaming_with_tools()"] 8ff41f3c_f250_f8de_8094_4f24860a10e0["test_runnable_events_v1.py"] 866e59ec_9b04_395f_5ae3_1a283a31e020 -->|defined in| 8ff41f3c_f250_f8de_8094_4f24860a10e0 aac0453e_34cd_4dce_c7fa_f176ab20140b["_collect_events()"] 866e59ec_9b04_395f_5ae3_1a283a31e020 -->|calls| aac0453e_34cd_4dce_c7fa_f176ab20140b 6ebe7fde_6e8f_dae5_d42f_9cea181617f5["_assert_events_equal_allow_superset_metadata()"] 866e59ec_9b04_395f_5ae3_1a283a31e020 -->|calls| 6ebe7fde_6e8f_dae5_d42f_9cea181617f5 style 866e59ec_9b04_395f_5ae3_1a283a31e020 fill:#6366f1,stroke:#818cf8,color:#fff
Relationship Graph
Source Code
libs/core/tests/unit_tests/runnables/test_runnable_events_v1.py lines 1064–1228
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}
# type ignores below because the tools don't appear to be runnables to type checkers
# we can remove as soon as that's fixed
events = await _collect_events(parameterless.astream_events({}, version="v1"))
_assert_events_equal_allow_superset_metadata(
events,
[
{
"data": {"input": {}},
"event": "on_tool_start",
"metadata": {},
"name": "parameterless",
"run_id": "",
"parent_ids": [],
"tags": [],
},
{
"data": {"chunk": "hello"},
"event": "on_tool_stream",
"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="v1"))
_assert_events_equal_allow_superset_metadata(
events,
[
{
"data": {"input": {}},
"event": "on_tool_start",
"metadata": {},
"name": "with_callbacks",
"run_id": "",
"parent_ids": [],
"tags": [],
},
{
"data": {"chunk": "world"},
"event": "on_tool_stream",
"metadata": {},
"name": "with_callbacks",
"run_id": "",
"parent_ids": [],
Domain
Subdomains
Source
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_v1.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_v1.py at line 1064.
What does test_event_streaming_with_tools() call?
test_event_streaming_with_tools() calls 2 function(s): _assert_events_equal_allow_superset_metadata, _collect_events.
Analyze Your Own Codebase
Get architecture documentation, dependency graphs, and domain analysis for your codebase in minutes.
Try Supermodel Free