feat: record SemanticCache operation-duration and distance metrics (FDE-87)

redisvl/extensions/cache/llm/_telemetry.py

@@ -0,0 +1,116 @@
+"""OpenTelemetry metrics for SemanticCache. Internal.
+
+Attributes stay low-cardinality: the cache name, the operation, the lookup
+result and the error type. Prompts, filters, entry ids and distances never
+become metric attributes.
+"""
+
+import functools
+import inspect
+from time import perf_counter
+from typing import Any, Callable, ParamSpec, TypeVar, cast
+
+from redisvl.query import VectorQuery
+from redisvl.utils import otel
+from redisvl.utils.log import get_logger
+
+logger = get_logger(__name__)
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+
+def _result(hits: list[Any]) -> str:
+    return "hit" if hits else "miss"
+
+
+def _record_duration(
+    cache: Any,
+    operation: str,
+    start: float,
+    hits: list[Any] | None = None,
+    error: Exception | None = None,
+) -> None:
+    # Telemetry must never change what the caller sees, so a failure here
+    # (an old opentelemetry-api installed by another package, say) is dropped.
+    try:
+        attributes = {otel.CACHE_NAME: cache.name, otel.CACHE_OPERATION: operation}
+        if error is not None:
+            attributes[otel.ERROR_TYPE] = otel.error_type(error)
+        elif hits is not None:
+            attributes[otel.CACHE_RESULT] = _result(hits)
+        otel.histogram(otel.SEMANTIC_CACHE_OPERATION_DURATION).record(
+            perf_counter() - start, attributes
+        )
+    except Exception:
+        logger.debug("Could not record a SemanticCache metric", exc_info=True)
+
+
+def timed(
+    operation: str, *, lookup: bool = False
+) -> Callable[[Callable[P, R]], Callable[P, R]]:
+    """Record the duration of a sync or async SemanticCache method.
+
+    A lookup also records whether it hit, judged from the hits it returns. A
+    failure records ``error.type`` instead and re-raises unchanged. A call that
+    never completes, such as a cancelled task, records nothing.
+    """
+
+    def decorate(method: Callable[P, R]) -> Callable[P, R]:
+        if inspect.iscoroutinefunction(method):
+
+            @functools.wraps(method)
+            async def async_wrapper(cache: Any, *args: Any, **kwargs: Any) -> Any:
+                start = perf_counter()
+                try:
+                    result = await method(cache, *args, **kwargs)
+                except Exception as error:
+                    _record_duration(cache, operation, start, error=error)
+                    raise
+                _record_duration(
+                    cache, operation, start, hits=result if lookup else None
+                )
+                return result
+
+            return cast(Callable[P, R], async_wrapper)
+
+        @functools.wraps(method)
+        def wrapper(cache: Any, *args: Any, **kwargs: Any) -> Any:
+            start = perf_counter()
+            try:
+                result: Any = method(cache, *args, **kwargs)
+            except Exception as error:
+                _record_duration(cache, operation, start, error=error)
+                raise
+            _record_duration(cache, operation, start, hits=result if lookup else None)
+            return result
+
+        return cast(Callable[P, R], wrapper)
+
+    return decorate
+
+
+def record_nearest_distance(
+    cache_name: str, candidates: list[dict[str, Any]], hits: list[Any]
+) -> None:
+    """Record the nearest candidate's distance, on a miss as well as a hit.
+
+    Nothing is recorded when there was no candidate at all: an empty cache, or
+    a filter that excluded every entry. Candidates arrive nearest first.
+    """
+    try:
+        distance = next(
+            (
+                candidate[VectorQuery.DISTANCE_ID]
+                for candidate in candidates
+                if candidate.get(VectorQuery.DISTANCE_ID) is not None
+            ),
+            None,
+        )
+        if distance is not None:
+            otel.histogram(otel.SEMANTIC_CACHE_DISTANCE).record(
+                float(distance),
+                {otel.CACHE_NAME: cache_name, otel.CACHE_RESULT: _result(hits)},
+            )
+    except Exception:
+        logger.debug("Could not record a SemanticCache metric", exc_info=True)

redisvl/extensions/cache/llm/semantic.py

@@ -3,6 +3,7 @@
 from pydantic import ValidationError
 from redis import Redis
 
+from redisvl.extensions.cache.llm import _telemetry
 from redisvl.extensions.cache.llm.base import BaseLLMCache
 from redisvl.extensions.cache.llm.schema import (
     CacheEntry,
@@ -350,6 +351,7 @@ async def aclear(self) -> None:
         """
         await super().aclear()
 
+    @_telemetry.timed("drop")
     def drop(self, ids: list[str] | None = None, keys: list[str] | None = None) -> None:
         """Drop specific entries from the cache by ID or Redis key.
 
@@ -374,6 +376,7 @@ def drop(self, ids: list[str] | None = None, keys: list[str] | None = None) -> N
         if keys is not None:
             self._index.drop_keys(keys)
 
+    @_telemetry.timed("drop")
     async def adrop(
         self, ids: list[str] | None = None, keys: list[str] | None = None
     ) -> None:
@@ -428,6 +431,7 @@ def _check_vector_dims(self, vector: list[float]):
         ].attrs.dims  # type: ignore
         validate_vector_dims(len(vector), schema_vector_dims)
 
+    @_telemetry.timed("check", lookup=True)
     def check(
         self,
         prompt: str | None = None,
@@ -508,9 +512,11 @@ def check(
 
         # Refresh TTL on all found keys
         self._expire_many(redis_keys)
+        _telemetry.record_nearest_distance(self.name, candidates, cache_hits)
 
         return cache_hits
 
+    @_telemetry.timed("check", lookup=True)
     async def acheck(
         self,
         prompt: str | None = None,
@@ -593,6 +599,7 @@ async def acheck(
 
         # Refresh TTL on all found keys
         await self._aexpire_many(redis_keys)
+        _telemetry.record_nearest_distance(self.name, candidates, cache_hits)
 
         return cache_hits
 
@@ -696,6 +703,7 @@ def _process_cache_results(
 
         return redis_keys, cache_hits
 
+    @_telemetry.timed("store")
     def store(
         self,
         prompt: str,
@@ -764,6 +772,7 @@ def store(
         # Return the key where the entry was stored
         return keys[0]
 
+    @_telemetry.timed("store")
     async def astore(
         self,
         prompt: str,
@@ -834,6 +843,7 @@ async def astore(
         # Return the key where the entry was stored
         return keys[0]
 
+    @_telemetry.timed("update")
     def update(self, key: str, **kwargs) -> None:
         """Update specific fields within an existing cache entry. If no fields
         are passed, then only the document TTL is refreshed.
@@ -875,6 +885,7 @@ def update(self, key: str, **kwargs) -> None:
         # Refresh TTL regardless of whether fields were updated
         self.expire(key)
 
+    @_telemetry.timed("update")
     async def aupdate(self, key: str, **kwargs) -> None:
         """Async update specific fields within an existing cache entry. If no fields
         are passed, then only the document TTL is refreshed.

redisvl/utils/otel.py

@@ -40,7 +40,7 @@ class SpanKind(Enum):  # type: ignore[no-redef]
 
 
 if TYPE_CHECKING:
-    from opentelemetry.metrics import Meter
+    from opentelemetry.metrics import Histogram, Meter
     from opentelemetry.trace import Tracer
 
 INSTRUMENTATION_SCOPE = "redisvl"
@@ -76,6 +76,27 @@ class SpanKind(Enum):  # type: ignore[no-redef]
 GEN_AI_RETRIEVAL_TOP_K = "gen_ai.retrieval.top_k"
 
 
+# Bucket advice. The GenAI duration buckets start at 10 ms, but a cache lookup
+# on local Redis takes about 250 us excluding embedding, so these start at
+# 100 us. Distance buckets are
+# dense where cosine thresholds usually sit and coarse up to the maximum of 2.
+# Applications can override both with SDK views.
+_HISTOGRAMS = {
+    SEMANTIC_CACHE_OPERATION_DURATION: (
+        "s",
+        "Duration of SemanticCache operations.",
+        (0.0001, 0.00025, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05)
+        + (0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0),
+    ),
+    SEMANTIC_CACHE_DISTANCE: (
+        "1",
+        "Vector distance of the nearest SemanticCache candidate per lookup.",
+        tuple(i / 100 for i in range(1, 11))
+        + (0.125, 0.15, 0.175, 0.2, 0.25, 0.3, 0.4, 0.5, 0.75, 1.0, 1.5, 2.0),
+    ),
+}
+
+
 class _NoOpSpan(ContextDecorator):
     """A span that records nothing, usable as its own context manager or
     decorator, like ``start_as_current_span``'s result."""
@@ -140,3 +161,31 @@ def get_meter() -> "Meter":
     if not _enabled():
         return _NoOpMeter()  # type: ignore[return-value]
     return metrics.get_meter(INSTRUMENTATION_SCOPE, __version__, schema_url=SCHEMA_URL)
+
+
+@lru_cache(maxsize=None)
+def histogram(name: str) -> "Histogram":
+    """Return one of RedisVL's histograms, created once on its meter."""
+    unit, description, boundaries = _HISTOGRAMS[name]
+    return get_meter().create_histogram(
+        name,
+        unit=unit,
+        description=description,
+        explicit_bucket_boundaries_advisory=boundaries,
+    )
+
+
+def error_type(error: BaseException) -> str:
+    """Name an exception's type as the SDK names it on exception events: the
+    qualified class name, prefixed by its module unless it is a builtin."""
+    cls = type(error)
+    if cls.__module__ == "builtins":
+        return cls.__qualname__
+    return f"{cls.__module__}.{cls.__qualname__}"
+
+
+def _clear_caches() -> None:
+    """Forget the cached tracer, meter and instruments. For tests."""
+    get_tracer.cache_clear()
+    get_meter.cache_clear()
+    histogram.cache_clear()

tests/conftest.py

@@ -326,8 +326,7 @@ def reset():
         metrics_internal._METER_PROVIDER_SET_ONCE = Once()
         metrics_internal._METER_PROVIDER = None
         metrics_internal._PROXY_METER_PROVIDER = metrics_internal._ProxyMeterProvider()
-        otel.get_tracer.cache_clear()
-        otel.get_meter.cache_clear()
+        otel._clear_caches()
 
     spans, reader = InMemorySpanExporter(), InMemoryMetricReader()
     tracer_provider = TracerProvider()