test_runnable_each() — langchain Function Reference

Architecture documentation for the test_runnable_each() function in test_runnable_events_v1.py from the langchain codebase.

Function python LangChainCore Runnables

Entity Profile

LangChainCore→ Runnables→ test_runnable_each() — langchain Function Reference

Dependency Diagram

graph TD
  9bcdf0d0_6aab_af80_a73f_441b3504a047["test_runnable_each()"]
  8ff41f3c_f250_f8de_8094_4f24860a10e0["test_runnable_events_v1.py"]
  9bcdf0d0_6aab_af80_a73f_441b3504a047 -->|defined in| 8ff41f3c_f250_f8de_8094_4f24860a10e0
  style 9bcdf0d0_6aab_af80_a73f_441b3504a047 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v1.py lines 1863–1873

async def test_runnable_each() -> None:
    """Test runnable each astream_events."""

    async def add_one(x: int) -> int:
        return x + 1

    add_one_map = RunnableLambda(add_one).map()
    assert await add_one_map.ainvoke([1, 2, 3]) == [2, 3, 4]

    with pytest.raises(NotImplementedError):
        _ = [_ async for _ in add_one_map.astream_events([1, 2, 3], version="v1")]