test_rate_limit_skips_cache_async() — langchain Function Reference

Source Code

libs/core/tests/unit_tests/language_models/chat_models/test_rate_limiting.py lines 244–270

async def test_rate_limit_skips_cache_async(output_version: str) -> None:
    """Test that rate limiting does not rate limit cache look ups."""
    cache = InMemoryCache()
    model = GenericFakeChatModel(
        messages=iter(["hello", "world", "!"]),
        rate_limiter=InMemoryRateLimiter(
            requests_per_second=20, check_every_n_seconds=0.1, max_bucket_size=1
        ),
        cache=cache,
        output_version=output_version,
    )

    tic = time.time()
    await model.ainvoke("foo")
    toc = time.time()
    # Should be larger than check every n seconds since the token bucket starts
    # with 0 tokens.
    assert 0.1 < toc - tic < 0.2

    for _ in range(2):
        # Cache hits
        tic = time.time()
        await model.ainvoke("foo")
        toc = time.time()
        # Should be larger than check every n seconds since the token bucket starts
        # with 0 tokens.
        assert toc - tic < 0.05