test_fallbacks_astream() — langchain Function Reference

Architecture documentation for the test_fallbacks_astream() function in test_fallbacks.py from the langchain codebase.

Function python LangChainCore Runnables

Entity Profile

LangChainCore→ Runnables→ test_fallbacks_astream() — langchain Function Reference

Dependency Diagram

graph TD
  1945f49c_4058_ac3d_65b8_998f3538f33d["test_fallbacks_astream()"]
  905e276b_98d0_eba1_e23b_8303081c1894["test_fallbacks.py"]
  1945f49c_4058_ac3d_65b8_998f3538f33d -->|defined in| 905e276b_98d0_eba1_e23b_8303081c1894
  style 1945f49c_4058_ac3d_65b8_998f3538f33d fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_fallbacks.py lines 306–318

async def test_fallbacks_astream() -> None:
    runnable = RunnableGenerator(_agenerate_immediate_error).with_fallbacks(
        [RunnableGenerator(_agenerate)]
    )
    expected = (c for c in "foo bar")
    async for c in runnable.astream({}):
        assert c == next(expected)

    runnable = RunnableGenerator(_agenerate_delayed_error).with_fallbacks(
        [RunnableGenerator(_agenerate)]
    )
    with pytest.raises(ValueError, match="delayed error"):
        _ = [_ async for _ in runnable.astream({})]