test_runnable_branch_astream() — langchain Function Reference

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

Function python LangChainCore Runnables

Entity Profile

LangChainCore→ Runnables→ test_runnable_branch_astream() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable.py lines 4725–4757

async def test_runnable_branch_astream() -> None:
    """Verify that astream works for RunnableBranch."""
    llm_res = "i'm a textbot"
    # sleep to better simulate a real stream
    llm = FakeStreamingListLLM(responses=[llm_res], sleep=0.01)

    branch = RunnableBranch[str, Any](
        (lambda x: x == "hello", llm),
        lambda x: x,
    )

    assert [_ async for _ in branch.astream("hello")] == list(llm_res)
    assert [_ async for _ in branch.astream("bye")] == ["bye"]

    # Verify that the async variant is used if available
    async def condition(x: str) -> bool:
        return x == "hello"

    async def repeat(x: str) -> str:
        return x + x

    async def reverse(x: str) -> str:
        return x[::-1]

    branch = RunnableBranch[str, Any]((condition, repeat), llm)

    assert [_ async for _ in branch.astream("hello")] == ["hello" * 2]
    assert [_ async for _ in branch.astream("bye")] == list(llm_res)

    branch = RunnableBranch[str, Any]((condition, llm), reverse)

    assert [_ async for _ in branch.astream("hello")] == list(llm_res)
    assert [_ async for _ in branch.astream("bye")] == ["eyb"]