test_async_batch_size() — langchain Function Reference

Architecture documentation for the test_async_batch_size() function in test_base.py from the langchain codebase.

Function python LangChainCore LanguageModelBase

Entity Profile

LangChainCore→ LanguageModelBase→ test_async_batch_size() — langchain Function Reference

Dependency Diagram

graph TD
  60992ee9_e6b8_e194_67e7_395447a0e61a["test_async_batch_size()"]
  8cb88ac4_61d9_baf3_9df4_9b3f5095927e["test_base.py"]
  60992ee9_e6b8_e194_67e7_395447a0e61a -->|defined in| 8cb88ac4_61d9_baf3_9df4_9b3f5095927e
  style 60992ee9_e6b8_e194_67e7_395447a0e61a fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/language_models/chat_models/test_base.py lines 121–145

async def test_async_batch_size(
    messages: list[BaseMessage], messages_2: list[BaseMessage]
) -> None:
    llm = FakeListChatModel(responses=[str(i) for i in range(100)])
    # The base endpoint doesn't support native batching,
    # so we expect batch_size to always be 1
    with collect_runs() as cb:
        await llm.abatch([messages, messages_2], {"callbacks": [cb]})
        assert all((r.extra or {}).get("batch_size") == 1 for r in cb.traced_runs)
        assert len(cb.traced_runs) == 2
    with collect_runs() as cb:
        await llm.abatch([messages], {"callbacks": [cb]})
        assert all((r.extra or {}).get("batch_size") == 1 for r in cb.traced_runs)
        assert len(cb.traced_runs) == 1

    with collect_runs() as cb:
        await llm.ainvoke(messages)
        assert len(cb.traced_runs) == 1
        assert (cb.traced_runs[0].extra or {}).get("batch_size") == 1

    with collect_runs() as cb:
        async for _ in llm.astream(messages):
            pass
        assert len(cb.traced_runs) == 1
        assert (cb.traced_runs[0].extra or {}).get("batch_size") == 1