scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-07 02:59:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103
litellm/tests/test_litellm/caching/test_redis_semantic_cache.py
ryan-crabbe-berri a112ba5f63
test: enforce PT012 so a pytest.raises block cannot hide dead assertions (#37748) 
* test: enforce PT012 so a pytest.raises block cannot hide dead assertions

`with pytest.raises(...)` stops at the first statement that raises. Anything
sequenced after it inside the block never runs, so an assertion written there is
never checked and the test still reports green.

Two sites were doing exactly that, and both assertions turned out to be wrong
once they started running. tests/llm_translation/test_prompt_factory.py asserted
the bedrock rejection names "requires at least one non-system message", which
holds. tests/proxy_unit_tests/test_proxy_server.py asserted the prisma startup
failure mentions "httpx.ConnectError", which never appears: the failure is an
httpx.ConnectError whose message is "All connection attempts failed", so that
test now asserts the type. Its DATABASE_URL override moves to monkeypatch, since
the old restore sat below the assertion and leaked the invalid URL into every
later DB test the moment the assertion started being able to fail.

The remaining 72 sites are rewritten without changing what they exercise: setup
that cannot raise moves above the block, a nested `patch` moves outside it, and
bodies with real control flow (a stream drain, an if/else on sync_mode, a
retry loop) move into a local closure the block calls.

Fixing PT012 unmasked two B017s, since ruff only reports a blind
pytest.raises(Exception) once the block holds a single statement.
tests/proxy_unit_tests/test_auth_checks.py narrows to the ProxyException
can_key_call_model actually raises. tests/local_testing/test_completion_cost.py
was asserting vertex_ai/medlm-medium has no cost entry, which stopped being true
at some point; that dead first half is gone and the rest of the test, which
checks medlm pricing resolves above zero, now runs instead of being skipped.

* chore(ci): ratchet TQ004 to 768 after the prisma test moved to monkeypatch
2026-08-20 19:36:26 -07:00

1486 lines
50 KiB
Python
Raw Permalink Blame History

import os
import sys
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
sys.path.insert(
0, os.path.abspath("../../..")
) # Adds the parent directory to the system path
# Tests for RedisSemanticCache
def test_redis_semantic_cache_initialization(monkeypatch):
# Mock the redisvl import
semantic_cache_mock = MagicMock()
with patch.dict(
"sys.modules",
{
"redisvl.extensions.llmcache": MagicMock(SemanticCache=semantic_cache_mock),
"redisvl.utils.vectorize": MagicMock(CustomTextVectorizer=MagicMock()),
},
):
from litellm.caching.redis_semantic_cache import RedisSemanticCache
# Set environment variables
monkeypatch.setenv("REDIS_HOST", "localhost")
monkeypatch.setenv("REDIS_PORT", "6379")
monkeypatch.setenv("REDIS_PASSWORD", "test_password")
# Initialize the cache with a similarity threshold
redis_semantic_cache = RedisSemanticCache(similarity_threshold=0.8)
# Verify the semantic cache was initialized with correct parameters
assert redis_semantic_cache.similarity_threshold == 0.8
# Use pytest.approx for floating point comparison to handle precision issues
assert redis_semantic_cache.distance_threshold == pytest.approx(0.2, abs=1e-10)
assert redis_semantic_cache.embedding_model == "text-embedding-ada-002"
# Test initialization with missing similarity_threshold
with pytest.raises(ValueError, match="similarity_threshold must be provided"):
RedisSemanticCache()
def test_redis_semantic_cache_get_cache(monkeypatch):
# Mock the redisvl import and embedding function
semantic_cache_mock = MagicMock()
custom_vectorizer_mock = MagicMock()
with patch.dict(
"sys.modules",
{
"redisvl.extensions.llmcache": MagicMock(SemanticCache=semantic_cache_mock),
"redisvl.utils.vectorize": MagicMock(
CustomTextVectorizer=custom_vectorizer_mock
),
},
):
from litellm.caching.redis_semantic_cache import RedisSemanticCache
# Set environment variables
monkeypatch.setenv("REDIS_HOST", "localhost")
monkeypatch.setenv("REDIS_PORT", "6379")
monkeypatch.setenv("REDIS_PASSWORD", "test_password")
# Initialize cache
redis_semantic_cache = RedisSemanticCache(similarity_threshold=0.8)
# Mock the llmcache.check method to return a result
mock_result = [
{
"prompt": "What is the capital of France?",
"response": '{"content": "Paris is the capital of France."}',
"vector_distance": 0.1, # Distance of 0.1 means similarity of 0.9
RedisSemanticCache.CACHE_KEY_FIELD_NAME: "test_key",
}
]
redis_semantic_cache.llmcache.check = MagicMock(return_value=mock_result)
# Mock the embedding function
with (
patch(
"litellm.embedding",
return_value={"data": [{"embedding": [0.1, 0.2, 0.3]}]},
),
patch.object(
redis_semantic_cache,
"_get_cache_key_filter_expression",
return_value="cache-key-filter",
),
):
# Test get_cache with a message
metadata = {}
result = redis_semantic_cache.get_cache(
key="test_key",
messages=[{"content": "What is the capital of France?"}],
metadata=metadata,
)
# Verify result is properly parsed
assert result == {"content": "Paris is the capital of France."}
assert metadata["semantic-similarity"] == pytest.approx(0.9)
# Verify llmcache.check was called
redis_semantic_cache.llmcache.check.assert_called_once_with(
prompt="What is the capital of France?",
vector=[0.1, 0.2, 0.3],
filter_expression="cache-key-filter",
)
def test_redis_semantic_cache_rejects_unscoped_cache_hit(monkeypatch):
semantic_cache_mock = MagicMock()
custom_vectorizer_mock = MagicMock()
with patch.dict(
"sys.modules",
{
"redisvl.extensions.llmcache": MagicMock(SemanticCache=semantic_cache_mock),
"redisvl.utils.vectorize": MagicMock(
CustomTextVectorizer=custom_vectorizer_mock
),
},
):
from litellm.caching.redis_semantic_cache import RedisSemanticCache
monkeypatch.setenv("REDIS_HOST", "localhost")
monkeypatch.setenv("REDIS_PORT", "6379")
monkeypatch.setenv("REDIS_PASSWORD", "test_password")
redis_semantic_cache = RedisSemanticCache(similarity_threshold=0.8)
redis_semantic_cache.llmcache.check = MagicMock(
return_value=[
{
"prompt": "What is the capital of France?",
"response": '{"content": "Paris"}',
"vector_distance": 0.1,
}
]
)
with (
patch(
"litellm.embedding",
return_value={"data": [{"embedding": [0.1, 0.2, 0.3]}]},
),
patch.object(
redis_semantic_cache,
"_get_cache_key_filter_expression",
return_value="cache-key-filter",
),
):
metadata = {}
result = redis_semantic_cache.get_cache(
key="test_key",
messages=[{"content": "What is the capital of France?"}],
metadata=metadata,
)
assert result is None
assert metadata["semantic-similarity"] == 0.0
def test_redis_semantic_cache_set_cache_stores_cache_key_filter(monkeypatch):
semantic_cache_mock = MagicMock()
custom_vectorizer_mock = MagicMock()
with patch.dict(
"sys.modules",
{
"redisvl.extensions.llmcache": MagicMock(SemanticCache=semantic_cache_mock),
"redisvl.utils.vectorize": MagicMock(
CustomTextVectorizer=custom_vectorizer_mock
),
},
):
from litellm.caching.redis_semantic_cache import RedisSemanticCache
monkeypatch.setenv("REDIS_HOST", "localhost")
monkeypatch.setenv("REDIS_PORT", "6379")
monkeypatch.setenv("REDIS_PASSWORD", "test_password")
redis_semantic_cache = RedisSemanticCache(similarity_threshold=0.8)
redis_semantic_cache.llmcache.store = MagicMock()
with patch(
"litellm.embedding",
return_value={"data": [{"embedding": [0.1, 0.2, 0.3]}]},
):
redis_semantic_cache.set_cache(
key="test_key",
value={"content": "Paris"},
messages=[{"content": "What is the capital of France?"}],
ttl=60,
)
redis_semantic_cache.llmcache.store.assert_called_once_with(
"What is the capital of France?",
"{'content': 'Paris'}",
vector=[0.1, 0.2, 0.3],
filters={RedisSemanticCache.CACHE_KEY_FIELD_NAME: "test_key"},
ttl=60,
)
def test_redis_semantic_cache_uses_isolated_index_for_old_schema(monkeypatch):
fallback_cache_mock = MagicMock()
semantic_cache_mock = MagicMock(
side_effect=[
ValueError("stored index schema differs from requested fields"),
fallback_cache_mock,
]
)
custom_vectorizer_mock = MagicMock()
with patch.dict(
"sys.modules",
{
"redisvl.extensions.llmcache": MagicMock(SemanticCache=semantic_cache_mock),
"redisvl.utils.vectorize": MagicMock(
CustomTextVectorizer=custom_vectorizer_mock
),
},
):
from litellm.caching.redis_semantic_cache import RedisSemanticCache
monkeypatch.setenv("REDIS_HOST", "localhost")
monkeypatch.setenv("REDIS_PORT", "6379")
monkeypatch.setenv("REDIS_PASSWORD", "test_password")
redis_semantic_cache = RedisSemanticCache(
similarity_threshold=0.8,
index_name="existing_index",
)
assert redis_semantic_cache.llmcache is fallback_cache_mock
assert semantic_cache_mock.call_args_list[0].kwargs["name"] == "existing_index"
assert (
semantic_cache_mock.call_args_list[1].kwargs["name"]
== "existing_index_isolated"
)
assert semantic_cache_mock.call_args_list[1].kwargs["filterable_fields"] == [
RedisSemanticCache._cache_key_filterable_field()
]
def test_redis_semantic_cache_overwrites_stale_isolated_index(monkeypatch):
fallback_cache_mock = MagicMock()
semantic_cache_mock = MagicMock(
side_effect=[
ValueError("Existing index schema does not match"),
ValueError("Existing index schema does not match"),
fallback_cache_mock,
]
)
custom_vectorizer_mock = MagicMock()
with patch.dict(
"sys.modules",
{
"redisvl.extensions.llmcache": MagicMock(SemanticCache=semantic_cache_mock),
"redisvl.utils.vectorize": MagicMock(
CustomTextVectorizer=custom_vectorizer_mock
),
},
):
from litellm.caching.redis_semantic_cache import RedisSemanticCache
monkeypatch.setenv("REDIS_HOST", "localhost")
monkeypatch.setenv("REDIS_PORT", "6379")
monkeypatch.setenv("REDIS_PASSWORD", "test_password")
redis_semantic_cache = RedisSemanticCache(
similarity_threshold=0.8,
index_name="existing_index",
)
assert redis_semantic_cache.llmcache is fallback_cache_mock
assert (
semantic_cache_mock.call_args_list[2].kwargs["name"]
== "existing_index_isolated"
)
assert semantic_cache_mock.call_args_list[2].kwargs["overwrite"] is True
assert semantic_cache_mock.call_args_list[2].kwargs["filterable_fields"] == [
RedisSemanticCache._cache_key_filterable_field()
]
def test_redis_semantic_cache_reraises_unexpected_isolated_index_error(monkeypatch):
semantic_cache_mock = MagicMock(
side_effect=[
ValueError("Existing index schema does not match"),
ValueError("connection failed"),
]
)
custom_vectorizer_mock = MagicMock()
with patch.dict(
"sys.modules",
{
"redisvl.extensions.llmcache": MagicMock(SemanticCache=semantic_cache_mock),
"redisvl.utils.vectorize": MagicMock(
CustomTextVectorizer=custom_vectorizer_mock
),
},
):
from litellm.caching.redis_semantic_cache import RedisSemanticCache
monkeypatch.setenv("REDIS_HOST", "localhost")
monkeypatch.setenv("REDIS_PORT", "6379")
monkeypatch.setenv("REDIS_PASSWORD", "test_password")
cache = RedisSemanticCache(
similarity_threshold=0.8,
index_name="existing_index",
)
with pytest.raises(ValueError, match="connection failed"):
_ = cache.llmcache
def test_redis_semantic_cache_reraises_unexpected_index_error():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
redis_semantic_cache = RedisSemanticCache.__new__(RedisSemanticCache)
redis_semantic_cache.distance_threshold = 0.2
semantic_cache_mock = MagicMock(side_effect=ValueError("connection failed"))
with pytest.raises(ValueError, match="connection failed"):
redis_semantic_cache._init_semantic_cache(
semantic_cache_cls=semantic_cache_mock,
index_name="existing_index",
redis_url="redis://localhost:6379",
cache_vectorizer=MagicMock(),
)
def test_redis_semantic_cache_matches_bytes_cache_key():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
redis_semantic_cache = RedisSemanticCache.__new__(RedisSemanticCache)
assert redis_semantic_cache._cache_hit_matches_key(
cache_hit={RedisSemanticCache.CACHE_KEY_FIELD_NAME: b"test_key"},
key="test_key",
)
def test_redis_semantic_cache_rejects_pre_isolation_unscoped_hit():
"""Pre-isolation entries with no cache-key field cannot be safely
reassigned to a caller's scope and are treated as misses."""
from litellm.caching.redis_semantic_cache import RedisSemanticCache
redis_semantic_cache = RedisSemanticCache.__new__(RedisSemanticCache)
cache_hit = {
"prompt": "What is the capital of France?",
"response": '{"content": "Paris"}',
"vector_distance": 0.1,
}
assert not redis_semantic_cache._cache_hit_matches_key(
cache_hit=cache_hit,
key="test_key",
)
def test_redis_semantic_cache_builds_filter_expression(monkeypatch):
class FakeTag:
def __init__(self, field_name):
self.field_name = field_name
def __eq__(self, value):
return (self.field_name, value)
with patch.dict("sys.modules", {"redisvl.query.filter": MagicMock(Tag=FakeTag)}):
from litellm.caching.redis_semantic_cache import RedisSemanticCache
redis_semantic_cache = RedisSemanticCache.__new__(RedisSemanticCache)
assert redis_semantic_cache._get_cache_key_filter_expression("test_key") == (
RedisSemanticCache.CACHE_KEY_FIELD_NAME,
"test_key",
)
@pytest.mark.asyncio
async def test_redis_semantic_cache_async_get_cache(monkeypatch):
# Mock the redisvl import
semantic_cache_mock = MagicMock()
custom_vectorizer_mock = MagicMock()
with patch.dict(
"sys.modules",
{
"redisvl.extensions.llmcache": MagicMock(SemanticCache=semantic_cache_mock),
"redisvl.utils.vectorize": MagicMock(
CustomTextVectorizer=custom_vectorizer_mock
),
},
):
from litellm.caching.redis_semantic_cache import RedisSemanticCache
# Set environment variables
monkeypatch.setenv("REDIS_HOST", "localhost")
monkeypatch.setenv("REDIS_PORT", "6379")
monkeypatch.setenv("REDIS_PASSWORD", "test_password")
# Initialize cache
redis_semantic_cache = RedisSemanticCache(similarity_threshold=0.8)
# Mock the async methods
mock_result = [
{
"prompt": "What is the capital of France?",
"response": '{"content": "Paris is the capital of France."}',
"vector_distance": 0.1, # Distance of 0.1 means similarity of 0.9
RedisSemanticCache.CACHE_KEY_FIELD_NAME: "test_key",
}
]
redis_semantic_cache.llmcache.acheck = AsyncMock(return_value=mock_result)
redis_semantic_cache._get_async_embedding = AsyncMock(
return_value=[0.1, 0.2, 0.3]
)
with patch.object(
redis_semantic_cache,
"_get_cache_key_filter_expression",
return_value="cache-key-filter",
):
# Test async_get_cache with a message
result = await redis_semantic_cache.async_get_cache(
key="test_key",
messages=[{"content": "What is the capital of France?"}],
metadata={},
)
# Verify result is properly parsed
assert result == {"content": "Paris is the capital of France."}
# Verify methods were called
redis_semantic_cache._get_async_embedding.assert_called_once()
redis_semantic_cache.llmcache.acheck.assert_called_once_with(
prompt="What is the capital of France?",
vector=[0.1, 0.2, 0.3],
filter_expression="cache-key-filter",
)
@pytest.mark.asyncio
async def test_redis_semantic_cache_async_get_cache_rejects_unscoped_hit(monkeypatch):
semantic_cache_mock = MagicMock()
custom_vectorizer_mock = MagicMock()
with patch.dict(
"sys.modules",
{
"redisvl.extensions.llmcache": MagicMock(SemanticCache=semantic_cache_mock),
"redisvl.utils.vectorize": MagicMock(
CustomTextVectorizer=custom_vectorizer_mock
),
},
):
from litellm.caching.redis_semantic_cache import RedisSemanticCache
monkeypatch.setenv("REDIS_HOST", "localhost")
monkeypatch.setenv("REDIS_PORT", "6379")
monkeypatch.setenv("REDIS_PASSWORD", "test_password")
redis_semantic_cache = RedisSemanticCache(similarity_threshold=0.8)
redis_semantic_cache.llmcache.acheck = AsyncMock(
return_value=[
{
"prompt": "What is the capital of France?",
"response": '{"content": "Paris"}',
"vector_distance": 0.1,
}
]
)
redis_semantic_cache._get_async_embedding = AsyncMock(
return_value=[0.1, 0.2, 0.3]
)
with patch.object(
redis_semantic_cache,
"_get_cache_key_filter_expression",
return_value="cache-key-filter",
):
result = await redis_semantic_cache.async_get_cache(
key="test_key",
messages=[{"content": "What is the capital of France?"}],
metadata={},
)
assert result is None
@pytest.mark.asyncio
async def test_redis_semantic_cache_async_set_cache_stores_cache_key_filter(
monkeypatch,
):
semantic_cache_mock = MagicMock()
custom_vectorizer_mock = MagicMock()
with patch.dict(
"sys.modules",
{
"redisvl.extensions.llmcache": MagicMock(SemanticCache=semantic_cache_mock),
"redisvl.utils.vectorize": MagicMock(
CustomTextVectorizer=custom_vectorizer_mock
),
},
):
from litellm.caching.redis_semantic_cache import RedisSemanticCache
monkeypatch.setenv("REDIS_HOST", "localhost")
monkeypatch.setenv("REDIS_PORT", "6379")
monkeypatch.setenv("REDIS_PASSWORD", "test_password")
redis_semantic_cache = RedisSemanticCache(similarity_threshold=0.8)
redis_semantic_cache.llmcache.astore = AsyncMock()
redis_semantic_cache._get_async_embedding = AsyncMock(
return_value=[0.1, 0.2, 0.3]
)
await redis_semantic_cache.async_set_cache(
key="test_key",
value={"content": "Paris"},
messages=[{"content": "What is the capital of France?"}],
ttl=60,
)
redis_semantic_cache.llmcache.astore.assert_called_once_with(
"What is the capital of France?",
"{'content': 'Paris'}",
vector=[0.1, 0.2, 0.3],
filters={RedisSemanticCache.CACHE_KEY_FIELD_NAME: "test_key"},
ttl=60,
)
def test_redis_semantic_cache_set_cache_uses_responses_string_input():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
redis_semantic_cache = RedisSemanticCache.__new__(RedisSemanticCache)
redis_semantic_cache.llmcache = MagicMock()
redis_semantic_cache._get_cache_filters = MagicMock(
return_value={RedisSemanticCache.CACHE_KEY_FIELD_NAME: "test_key"}
)
redis_semantic_cache._get_ttl = MagicMock(return_value=None)
redis_semantic_cache._get_embedding = MagicMock(return_value=[0.1, 0.2, 0.3])
redis_semantic_cache.set_cache(
key="test_key",
value={"content": "Paris"},
input="What is the capital of France?",
)
redis_semantic_cache.llmcache.store.assert_called_once_with(
"What is the capital of France?",
"{'content': 'Paris'}",
vector=[0.1, 0.2, 0.3],
filters={RedisSemanticCache.CACHE_KEY_FIELD_NAME: "test_key"},
)
def test_redis_semantic_cache_get_cache_uses_responses_string_input():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
redis_semantic_cache = RedisSemanticCache.__new__(RedisSemanticCache)
redis_semantic_cache.similarity_threshold = 0.8
redis_semantic_cache.llmcache = MagicMock()
redis_semantic_cache.llmcache.check = MagicMock(
return_value=[
{
"prompt": "What is the capital of France?",
"response": '{"content": "Paris"}',
"vector_distance": 0.1,
RedisSemanticCache.CACHE_KEY_FIELD_NAME: "test_key",
}
]
)
redis_semantic_cache._get_embedding = MagicMock(return_value=[0.1, 0.2, 0.3])
with patch.object(
redis_semantic_cache,
"_get_cache_key_filter_expression",
return_value="cache-key-filter",
):
metadata = {}
result = redis_semantic_cache.get_cache(
key="test_key",
input="What is the capital of France?",
metadata=metadata,
)
assert result == {"content": "Paris"}
assert metadata["semantic-similarity"] == pytest.approx(0.9)
redis_semantic_cache.llmcache.check.assert_called_once_with(
prompt="What is the capital of France?",
vector=[0.1, 0.2, 0.3],
filter_expression="cache-key-filter",
)
def test_redis_semantic_cache_set_cache_flattens_structured_responses_input():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
redis_semantic_cache = RedisSemanticCache.__new__(RedisSemanticCache)
redis_semantic_cache.llmcache = MagicMock()
redis_semantic_cache._get_cache_filters = MagicMock(
return_value={RedisSemanticCache.CACHE_KEY_FIELD_NAME: "test_key"}
)
redis_semantic_cache._get_ttl = MagicMock(return_value=None)
redis_semantic_cache._get_embedding = MagicMock(return_value=[0.1, 0.2, 0.3])
redis_semantic_cache.set_cache(
key="test_key",
value={"content": "Paris"},
input=[
{
"role": "user",
"content": [
{"type": "input_text", "text": "What is the capital of France?"},
{"type": "input_text", "text": "Answer briefly."},
{
"type": "input_image",
"image_url": "https://example.com/paris.png",
},
],
}
],
)
redis_semantic_cache.llmcache.store.assert_called_once_with(
"What is the capital of France?\nAnswer briefly.",
"{'content': 'Paris'}",
vector=[0.1, 0.2, 0.3],
filters={RedisSemanticCache.CACHE_KEY_FIELD_NAME: "test_key"},
)
def test_redis_semantic_cache_prompt_extraction_prefers_messages():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
prompt = RedisSemanticCache._get_prompt_from_kwargs(
messages=[{"content": "message prompt"}],
input="responses prompt",
)
assert prompt == "message prompt"
def test_redis_semantic_cache_prompt_extraction_handles_model_objects():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
class ModelDumpInput:
def model_dump(self):
return {"content": [{"text": "model dump prompt"}]}
class DictInput:
def dict(self):
return {"content": [{"output_text": "dict prompt"}]}
prompt = RedisSemanticCache._get_prompt_from_kwargs(
input=[
ModelDumpInput(),
DictInput(),
{"content": [{"input_text": "inline prompt"}]},
{"content": [{"type": "input_image", "image_url": "https://example.com"}]},
]
)
assert prompt == "model dump prompt\ndict prompt\ninline prompt"
def test_redis_semantic_cache_prompt_extraction_returns_none_without_text():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
assert RedisSemanticCache._get_prompt_from_kwargs() is None
assert RedisSemanticCache._get_prompt_from_kwargs(input=None) is None
assert RedisSemanticCache._get_prompt_from_kwargs(input=" ") is None
assert (
RedisSemanticCache._get_prompt_from_kwargs(
input=[{"type": "input_image", "image_url": "https://example.com"}]
)
is None
)
def test_redis_semantic_cache_prompt_extraction_skips_blank_dict_text_keys():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
prompt = RedisSemanticCache._get_prompt_from_kwargs(
input={"text": " ", "input_text": "fallback prompt"}
)
assert prompt == "fallback prompt"
def test_redis_semantic_cache_prompt_extraction_skips_blank_object_text_keys():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
class ResponseInput:
text = " "
input_text = "fallback prompt"
prompt = RedisSemanticCache._get_prompt_from_kwargs(input=ResponseInput())
assert prompt == "fallback prompt"
def test_redis_semantic_cache_prompt_extraction_handles_object_content():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
class ResponseInput:
content = [{"text": "object content prompt"}]
prompt = RedisSemanticCache._get_prompt_from_kwargs(input=ResponseInput())
assert prompt == "object content prompt"
def test_redis_semantic_cache_set_cache_skips_blank_responses_input():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
redis_semantic_cache = RedisSemanticCache.__new__(RedisSemanticCache)
redis_semantic_cache.llmcache = MagicMock()
redis_semantic_cache.set_cache(
key="test_key",
value={"content": "Paris"},
input=" ",
)
redis_semantic_cache.llmcache.store.assert_not_called()
def test_redis_semantic_cache_get_cache_sets_similarity_on_blank_responses_input():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
redis_semantic_cache = RedisSemanticCache.__new__(RedisSemanticCache)
redis_semantic_cache.llmcache = MagicMock()
metadata = {}
result = redis_semantic_cache.get_cache(
key="test_key",
input=" ",
metadata=metadata,
)
assert result is None
assert metadata["semantic-similarity"] == 0.0
redis_semantic_cache.llmcache.check.assert_not_called()
def test_redis_semantic_cache_get_cache_sets_similarity_when_no_results():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
redis_semantic_cache = RedisSemanticCache.__new__(RedisSemanticCache)
redis_semantic_cache.llmcache = MagicMock()
redis_semantic_cache.llmcache.check = MagicMock(return_value=[])
redis_semantic_cache._get_embedding = MagicMock(return_value=[0.1, 0.2, 0.3])
with patch.object(
redis_semantic_cache,
"_get_cache_key_filter_expression",
return_value="cache-key-filter",
):
metadata = {}
result = redis_semantic_cache.get_cache(
key="test_key",
input="What is the capital of France?",
metadata=metadata,
)
assert result is None
assert metadata["semantic-similarity"] == 0.0
redis_semantic_cache.llmcache.check.assert_called_once_with(
prompt="What is the capital of France?",
vector=[0.1, 0.2, 0.3],
filter_expression="cache-key-filter",
)
@pytest.mark.asyncio
async def test_redis_semantic_cache_async_paths_use_responses_string_input():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
redis_semantic_cache = RedisSemanticCache.__new__(RedisSemanticCache)
redis_semantic_cache.similarity_threshold = 0.8
redis_semantic_cache.llmcache = MagicMock()
redis_semantic_cache.llmcache.astore = AsyncMock()
redis_semantic_cache.llmcache.acheck = AsyncMock(
return_value=[
{
"prompt": "What is the capital of France?",
"response": '{"content": "Paris"}',
"vector_distance": 0.1,
RedisSemanticCache.CACHE_KEY_FIELD_NAME: "test_key",
}
]
)
redis_semantic_cache._get_cache_filters = MagicMock(
return_value={RedisSemanticCache.CACHE_KEY_FIELD_NAME: "test_key"}
)
redis_semantic_cache._get_ttl = MagicMock(return_value=None)
redis_semantic_cache._get_async_embedding = AsyncMock(return_value=[0.1, 0.2, 0.3])
await redis_semantic_cache.async_set_cache(
key="test_key",
value={"content": "Paris"},
input="What is the capital of France?",
)
with patch.object(
redis_semantic_cache,
"_get_cache_key_filter_expression",
return_value="cache-key-filter",
):
metadata = {}
result = await redis_semantic_cache.async_get_cache(
key="test_key",
input="What is the capital of France?",
metadata=metadata,
)
redis_semantic_cache.llmcache.astore.assert_called_once_with(
"What is the capital of France?",
"{'content': 'Paris'}",
vector=[0.1, 0.2, 0.3],
filters={RedisSemanticCache.CACHE_KEY_FIELD_NAME: "test_key"},
)
assert result == {"content": "Paris"}
assert metadata["semantic-similarity"] == pytest.approx(0.9)
redis_semantic_cache.llmcache.acheck.assert_called_once_with(
prompt="What is the capital of France?",
vector=[0.1, 0.2, 0.3],
filter_expression="cache-key-filter",
)
@pytest.mark.asyncio
async def test_redis_semantic_cache_async_paths_set_similarity_on_misses():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
redis_semantic_cache = RedisSemanticCache.__new__(RedisSemanticCache)
redis_semantic_cache.llmcache = MagicMock()
redis_semantic_cache.llmcache.astore = AsyncMock()
redis_semantic_cache.llmcache.acheck = AsyncMock(return_value=[])
redis_semantic_cache._get_async_embedding = AsyncMock(return_value=[0.1, 0.2, 0.3])
await redis_semantic_cache.async_set_cache(
key="test_key",
value={"content": "Paris"},
input=" ",
)
redis_semantic_cache.llmcache.astore.assert_not_called()
redis_semantic_cache._get_async_embedding.assert_not_called()
blank_metadata = {}
blank_result = await redis_semantic_cache.async_get_cache(
key="test_key",
input=" ",
metadata=blank_metadata,
)
assert blank_result is None
assert blank_metadata["semantic-similarity"] == 0.0
redis_semantic_cache.llmcache.acheck.assert_not_called()
redis_semantic_cache._get_async_embedding.assert_not_called()
with patch.object(
redis_semantic_cache,
"_get_cache_key_filter_expression",
return_value="cache-key-filter",
):
miss_metadata = {}
miss_result = await redis_semantic_cache.async_get_cache(
key="test_key",
input="What is the capital of France?",
metadata=miss_metadata,
)
assert miss_result is None
assert miss_metadata["semantic-similarity"] == 0.0
redis_semantic_cache.llmcache.acheck.assert_called_once_with(
prompt="What is the capital of France?",
vector=[0.1, 0.2, 0.3],
filter_expression="cache-key-filter",
)
def test_redis_get_embedding_routes_through_router(monkeypatch):
import sys
import types
from litellm.caching.redis_semantic_cache import RedisSemanticCache
cache = RedisSemanticCache.__new__(RedisSemanticCache)
cache.embedding_model = "sem-embed"
router = MagicMock()
router.get_configured_token_limits.return_value = (None, None)
router.embedding = MagicMock(return_value={"data": [{"embedding": [0.5, 0.6]}]})
fake_proxy = types.ModuleType("litellm.proxy.proxy_server")
fake_proxy.llm_router = router
fake_proxy.llm_model_list = [{"model_name": "sem-embed"}]
monkeypatch.setitem(sys.modules, "litellm.proxy.proxy_server", fake_proxy)
with patch("litellm.embedding") as direct_embed:
vec = cache._get_embedding("hello", metadata={"user_api_key": "sk-x"})
assert vec == [0.5, 0.6]
router.embedding.assert_called_once()
assert router.embedding.call_args.kwargs["model"] == "sem-embed"
assert router.embedding.call_args.kwargs["input"] == "hello"
assert router.embedding.call_args.kwargs["cache"] == {
"no-store": True,
"no-cache": True,
}
assert router.embedding.call_args.kwargs["metadata"] == {
"user_api_key": "sk-x",
"semantic-cache-embedding": True,
}
direct_embed.assert_not_called()
def test_redis_get_embedding_falls_back_to_direct(monkeypatch):
import sys
import types
from litellm.caching.redis_semantic_cache import RedisSemanticCache
cache = RedisSemanticCache.__new__(RedisSemanticCache)
cache.embedding_model = "text-embedding-ada-002"
fake_proxy = types.ModuleType("litellm.proxy.proxy_server")
fake_proxy.llm_router = None
fake_proxy.llm_model_list = None
monkeypatch.setitem(sys.modules, "litellm.proxy.proxy_server", fake_proxy)
with patch(
"litellm.embedding", return_value={"data": [{"embedding": [0.1, 0.2]}]}
) as direct_embed:
vec = cache._get_embedding("hello")
assert vec == [0.1, 0.2]
direct_embed.assert_called_once()
def test_cache_get_cache_passes_responses_input_to_backend_cache():
from litellm.caching.caching import Cache
cache = Cache.__new__(Cache)
cache.cache = MagicMock()
cache.cache.get_cache = MagicMock(return_value=None)
cache.should_use_cache = MagicMock(return_value=True)
cache.get_cache_key = MagicMock(return_value="test_key")
metadata = {}
cache.get_cache(
input="What is the capital of France?",
metadata=metadata,
cache={},
)
cache.cache.get_cache.assert_called_once_with(
"test_key",
input="What is the capital of France?",
metadata=metadata,
)
def test_cache_get_cache_filters_non_lookup_kwargs_from_backend_cache():
from litellm.caching.caching import Cache
cache = Cache.__new__(Cache)
cache.cache = MagicMock()
cache.should_use_cache = MagicMock(return_value=True)
cache.get_cache_key = MagicMock(return_value="test_key")
cache._get_cache_logic = MagicMock(return_value={"content": "Paris"})
def _cache_hit(_cache_key, **cache_kwargs):
cache_kwargs["metadata"]["semantic-similarity"] = 0.7
return {"content": "Paris"}
cache.cache.get_cache = MagicMock(side_effect=_cache_hit)
metadata = {"user_api_key": "sk-secret", "trace_id": "trace-id"}
result = cache.get_cache(
input="What is the capital of France?",
metadata=metadata,
cache={"s-maxage": 10},
api_key="sk-secret",
headers={"authorization": "Bearer sk-secret"},
)
assert result == {"content": "Paris"}
assert metadata == {
"user_api_key": "sk-secret",
"trace_id": "trace-id",
"semantic-similarity": 0.7,
}
forwarded_kwargs = cache.cache.get_cache.call_args.kwargs
assert forwarded_kwargs == {
"input": "What is the capital of France?",
"metadata": {
"user_api_key": "sk-secret",
"trace_id": "trace-id",
"semantic-similarity": 0.7,
},
}
assert forwarded_kwargs["metadata"] is not metadata
cache._get_cache_logic.assert_called_once_with(
cached_result={"content": "Paris"},
max_age=10,
)
def test_cache_get_cache_filters_sensitive_kwargs_without_metadata():
from litellm.caching.caching import Cache
cache = Cache.__new__(Cache)
cache.cache = MagicMock()
cache.cache.get_cache = MagicMock(return_value={"content": "Paris"})
cache.should_use_cache = MagicMock(return_value=True)
cache.get_cache_key = MagicMock(return_value="test_key")
cache._get_cache_logic = MagicMock(return_value={"content": "Paris"})
result = cache.get_cache(
input="What is the capital of France?",
cache={"s-maxage": 10},
api_key="sk-secret",
headers={"authorization": "Bearer sk-secret"},
)
assert result == {"content": "Paris"}
cache.cache.get_cache.assert_called_once_with(
"test_key",
input="What is the capital of France?",
)
def test_cache_get_cache_passes_responses_input_to_dynamic_cache():
from litellm.caching.caching import Cache
cache = Cache.__new__(Cache)
cache.should_use_cache = MagicMock(return_value=True)
cache.get_cache_key = MagicMock(return_value="test_key")
cache._get_cache_logic = MagicMock(return_value={"content": "Paris"})
dynamic_cache_object = MagicMock()
dynamic_cache_object.get_cache = MagicMock(return_value={"content": "Paris"})
metadata = {}
result = cache.get_cache(
dynamic_cache_object=dynamic_cache_object,
input="What is the capital of France?",
metadata=metadata,
cache={},
)
assert result == {"content": "Paris"}
dynamic_cache_object.get_cache.assert_called_once_with(
"test_key",
input="What is the capital of France?",
metadata=metadata,
)
cache._get_cache_logic.assert_called_once_with(
cached_result={"content": "Paris"},
max_age=float("inf"),
)
def test_redis_sync_set_cache_passes_precomputed_vector():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
cache = RedisSemanticCache.__new__(RedisSemanticCache)
cache.llmcache = MagicMock()
cache._get_cache_filters = MagicMock(
return_value={RedisSemanticCache.CACHE_KEY_FIELD_NAME: "test_key"}
)
cache._get_ttl = MagicMock(return_value=None)
cache._get_embedding = MagicMock(return_value=[0.1, 0.2, 0.3])
cache.set_cache(
key="test_key",
value={"content": "Paris"},
messages=[{"content": "What is the capital of France?"}],
)
cache._get_embedding.assert_called_once()
cache.llmcache.store.assert_called_once_with(
"What is the capital of France?",
"{'content': 'Paris'}",
vector=[0.1, 0.2, 0.3],
filters={RedisSemanticCache.CACHE_KEY_FIELD_NAME: "test_key"},
)
def test_redis_sync_get_cache_passes_precomputed_vector():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
cache = RedisSemanticCache.__new__(RedisSemanticCache)
cache.similarity_threshold = 0.8
cache.llmcache = MagicMock()
cache.llmcache.check = MagicMock(
return_value=[
{
"prompt": "What is the capital of France?",
"response": '{"content": "Paris"}',
"vector_distance": 0.1,
RedisSemanticCache.CACHE_KEY_FIELD_NAME: "test_key",
}
]
)
cache._get_embedding = MagicMock(return_value=[0.1, 0.2, 0.3])
with patch.object(
cache, "_get_cache_key_filter_expression", return_value="cache-key-filter"
):
result = cache.get_cache(
key="test_key",
messages=[{"content": "What is the capital of France?"}],
metadata={},
)
assert result == {"content": "Paris"}
cache._get_embedding.assert_called_once()
cache.llmcache.check.assert_called_once_with(
prompt="What is the capital of France?",
vector=[0.1, 0.2, 0.3],
filter_expression="cache-key-filter",
)
@pytest.mark.asyncio
async def test_redis_async_embedding_forwards_full_metadata(monkeypatch):
import sys
import types
from litellm.caching.redis_semantic_cache import RedisSemanticCache
cache = RedisSemanticCache.__new__(RedisSemanticCache)
cache.embedding_model = "sem-embed"
router = MagicMock()
router.get_configured_token_limits.return_value = (None, None)
router.aembedding = AsyncMock(return_value={"data": [{"embedding": [0.1, 0.2]}]})
fake_proxy = types.ModuleType("litellm.proxy.proxy_server")
fake_proxy.llm_router = router
fake_proxy.llm_model_list = [{"model_name": "sem-embed"}]
monkeypatch.setitem(sys.modules, "litellm.proxy.proxy_server", fake_proxy)
await cache._get_async_embedding(
"hello",
metadata={"user_api_key": "sk-x", "user_api_key_team_id": "team-1"},
)
md = router.aembedding.call_args.kwargs["metadata"]
assert md["user_api_key"] == "sk-x"
assert md["user_api_key_team_id"] == "team-1" # FAILS today: team_id is dropped
assert md["semantic-cache-embedding"] is True
LONG_PROMPT = " ".join(f"token{i}" for i in range(300))
def _proxy_with_router(monkeypatch: pytest.MonkeyPatch, router: MagicMock, model_name: str) -> None:
import sys
import types
fake_proxy = types.ModuleType("litellm.proxy.proxy_server")
fake_proxy.llm_router = router
fake_proxy.llm_model_list = [{"model_name": model_name}]
monkeypatch.setitem(sys.modules, "litellm.proxy.proxy_server", fake_proxy)
def _token_count(model: str, text: str) -> int:
import litellm
return len(litellm.encode(model=model, text=text))
def test_redis_get_embedding_truncates_to_deployment_max_input_tokens(monkeypatch):
from litellm.caching.redis_semantic_cache import RedisSemanticCache
cache = RedisSemanticCache.__new__(RedisSemanticCache)
cache.embedding_model = "sem-embed"
router = MagicMock()
router.get_configured_token_limits.return_value = (5, None)
router.embedding = MagicMock(return_value={"data": [{"embedding": [0.5, 0.6]}]})
_proxy_with_router(monkeypatch, router, "sem-embed")
assert cache._get_embedding(LONG_PROMPT) == [0.5, 0.6]
sent_input = router.embedding.call_args.kwargs["input"]
assert LONG_PROMPT.startswith(sent_input)
assert _token_count("sem-embed", sent_input) == 5
assert _token_count("sem-embed", LONG_PROMPT) > 5
@pytest.mark.asyncio
async def test_redis_async_embedding_explicit_limit_beats_deployment_limit(monkeypatch):
from litellm.caching.redis_semantic_cache import RedisSemanticCache
cache = RedisSemanticCache.__new__(RedisSemanticCache)
cache.embedding_model = "sem-embed"
cache.embedding_max_input_tokens = 3
router = MagicMock()
router.get_configured_token_limits.return_value = (8191, None)
router.aembedding = AsyncMock(return_value={"data": [{"embedding": [0.1, 0.2]}]})
_proxy_with_router(monkeypatch, router, "sem-embed")
assert await cache._get_async_embedding(LONG_PROMPT) == [0.1, 0.2]
sent_input = router.aembedding.call_args.kwargs["input"]
assert _token_count("sem-embed", sent_input) == 3
def test_redis_get_embedding_truncates_direct_path_with_explicit_limit(monkeypatch):
import sys
import types
from litellm.caching.redis_semantic_cache import RedisSemanticCache
cache = RedisSemanticCache.__new__(RedisSemanticCache)
cache.embedding_model = "text-embedding-3-small"
cache.embedding_max_input_tokens = 4
fake_proxy = types.ModuleType("litellm.proxy.proxy_server")
fake_proxy.llm_router = None
fake_proxy.llm_model_list = None
monkeypatch.setitem(sys.modules, "litellm.proxy.proxy_server", fake_proxy)
with patch(
"litellm.embedding", return_value={"data": [{"embedding": [0.1, 0.2]}]}
) as direct_embed:
cache._get_embedding(LONG_PROMPT)
sent_input = direct_embed.call_args.kwargs["input"]
assert _token_count("text-embedding-3-small", sent_input) == 4
def test_redis_semantic_cache_init_stores_embedding_max_input_tokens(monkeypatch):
from litellm.caching.redis_semantic_cache import RedisSemanticCache
cache = RedisSemanticCache(
redis_url="redis://localhost:6379",
similarity_threshold=0.8,
embedding_max_input_tokens=512,
)
assert cache.embedding_max_input_tokens == 512
default_cache = RedisSemanticCache(redis_url="redis://localhost:6379", similarity_threshold=0.8)
assert default_cache.embedding_max_input_tokens is None
def test_redis_init_defers_redisvl_construction(monkeypatch):
semantic_cache_mock = MagicMock()
custom_vectorizer_mock = MagicMock()
with patch.dict(
"sys.modules",
{
"redisvl.extensions.llmcache": MagicMock(SemanticCache=semantic_cache_mock),
"redisvl.utils.vectorize": MagicMock(
CustomTextVectorizer=custom_vectorizer_mock
),
},
):
from litellm.caching.redis_semantic_cache import RedisSemanticCache
monkeypatch.setenv("REDIS_HOST", "localhost")
monkeypatch.setenv("REDIS_PORT", "6379")
monkeypatch.setenv("REDIS_PASSWORD", "test_password")
cache = RedisSemanticCache(similarity_threshold=0.8)
semantic_cache_mock.assert_not_called()
custom_vectorizer_mock.assert_not_called()
first = cache.llmcache
semantic_cache_mock.assert_called_once()
custom_vectorizer_mock.assert_called_once()
second = cache.llmcache
assert first is second
semantic_cache_mock.assert_called_once()
def test_redis_failed_llmcache_build_is_not_memoized(monkeypatch):
built_cache = MagicMock()
semantic_cache_mock = MagicMock(
side_effect=[ConnectionError("redis down"), built_cache]
)
custom_vectorizer_mock = MagicMock()
with patch.dict(
"sys.modules",
{
"redisvl.extensions.llmcache": MagicMock(SemanticCache=semantic_cache_mock),
"redisvl.utils.vectorize": MagicMock(
CustomTextVectorizer=custom_vectorizer_mock
),
},
):
from litellm.caching.redis_semantic_cache import RedisSemanticCache
monkeypatch.setenv("REDIS_HOST", "localhost")
monkeypatch.setenv("REDIS_PORT", "6379")
monkeypatch.setenv("REDIS_PASSWORD", "test_password")
cache = RedisSemanticCache(similarity_threshold=0.8)
with pytest.raises(ConnectionError, match="redis down"):
_ = cache.llmcache
assert cache.llmcache is built_cache
assert semantic_cache_mock.call_count == 2
def test_redis_llmcache_setter_supported():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
cache = RedisSemanticCache.__new__(RedisSemanticCache)
sentinel = MagicMock()
cache.llmcache = sentinel
assert cache.llmcache is sentinel
def _router_proxy_module(router, model_name):
import types
fake_proxy = types.ModuleType("litellm.proxy.proxy_server")
fake_proxy.llm_router = router
fake_proxy.llm_model_list = [{"model_name": model_name}]
return fake_proxy
def test_redis_sync_embedding_call_is_bounded(monkeypatch):
import sys
from litellm.caching.redis_semantic_cache import RedisSemanticCache
cache = RedisSemanticCache.__new__(RedisSemanticCache)
cache.embedding_model = "sem-embed"
cache.embedding_timeout = 1.5
router = MagicMock()
router.get_configured_token_limits.return_value = (None, None)
router.embedding = MagicMock(return_value={"data": [{"embedding": [0.5, 0.6]}]})
monkeypatch.setitem(
sys.modules,
"litellm.proxy.proxy_server",
_router_proxy_module(router, "sem-embed"),
)
assert cache._get_embedding("hello") == [0.5, 0.6]
assert router.embedding.call_args.kwargs["timeout"] == 1.5
assert router.embedding.call_args.kwargs["num_retries"] == 0
@pytest.mark.asyncio
async def test_redis_async_embedding_call_is_bounded(monkeypatch):
import sys
from litellm.caching.redis_semantic_cache import RedisSemanticCache
cache = RedisSemanticCache.__new__(RedisSemanticCache)
cache.embedding_model = "sem-embed"
cache.embedding_timeout = 1.5
router = MagicMock()
router.get_configured_token_limits.return_value = (None, None)
router.aembedding = AsyncMock(return_value={"data": [{"embedding": [0.5, 0.6]}]})
monkeypatch.setitem(
sys.modules,
"litellm.proxy.proxy_server",
_router_proxy_module(router, "sem-embed"),
)
assert await cache._get_async_embedding("hello") == [0.5, 0.6]
assert router.aembedding.call_args.kwargs["timeout"] == 1.5
assert router.aembedding.call_args.kwargs["num_retries"] == 0
@pytest.mark.asyncio
async def test_redis_async_embedding_gives_up_on_unresponsive_endpoint(monkeypatch):
import asyncio
import sys
import time
from litellm.caching.redis_semantic_cache import RedisSemanticCache
cache = RedisSemanticCache.__new__(RedisSemanticCache)
cache.embedding_model = "sem-embed"
cache.embedding_timeout = 0.05
async def never_responds(**kwargs):
await asyncio.sleep(3)
return {"data": [{"embedding": [0.1, 0.2]}]}
router = MagicMock()
router.get_configured_token_limits.return_value = (None, None)
router.aembedding = never_responds
monkeypatch.setitem(
sys.modules,
"litellm.proxy.proxy_server",
_router_proxy_module(router, "sem-embed"),
)
started = time.monotonic()
with pytest.raises(ValueError, match="Failed to generate embedding"):
await cache._get_async_embedding("hello")
assert time.monotonic() - started < 1.0
@pytest.mark.asyncio
async def test_redis_async_get_cache_fails_open_when_embedding_hangs(monkeypatch):
import asyncio
import sys
import time
from litellm.caching.redis_semantic_cache import RedisSemanticCache
cache = RedisSemanticCache.__new__(RedisSemanticCache)
cache.embedding_model = "sem-embed"
cache.embedding_timeout = 0.05
cache.similarity_threshold = 0.8
cache.distance_threshold = 0.2
cache.llmcache = MagicMock()
async def never_responds(**kwargs):
await asyncio.sleep(3)
return {"data": [{"embedding": [0.1, 0.2]}]}
router = MagicMock()
router.get_configured_token_limits.return_value = (None, None)
router.aembedding = never_responds
monkeypatch.setitem(
sys.modules,
"litellm.proxy.proxy_server",
_router_proxy_module(router, "sem-embed"),
)
metadata = {}
started = time.monotonic()
result = await cache.async_get_cache(
key="test_key",
messages=[{"role": "user", "content": "What is the capital of France?"}],
metadata=metadata,
)
elapsed = time.monotonic() - started
assert result is None
assert metadata["semantic-similarity"] == 0.0
assert elapsed < 1.0
cache.llmcache.acheck.assert_not_called()
def test_cache_forwards_semantic_cache_embedding_timeout():
from litellm.caching.caching import Cache
from litellm.types.caching import LiteLLMCacheType
with patch("litellm.caching.caching.RedisSemanticCache") as backend:
Cache(
type=LiteLLMCacheType.REDIS_SEMANTIC,
similarity_threshold=0.8,
redis_url="redis://localhost:6379",
semantic_cache_embedding_timeout=2.5,
)
assert backend.call_args.kwargs["embedding_timeout"] == 2.5
def test_redis_semantic_cache_defaults_embedding_timeout():
from litellm.caching.redis_semantic_cache import RedisSemanticCache
from litellm.constants import SEMANTIC_CACHE_EMBEDDING_TIMEOUT_SECONDS
cache = RedisSemanticCache.__new__(RedisSemanticCache)
assert cache.embedding_timeout == SEMANTIC_CACHE_EMBEDDING_TIMEOUT_SECONDS
assert SEMANTIC_CACHE_EMBEDDING_TIMEOUT_SECONDS < 60
Reference in a new issue View git blame Copy permalink