test_sync_in_sync_lambdas() — langchain Function Reference

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

Function python LangChainCore Runnables calls 2

Entity Profile

LangChainCore→ Runnables→ test_sync_in_sync_lambdas() — langchain Function Reference

Dependency Diagram

graph TD
  41432541_5739_2a94_468d_fa927530699f["test_sync_in_sync_lambdas()"]
  33c02978_2077_5819_7048_bc2a81e80625["test_runnable_events_v2.py"]
  41432541_5739_2a94_468d_fa927530699f -->|defined in| 33c02978_2077_5819_7048_bc2a81e80625
  96cda70a_ff2d_eca4_0fa9_2ffa9cde3432["stream()"]
  41432541_5739_2a94_468d_fa927530699f -->|calls| 96cda70a_ff2d_eca4_0fa9_2ffa9cde3432
  716d2a5e_dc8e_3cae_e044_b56b06bee655["_collect_events()"]
  41432541_5739_2a94_468d_fa927530699f -->|calls| 716d2a5e_dc8e_3cae_e044_b56b06bee655
  style 41432541_5739_2a94_468d_fa927530699f fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py lines 2101–2118

async def test_sync_in_sync_lambdas() -> None:
    """Test invoking nested runnable lambda."""

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

    add_one_ = RunnableLambda(add_one)

    def add_one_proxy(x: int, config: RunnableConfig) -> int:
        # Use sync streaming
        streaming = add_one_.stream(x, config)
        results = list(streaming)
        return results[0]

    add_one_proxy_ = RunnableLambda(add_one_proxy)

    events = await _collect_events(add_one_proxy_.astream_events(1, version="v2"))
    _assert_events_equal_allow_superset_metadata(events, EXPECTED_EVENTS)