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
Calls
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_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