test_rate_limit_skips_cache() — langchain Function Reference

Source Code

libs/core/tests/unit_tests/language_models/chat_models/test_rate_limiting.py lines 182–222

def test_rate_limit_skips_cache() -> 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,
            # At 20 requests per second we see a refresh every 0.05 seconds
        ),
        cache=cache,
    )

    tic = time.time()
    model.invoke("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()
        model.invoke("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

    # Test verifies that there's only a single key
    # Test also verifies that rate_limiter information is not part of the
    # cache key
    assert list(cache._cache) == [
        (
            '[{"lc": 1, "type": "constructor", "id": ["langchain", "schema", '
            '"messages", "HumanMessage"], "kwargs": {"content": "foo", '
            '"type": "human"}}]',
            "[('_type', 'generic-fake-chat-model'), ('stop', None)]",
        )
    ]