langchain Architecture Docs
Home / Function/ test_event_stream_with_retry() — langchain Function Reference

test_event_stream_with_retry() — langchain Function Reference

Architecture documentation for the test_event_stream_with_retry() function in test_runnable_events_v1.py from the langchain codebase.

Function python LangChainCore Runnables calls 2

Entity Profile

Dependency Diagram

graph TD
  57c0d7b0_4be7_678d_5c6f_d567215b6864["test_event_stream_with_retry()"]
  8ff41f3c_f250_f8de_8094_4f24860a10e0["test_runnable_events_v1.py"]
  57c0d7b0_4be7_678d_5c6f_d567215b6864 -->|defined in| 8ff41f3c_f250_f8de_8094_4f24860a10e0
  a14aef27_93bc_0ce1_f208_f5841ae9507f["_with_nulled_run_id()"]
  57c0d7b0_4be7_678d_5c6f_d567215b6864 -->|calls| a14aef27_93bc_0ce1_f208_f5841ae9507f
  6ebe7fde_6e8f_dae5_d42f_9cea181617f5["_assert_events_equal_allow_superset_metadata()"]
  57c0d7b0_4be7_678d_5c6f_d567215b6864 -->|calls| 6ebe7fde_6e8f_dae5_d42f_9cea181617f5
  style 57c0d7b0_4be7_678d_5c6f_d567215b6864 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v1.py lines 1641–1728

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

    def success(_: str) -> str:
        return "success"

    def fail(_: str) -> None:
        """Simple func."""
        msg = "fail"
        raise ValueError(msg)

    chain = RunnableLambda(success) | RunnableLambda(fail).with_retry(
        stop_after_attempt=1,
    )
    iterable = chain.astream_events("q", version="v1")

    events = []

    try:
        for _ in range(10):
            next_chunk = await anext(iterable)
            events.append(next_chunk)
    except Exception:
        pass

    events = _with_nulled_run_id(events)
    for event in events:
        event["tags"] = sorted(event["tags"])

    _assert_events_equal_allow_superset_metadata(
        events,
        [
            {
                "data": {"input": "q"},
                "event": "on_chain_start",
                "metadata": {},
                "name": "RunnableSequence",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
            {
                "data": {},
                "event": "on_chain_start",
                "metadata": {},
                "name": "success",
                "run_id": "",
                "parent_ids": [],
                "tags": ["seq:step:1"],
            },
            {
                "data": {"chunk": "success"},
                "event": "on_chain_stream",
                "metadata": {},
                "name": "success",
                "run_id": "",
                "parent_ids": [],
                "tags": ["seq:step:1"],
            },
            {
                "data": {},
                "event": "on_chain_start",
                "metadata": {},
                "name": "fail",
                "run_id": "",
                "parent_ids": [],
                "tags": ["seq:step:2"],
            },
            {
                "data": {"input": "q", "output": "success"},
                "event": "on_chain_end",
                "metadata": {},
                "name": "success",
                "run_id": "",
                "parent_ids": [],
                "tags": ["seq:step:1"],
            },
            {
                "data": {"input": "success", "output": None},
                "event": "on_chain_end",
                "metadata": {},

Domain

Subdomains

Defined In

Calls

Source

Frequently Asked Questions

What does test_event_stream_with_retry() do?
test_event_stream_with_retry() 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_retry() defined?
test_event_stream_with_retry() is defined in libs/core/tests/unit_tests/runnables/test_runnable_events_v1.py at line 1641.
What does test_event_stream_with_retry() call?
test_event_stream_with_retry() calls 2 function(s): _assert_events_equal_allow_superset_metadata, _with_nulled_run_id.

Analyze Your Own Codebase

Get architecture documentation, dependency graphs, and domain analysis for your codebase in minutes.

Try Supermodel Free