Home / Function/ test_event_stream_with_simple_function_tool() — langchain Function Reference

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_v1.py from the langchain codebase.

Entity Profile

Dependency Diagram

graph TD
  0456b5d0_23e7_8258_27a4_1e88a877075e["test_event_stream_with_simple_function_tool()"]
  8ff41f3c_f250_f8de_8094_4f24860a10e0["test_runnable_events_v1.py"]
  0456b5d0_23e7_8258_27a4_1e88a877075e -->|defined in| 8ff41f3c_f250_f8de_8094_4f24860a10e0
  aac0453e_34cd_4dce_c7fa_f176ab20140b["_collect_events()"]
  0456b5d0_23e7_8258_27a4_1e88a877075e -->|calls| aac0453e_34cd_4dce_c7fa_f176ab20140b
  6ebe7fde_6e8f_dae5_d42f_9cea181617f5["_assert_events_equal_allow_superset_metadata()"]
  0456b5d0_23e7_8258_27a4_1e88a877075e -->|calls| 6ebe7fde_6e8f_dae5_d42f_9cea181617f5
  style 0456b5d0_23e7_8258_27a4_1e88a877075e fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v1.py lines 78–169

async def test_event_stream_with_simple_function_tool() -> None:
    """Test the event stream with a function and tool."""

    def foo(_: int) -> dict:
        """Foo."""
        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="v1"))
    _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

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_v1.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_v1.py at line 78.
What does test_event_stream_with_simple_function_tool() call?
test_event_stream_with_simple_function_tool() 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