test_batch_concurrency() — langchain Function Reference

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

Function python LangChainCore Runnables

Entity Profile

LangChainCore→ Runnables→ test_batch_concurrency() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_concurrency.py lines 79–107

def test_batch_concurrency() -> None:
    """Test that batch respects max_concurrency."""
    running_tasks = 0
    max_running_tasks = 0

    lock = Lock()

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

        time.sleep(0.1)  # Simulate work

        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 = runnable.batch(list(range(num_tasks)), config=config)

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