test_abatch_concurrency() — langchain Function Reference

Architecture documentation for the test_abatch_concurrency() function in test_concurrency.py from the langchain codebase.

Function python LangChainCore Runnables

Entity Profile

LangChainCore→ Runnables→ test_abatch_concurrency() — langchain Function Reference

Dependency Diagram

graph TD
  ace379a2_ec5a_6b0f_b01f_3e9b4db53603["test_abatch_concurrency()"]
  4a52821c_3a01_e8ab_b3f9_fe358ca00824["test_concurrency.py"]
  ace379a2_ec5a_6b0f_b01f_3e9b4db53603 -->|defined in| 4a52821c_3a01_e8ab_b3f9_fe358ca00824
  style ace379a2_ec5a_6b0f_b01f_3e9b4db53603 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_concurrency.py lines 14–41

async def test_abatch_concurrency() -> None:
    """Test that abatch respects max_concurrency."""
    running_tasks = 0
    max_running_tasks = 0
    lock = asyncio.Lock()

    async def tracked_function(x: Any) -> str:
        nonlocal running_tasks, max_running_tasks
        async with lock:
            running_tasks += 1
            max_running_tasks = max(max_running_tasks, running_tasks)

        await asyncio.sleep(0.1)  # Simulate work

        async with lock:
            running_tasks -= 1

        return f"Completed {x}"

    runnable = RunnableLambda(tracked_function)
    num_tasks = 10
    max_concurrency = 3

    config = RunnableConfig(max_concurrency=max_concurrency)
    results = await runnable.abatch(list(range(num_tasks)), config=config)

    assert len(results) == num_tasks
    assert max_running_tasks <= max_concurrency