test_deep_astream() — langchain Function Reference

Architecture documentation for the test_deep_astream() function in test_runnable.py from the langchain codebase.

Function python LangChainCore Runnables

Entity Profile

LangChainCore→ Runnables→ test_deep_astream() — langchain Function Reference

Dependency Diagram

graph TD
  15dc48f6_796a_0c82_6685_11d163be0043["test_deep_astream()"]
  26df6ad8_0189_51d0_c3c1_6c3248893ff5["test_runnable.py"]
  15dc48f6_796a_0c82_6685_11d163be0043 -->|defined in| 26df6ad8_0189_51d0_c3c1_6c3248893ff5
  style 15dc48f6_796a_0c82_6685_11d163be0043 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable.py lines 3649–3670

async def test_deep_astream() -> None:
    prompt = (
        SystemMessagePromptTemplate.from_template("You are a nice assistant.")
        + "{question}"
    )
    llm = FakeStreamingListLLM(responses=["foo-lish"])

    chain = prompt | llm | StrOutputParser()

    stream = chain.astream({"question": "What up"})

    chunks = [chunk async for chunk in stream]

    assert len(chunks) == len("foo-lish")
    assert "".join(chunks) == "foo-lish"

    chunks = []
    async for chunk in (chain | RunnablePassthrough()).astream({"question": "What up"}):
        chunks.append(chunk)

    assert len(chunks) == len("foo-lish")
    assert "".join(chunks) == "foo-lish"