test_runnable_lambda_astream() — langchain Function Reference

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

Function python LangChainCore Runnables

Entity Profile

LangChainCore→ Runnables→ test_runnable_lambda_astream() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable.py lines 4093–4134

async def test_runnable_lambda_astream() -> None:
    """Test that astream works for both normal functions & those returning Runnable."""

    # Wrapper to make a normal function async
    def awrapper(func: Callable[..., Any]) -> Callable[..., Awaitable[Any]]:
        async def afunc(*args: Any, **kwargs: Any) -> Any:
            return func(*args, **kwargs)

        return afunc

    # Normal output should work
    output: list[Any] = [
        chunk
        async for chunk in RunnableLambda(
            func=id,
            afunc=awrapper(range),  # id func is just dummy
        ).astream(5)
    ]
    assert output == [range(5)]

    # Normal output using func should also work
    output = [_ async for _ in RunnableLambda(range).astream(5)]
    assert output == [range(5)]

    # Runnable output should also work
    llm_res = "i'm a textbot"
    # sleep to better simulate a real stream
    llm = FakeStreamingListLLM(responses=[llm_res], sleep=0.01)

    output = [
        _
        async for _ in RunnableLambda(
            func=id,
            afunc=awrapper(lambda _: llm),
        ).astream("")
    ]
    assert output == list(llm_res)

    output = [
        chunk async for chunk in RunnableLambda[str, str](lambda _: llm).astream("")
    ]
    assert output == list(llm_res)