test_runnable_each() — langchain Function Reference

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

Function python LangChainCore Runnables

Entity Profile

LangChainCore→ Runnables→ test_runnable_each() — langchain Function Reference

Dependency Diagram

graph TD
  b19d2c59_018c_3a36_fd9a_0c38e6009fe6["test_runnable_each()"]
  33c02978_2077_5819_7048_bc2a81e80625["test_runnable_events_v2.py"]
  b19d2c59_018c_3a36_fd9a_0c38e6009fe6 -->|defined in| 33c02978_2077_5819_7048_bc2a81e80625
  style b19d2c59_018c_3a36_fd9a_0c38e6009fe6 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py lines 1811–1821

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="v2")]