test_event_stream_with_simple_function_tool() — langchain Function Reference
Architecture documentation for the test_event_stream_with_simple_function_tool() function in test_runnable_events_v2.py from the langchain codebase.
Entity Profile
Dependency Diagram
graph TD 97b265e9_1012_d5e6_2070_316d2916dd8c["test_event_stream_with_simple_function_tool()"] 33c02978_2077_5819_7048_bc2a81e80625["test_runnable_events_v2.py"] 97b265e9_1012_d5e6_2070_316d2916dd8c -->|defined in| 33c02978_2077_5819_7048_bc2a81e80625 716d2a5e_dc8e_3cae_e044_b56b06bee655["_collect_events()"] 97b265e9_1012_d5e6_2070_316d2916dd8c -->|calls| 716d2a5e_dc8e_3cae_e044_b56b06bee655 style 97b265e9_1012_d5e6_2070_316d2916dd8c fill:#6366f1,stroke:#818cf8,color:#fff
Relationship Graph
Source Code
libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py lines 93–185
async def test_event_stream_with_simple_function_tool() -> None:
"""Test the event stream with a function and tool."""
def foo(x: int) -> dict:
"""Foo."""
_ = x
return {"x": 5}
@tool
def get_docs(x: int) -> list[Document]:
"""Hello Doc."""
_ = x
return [Document(page_content="hello")]
chain = RunnableLambda(foo) | get_docs
events = await _collect_events(chain.astream_events({}, version="v2"))
_assert_events_equal_allow_superset_metadata(
events,
[
{
"event": "on_chain_start",
"run_id": "",
"parent_ids": [],
"name": "RunnableSequence",
"tags": [],
"metadata": {},
"data": {"input": {}},
},
{
"event": "on_chain_start",
"name": "foo",
"run_id": "",
"parent_ids": [],
"tags": ["seq:step:1"],
"metadata": {},
"data": {},
},
{
"event": "on_chain_stream",
"name": "foo",
"run_id": "",
"parent_ids": [],
"tags": ["seq:step:1"],
"metadata": {},
"data": {"chunk": {"x": 5}},
},
{
"event": "on_chain_end",
"name": "foo",
"run_id": "",
"parent_ids": [],
"tags": ["seq:step:1"],
"metadata": {},
"data": {"input": {}, "output": {"x": 5}},
},
{
"event": "on_tool_start",
"name": "get_docs",
"run_id": "",
"parent_ids": [],
"tags": ["seq:step:2"],
"metadata": {},
"data": {"input": {"x": 5}},
},
{
"event": "on_tool_end",
"name": "get_docs",
"run_id": "",
"parent_ids": [],
"tags": ["seq:step:2"],
"metadata": {},
"data": {"input": {"x": 5}, "output": [Document(page_content="hello")]},
},
{
"event": "on_chain_stream",
"run_id": "",
"parent_ids": [],
"tags": [],
"metadata": {},
"name": "RunnableSequence",
"data": {"chunk": [Document(page_content="hello")]},
Domain
Subdomains
Calls
Source
Frequently Asked Questions
What does test_event_stream_with_simple_function_tool() do?
test_event_stream_with_simple_function_tool() 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_simple_function_tool() defined?
test_event_stream_with_simple_function_tool() is defined in libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py at line 93.
What does test_event_stream_with_simple_function_tool() call?
test_event_stream_with_simple_function_tool() 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