test_fallbacks_stream() — langchain Function Reference

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

Function python CoreAbstractions RunnableInterface

Entity Profile

CoreAbstractions→ RunnableInterface→ test_fallbacks_stream() — langchain Function Reference

Dependency Diagram

graph TD
  e70365dd_f083_4c88_44c4_6461872fdd6c["test_fallbacks_stream()"]
  d62d478e_a330_c328_7da4_58e88d780807["test_fallbacks.py"]
  e70365dd_f083_4c88_44c4_6461872fdd6c -->|defined in| d62d478e_a330_c328_7da4_58e88d780807
  style e70365dd_f083_4c88_44c4_6461872fdd6c fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_fallbacks.py lines 278–288

def test_fallbacks_stream() -> None:
    runnable = RunnableGenerator(_generate_immediate_error).with_fallbacks(
        [RunnableGenerator(_generate)]
    )
    assert list(runnable.stream({})) == list("foo bar")

    runnable = RunnableGenerator(_generate_delayed_error).with_fallbacks(
        [RunnableGenerator(_generate)]
    )
    with pytest.raises(ValueError, match="delayed error"):
        list(runnable.stream({}))