test_abatch() — langchain Function Reference

Architecture documentation for the test_abatch() function in test_indexing.py from the langchain codebase.

Function python LangChainCore ApiManagement calls 1

Entity Profile

LangChainCore→ ApiManagement→ test_abatch() — langchain Function Reference

Dependency Diagram

graph TD
  1536cbef_ac5e_1c91_31f0_6aaf7042984b["test_abatch()"]
  9bc4e8b6_769a_ae11_3fc1_309cb678c248["test_indexing.py"]
  1536cbef_ac5e_1c91_31f0_6aaf7042984b -->|defined in| 9bc4e8b6_769a_ae11_3fc1_309cb678c248
  c4b4ca1b_606b_2d87_3b94_0903eecdc291["_to_async_iter()"]
  1536cbef_ac5e_1c91_31f0_6aaf7042984b -->|calls| c4b4ca1b_606b_2d87_3b94_0903eecdc291
  style 1536cbef_ac5e_1c91_31f0_6aaf7042984b fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain/tests/unit_tests/indexes/test_indexing.py lines 1355–1371

async def test_abatch() -> None:
    """Test the abatch function."""
    batches = _abatch(5, _to_async_iter(range(12)))
    assert isinstance(batches, AsyncIterator)
    assert [batch async for batch in batches] == [
        [0, 1, 2, 3, 4],
        [5, 6, 7, 8, 9],
        [10, 11],
    ]

    batches = _abatch(1, _to_async_iter(range(3)))
    assert isinstance(batches, AsyncIterator)
    assert [batch async for batch in batches] == [[0], [1], [2]]

    batches = _abatch(2, _to_async_iter(range(5)))
    assert isinstance(batches, AsyncIterator)
    assert [batch async for batch in batches] == [[0, 1], [2, 3], [4]]