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

test_event_stream_with_retriever_and_formatter() — langchain Function Reference

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

Function python LangChainCore Runnables calls 2

Entity Profile

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v1.py lines 1308–1426

async def test_event_stream_with_retriever_and_formatter() -> None:
    """Test the event stream with a retriever."""
    retriever = HardCodedRetriever(
        documents=[
            Document(
                page_content="hello world!",
                metadata={"foo": "bar"},
            ),
            Document(
                page_content="goodbye world!",
                metadata={"food": "spare"},
            ),
        ]
    )

    def format_docs(docs: list[Document]) -> str:
        """Format the docs."""
        return ", ".join([doc.page_content for doc in docs])

    chain = retriever | format_docs
    events = await _collect_events(chain.astream_events("hello", version="v1"))
    _assert_events_equal_allow_superset_metadata(
        events,
        [
            {
                "data": {"input": "hello"},
                "event": "on_chain_start",
                "metadata": {},
                "name": "RunnableSequence",
                "run_id": "",
                "parent_ids": [],
                "tags": [],
            },
            {
                "data": {"input": {"query": "hello"}},
                "event": "on_retriever_start",
                "metadata": {},
                "name": "HardCodedRetriever",
                "run_id": "",
                "parent_ids": [],
                "tags": ["seq:step:1"],
            },
            {
                "data": {
                    "input": {"query": "hello"},
                    "output": {
                        "documents": [
                            Document(
                                page_content="hello world!", metadata={"foo": "bar"}
                            ),
                            Document(
                                page_content="goodbye world!",
                                metadata={"food": "spare"},
                            ),
                        ]
                    },
                },
                "event": "on_retriever_end",
                "metadata": {},
                "name": "HardCodedRetriever",
                "run_id": "",
                "parent_ids": [],
                "tags": ["seq:step:1"],
            },
            {
                "data": {},
                "event": "on_chain_start",
                "metadata": {},
                "name": "format_docs",
                "run_id": "",
                "parent_ids": [],
                "tags": ["seq:step:2"],
            },
            {
                "data": {"chunk": "hello world!, goodbye world!"},
                "event": "on_chain_stream",
                "metadata": {},
                "name": "format_docs",
                "run_id": "",
                "parent_ids": [],
                "tags": ["seq:step:2"],

Domain

Subdomains

Defined In

Calls

Source

Frequently Asked Questions

What does test_event_stream_with_retriever_and_formatter() do?
test_event_stream_with_retriever_and_formatter() 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_retriever_and_formatter() defined?
test_event_stream_with_retriever_and_formatter() is defined in libs/core/tests/unit_tests/runnables/test_runnable_events_v1.py at line 1308.
What does test_event_stream_with_retriever_and_formatter() call?
test_event_stream_with_retriever_and_formatter() 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