test_runnable_lambda_stream() — langchain Function Reference

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

Function python LangChainCore Runnables

Entity Profile

LangChainCore→ Runnables→ test_runnable_lambda_stream() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable.py lines 4047–4059

def test_runnable_lambda_stream() -> None:
    """Test that stream works for both normal functions & those returning Runnable."""
    # Normal output should work
    output: list[Any] = list(RunnableLambda(range).stream(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 = list(RunnableLambda[str, str](lambda _: llm).stream(""))
    assert output == list(llm_res)