diff --git a/litellm/caching/caching.py b/litellm/caching/caching.py
index d6dd2a073af..be82f5def1f 100644
--- a/litellm/caching/caching.py
+++ b/litellm/caching/caching.py
@@ -679,7 +679,14 @@ class Cache:
             cache_key, cached_data, kwargs = self._add_cache_logic(result=result, **kwargs)
             self.cache.set_cache(cache_key, cached_data, **kwargs)
         except Exception as e:
-            log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Cache: exception in add_cache", e)
+            self._log_add_cache_failure(e)
+
+    def _log_add_cache_failure(self, exc: Exception) -> None:
+        message: Final = "LiteLLM Cache: exception in add_cache"
+        if isinstance(self.cache, RedisCache):
+            log_redis_failure(verbose_logger, logging.ERROR, message, exc)
+            return
+        verbose_logger.error("%s: %s", message, exc)
 
     async def async_add_cache(self, result, dynamic_cache_object: BaseCache | None = None, **kwargs):
         """
@@ -698,7 +705,7 @@ class Cache:
                 else:
                     await self.cache.async_set_cache(cache_key, cached_data, **kwargs)
         except Exception as e:
-            log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Cache: exception in add_cache", e)
+            self._log_add_cache_failure(e)
 
     def _convert_to_cached_embedding(
         self,
@@ -877,7 +884,7 @@ class Cache:
             else:
                 await self.cache.async_set_cache_pipeline(cache_list=cache_list, **kwargs)
         except Exception as e:
-            log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Cache: exception in add_cache", e)
+            self._log_add_cache_failure(e)
 
     def should_use_cache(self, **kwargs):
         """
diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py
index d0cefcb6086..b4b2b1a334c 100644
--- a/litellm/caching/redis_cache.py
+++ b/litellm/caching/redis_cache.py
@@ -15,6 +15,7 @@ import hashlib
 import inspect
 import json
 import logging
+import threading
 import time
 from collections.abc import Awaitable, Callable, Iterator, Sequence
 from contextvars import ContextVar
@@ -32,6 +33,7 @@ from litellm.constants import (
     REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD,
     REDIS_CIRCUIT_BREAKER_RECOVERY_TIMEOUT,
     REDIS_CIRCUIT_BREAKER_TIMEOUT_MIN_DURATION,
+    REDIS_TIMEOUT_LOG_INTERVAL,
 )
 from litellm.litellm_core_utils.core_helpers import _get_parent_otel_span_from_kwargs
 from litellm.litellm_core_utils.coroutine_checker import coroutine_checker
@@ -340,7 +342,7 @@ def _explicit_causes(exc: BaseException) -> Iterator[BaseException]:
         current = current.__cause__
 
 
-def _is_redis_timeout_failure(exc: BaseException) -> bool:
+def is_redis_timeout_failure(exc: BaseException) -> bool:
     """True when ``exc`` or any exception it was explicitly raised ``from`` is a timeout.
 
     redis-py's blocking pool reports a pool wait timeout as ``ConnectionError`` chained from
@@ -414,7 +416,7 @@ def _record_swallowed_redis_failure(breaker: RedisCircuitBreaker, exc: BaseExcep
     """
     if not _is_redis_health_failure(exc):
         return
-    breaker.record_failure(is_timeout=_is_redis_timeout_failure(exc))
+    breaker.record_failure(is_timeout=is_redis_timeout_failure(exc))
     _swallowed_redis_failures.set(_swallowed_redis_failures.get() + 1)
 
 
@@ -422,13 +424,58 @@ class RedisCircuitBreakerOpenError(Exception):
     pass
 
 
+class _RedisTimeoutLogThrottle:
+    """Admits one Redis timeout log line per interval and counts the timeouts it suppressed in between."""
+
+    def __init__(self, interval: float, clock: Callable[[], float] = time.monotonic) -> None:
+        self.interval = interval
+        self._clock = clock
+        self._lock = threading.Lock()
+        self._last_logged_at: float | None = None
+        self._suppressed = 0
+
+    def admit(self) -> int | None:
+        """Return the number of timeouts suppressed since the last admitted line, or None to suppress this one."""
+        with self._lock:
+            now: Final = self._clock()
+            if self._last_logged_at is not None and now - self._last_logged_at < self.interval:
+                self._suppressed += 1
+                return None
+            suppressed: Final = self._suppressed
+            self._suppressed = 0
+            self._last_logged_at = now
+            return suppressed
+
+
+_redis_timeout_log_throttle: Final = _RedisTimeoutLogThrottle(REDIS_TIMEOUT_LOG_INTERVAL)
+
+
 def log_redis_failure(
     logger: logging.Logger, level: int, message: str, exc: BaseException, with_traceback: bool = False
 ) -> None:
     if isinstance(exc, RedisCircuitBreakerOpenError):
-        logger.debug("%s: %s", message, exc)
+        logger.debug("%s: %s", message, exc, stacklevel=2)
         return
-    logger.log(level, "%s: %s", message, exc, exc_info=exc if with_traceback else None)
+    exc_info: Final = exc if with_traceback else None
+    if not is_redis_timeout_failure(exc):
+        logger.log(level, "%s: %s", message, exc, exc_info=exc_info, stacklevel=2)
+        return
+    suppressed: Final = _redis_timeout_log_throttle.admit()
+    if suppressed is None:
+        logger.debug("%s: %s", message, exc, stacklevel=2)
+        return
+    if suppressed == 0:
+        logger.log(level, "%s: %s", message, exc, exc_info=exc_info, stacklevel=2)
+        return
+    logger.log(
+        level,
+        "%s: %s (%d more Redis timeouts since the previous Redis timeout line were logged at DEBUG)",
+        message,
+        exc,
+        suppressed,
+        exc_info=exc_info,
+        stacklevel=2,
+    )
 
 
 @dataclass(frozen=True, slots=True)
@@ -475,7 +522,7 @@ async def _run_under_circuit_breaker(
         result: Final = await call()
     except Exception as e:
         if _is_redis_health_failure(e):
-            breaker.record_failure(is_timeout=_is_redis_timeout_failure(e))
+            breaker.record_failure(is_timeout=is_redis_timeout_failure(e))
         raise
     _exit_circuit_breaker(breaker, admission)
     return result
@@ -492,7 +539,7 @@ def _run_under_circuit_breaker_sync(
         result: Final = call()
     except Exception as e:
         if _is_redis_health_failure(e):
-            breaker.record_failure(is_timeout=_is_redis_timeout_failure(e))
+            breaker.record_failure(is_timeout=is_redis_timeout_failure(e))
         raise
     _exit_circuit_breaker(breaker, admission)
     return result
@@ -801,10 +848,8 @@ class RedisCache(BaseCache):
             ## LOGGING ##
             end_time = time.time()
             _duration = end_time - start_time
-            verbose_logger.error(
-                "LiteLLM Redis Caching: increment_cache() - Got exception from REDIS %s, Writing value=%s",
-                str(e),
-                value,
+            log_redis_failure(
+                verbose_logger, logging.ERROR, "LiteLLM Redis Caching: increment_cache() - Got exception from REDIS", e
             )
             raise e
 
@@ -1010,11 +1055,8 @@ class RedisCache(BaseCache):
                     call_type=f"async_set_cache <- {_get_call_stack_info()}",
                 )
             )
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async set() - Got exception from REDIS %s, key=%r, value=%r",
-                str(e),
-                key,
-                value,
+            log_redis_failure(
+                verbose_logger, logging.ERROR, "LiteLLM Redis Caching: async set() - Got exception from REDIS", e
             )
             raise e
 
@@ -1062,10 +1104,8 @@ class RedisCache(BaseCache):
                     event_metadata={"key": key},
                 )
             )
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async set() - Got exception from REDIS %s, Writing value=%s",
-                str(e),
-                value,
+            log_redis_failure(
+                verbose_logger, logging.ERROR, "LiteLLM Redis Caching: async set() - Got exception from REDIS", e
             )
             _record_swallowed_redis_failure(self._circuit_breaker, e)
 
@@ -1112,7 +1152,6 @@ class RedisCache(BaseCache):
         start_time: Final = time.time()
 
         print_verbose(f"Set Async Redis Cache: key list: {cache_list}\nttl={ttl}, redis_version={self.redis_version}")
-        cache_value: Final = None
         try:
             async with _redis_client.pipeline(transaction=False) as pipe:
                 results: Final = await self._pipeline_helper(pipe, cache_list, ttl)
@@ -1149,10 +1188,11 @@ class RedisCache(BaseCache):
                 )
             )
 
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async set_cache_pipeline() - Got exception from REDIS %s, Writing value=%s",
-                str(e),
-                cache_value,
+            log_redis_failure(
+                verbose_logger,
+                logging.ERROR,
+                "LiteLLM Redis Caching: async set_cache_pipeline() - Got exception from REDIS",
+                e,
             )
             _record_swallowed_redis_failure(self._circuit_breaker, e)
 
@@ -1191,8 +1231,11 @@ class RedisCache(BaseCache):
                     end_time=time.time(),
                 )
             )
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async_set_cache_pipeline_with_ttls() - Got exception from REDIS %s", str(e)
+            log_redis_failure(
+                verbose_logger,
+                logging.ERROR,
+                "LiteLLM Redis Caching: async_set_cache_pipeline_with_ttls() - Got exception from REDIS",
+                e,
             )
             _record_swallowed_redis_failure(self._circuit_breaker, e)
 
@@ -1235,10 +1278,8 @@ class RedisCache(BaseCache):
                 )
             )
             # NON blocking - notify users Redis is throwing an exception
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async set() - Got exception from REDIS %s, Writing value=%s",
-                str(e),
-                value,
+            log_redis_failure(
+                verbose_logger, logging.ERROR, "LiteLLM Redis Caching: async set() - Got exception from REDIS", e
             )
             raise e
 
@@ -1274,10 +1315,11 @@ class RedisCache(BaseCache):
                 )
             )
             # NON blocking - notify users Redis is throwing an exception
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async set_cache_sadd() - Got exception from REDIS %s, Writing value=%s",
-                str(e),
-                value,
+            log_redis_failure(
+                verbose_logger,
+                logging.ERROR,
+                "LiteLLM Redis Caching: async set_cache_sadd() - Got exception from REDIS",
+                e,
             )
             _record_swallowed_redis_failure(self._circuit_breaker, e)
 
@@ -1359,10 +1401,11 @@ class RedisCache(BaseCache):
                     parent_otel_span=parent_otel_span,
                 )
             )
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async async_increment() - Got exception from REDIS %s, Writing value=%s",
-                str(e),
-                value,
+            log_redis_failure(
+                verbose_logger,
+                logging.ERROR,
+                "LiteLLM Redis Caching: async async_increment() - Got exception from REDIS",
+                e,
             )
             raise e
 
@@ -1448,7 +1491,9 @@ class RedisCache(BaseCache):
             print_verbose(f"Got Redis Cache: key: {key}, cached_response {cached_response}")
             return self._get_cache_logic(cached_response=cached_response)
         except Exception as e:
-            verbose_logger.error("litellm.caching.caching: get() - Got exception from REDIS: %s", e)
+            log_redis_failure(
+                verbose_logger, logging.ERROR, "litellm.caching.caching: get() - Got exception from REDIS", e
+            )
             _record_swallowed_redis_failure(self._circuit_breaker, e)
 
     def _run_redis_mget_operation(self, keys: list[str]) -> Sequence[bytes | str | None]:
@@ -1526,7 +1571,7 @@ class RedisCache(BaseCache):
                 end_time=failed_at,
                 parent_otel_span=parent_otel_span,
             )
-            verbose_logger.error("Error occurred in batch get cache - %s", e)
+            log_redis_failure(verbose_logger, logging.ERROR, "Error occurred in batch get cache", e)
             _record_swallowed_redis_failure(self._circuit_breaker, e)
             return key_value_dict
 
@@ -1645,7 +1690,7 @@ class RedisCache(BaseCache):
                     parent_otel_span=parent_otel_span,
                 )
             )
-            verbose_logger.error("Error occurred in async batch get cache - %s", e)
+            log_redis_failure(verbose_logger, logging.ERROR, "Error occurred in async batch get cache", e)
             _record_swallowed_redis_failure(self._circuit_breaker, e)
             return key_value_dict
 
@@ -1870,9 +1915,11 @@ class RedisCache(BaseCache):
                     parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs),
                 )
             )
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async increment_pipeline() - Got exception from REDIS %s",
-                str(e),
+            log_redis_failure(
+                verbose_logger,
+                logging.ERROR,
+                "LiteLLM Redis Caching: async increment_pipeline() - Got exception from REDIS",
+                e,
             )
             raise e
 
@@ -1949,7 +1996,7 @@ class RedisCache(BaseCache):
                     call_type=f"async_rpush <- {_get_call_stack_info()}",
                 )
             )
-            verbose_logger.error("LiteLLM Redis Cache RPUSH: - Got exception from REDIS : %s", e)
+            log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Redis Cache RPUSH: - Got exception from REDIS", e)
             raise e
 
     async def _pipeline_rpush_helper(
@@ -2017,9 +2064,11 @@ class RedisCache(BaseCache):
                     call_type=f"async_rpush_pipeline <- {_get_call_stack_info()}",
                 )
             )
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async_rpush_pipeline() - Got exception from REDIS %s",
-                str(e),
+            log_redis_failure(
+                verbose_logger,
+                logging.ERROR,
+                "LiteLLM Redis Caching: async_rpush_pipeline() - Got exception from REDIS",
+                e,
             )
             raise e
 
@@ -2095,7 +2144,7 @@ class RedisCache(BaseCache):
                     call_type=f"async_lpop <- {_get_call_stack_info()}",
                 )
             )
-            verbose_logger.error("LiteLLM Redis Cache LPOP: - Got exception from REDIS : %s", e)
+            log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Redis Cache LPOP: - Got exception from REDIS", e)
             raise e
 
     async def _pipeline_lpop_helper(
@@ -2206,8 +2255,10 @@ class RedisCache(BaseCache):
                     call_type=f"async_lpop_pipeline <- {_get_call_stack_info()}",
                 )
             )
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async_lpop_pipeline() - Got exception from REDIS %s",
-                str(e),
+            log_redis_failure(
+                verbose_logger,
+                logging.ERROR,
+                "LiteLLM Redis Caching: async_lpop_pipeline() - Got exception from REDIS",
+                e,
             )
             raise e
diff --git a/litellm/constants.py b/litellm/constants.py
index 5751e6e46af..b370119404a 100644
--- a/litellm/constants.py
+++ b/litellm/constants.py
@@ -461,6 +461,7 @@ REDIS_CIRCUIT_BREAKER_ENABLED: Final = os.getenv("REDIS_CIRCUIT_BREAKER_ENABLED"
 # minimum seconds a timeout-only failure streak must span before it can open the breaker,
 # so one event-loop stall timing out many queued calls at once does not trip it
 REDIS_CIRCUIT_BREAKER_TIMEOUT_MIN_DURATION: Final = float(os.getenv("REDIS_CIRCUIT_BREAKER_TIMEOUT_MIN_DURATION", 5.0))
+REDIS_TIMEOUT_LOG_INTERVAL: Final = float(os.getenv("REDIS_TIMEOUT_LOG_INTERVAL", "5.0"))
 # Seconds of idle before a Redis cluster connection is validated with a PING and
 # reconnected if dead, so a connection silently dropped by a cluster restart
 # (e.g. ElastiCache Serverless maintenance) is not reused while broken
diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py
index e7b8dd01dae..25274dec1c6 100644
--- a/litellm/proxy/proxy_server.py
+++ b/litellm/proxy/proxy_server.py
@@ -250,7 +250,7 @@ import litellm._redis
 from litellm import Router
 from litellm._logging import _redact_string, verbose_proxy_logger, verbose_router_logger
 from litellm.caching.caching import DualCache, RedisCache
-from litellm.caching.redis_cache import RedisCircuitBreakerOpenError
+from litellm.caching.redis_cache import RedisCircuitBreakerOpenError, is_redis_timeout_failure
 from litellm.caching.redis_cluster_cache import RedisClusterCache
 from litellm.constants import (
     _REALTIME_BODY_CACHE_SIZE,
@@ -3450,8 +3450,10 @@ async def _invalidate_spend_counter(counter_key: str):
 async def _apply_spend_counter_increments(pending: Sequence[PendingSpendIncrement]) -> None:
     try:
         await increment_spend_counters_pipeline(pending=pending)
-    except RedisCircuitBreakerOpenError:
-        return
+    except Exception as e:
+        if isinstance(e, RedisCircuitBreakerOpenError) or is_redis_timeout_failure(e):
+            return
+        raise
 
 
 async def increment_spend_counters_pipeline(pending: Sequence[PendingSpendIncrement]) -> None:
diff --git a/tests/test_litellm/caching/test_caching.py b/tests/test_litellm/caching/test_caching.py
index 4d0ec0fb677..c7ec8abc31e 100644
--- a/tests/test_litellm/caching/test_caching.py
+++ b/tests/test_litellm/caching/test_caching.py
@@ -1,9 +1,12 @@
 import logging
 import re
+from unittest.mock import MagicMock
 
 import pytest
 
+import litellm.caching.redis_cache as redis_cache_module
 from litellm.caching.caching import Cache
+from litellm.caching.redis_cache import RedisCache, _RedisTimeoutLogThrottle
 from litellm.types.caching import LiteLLMCacheType, SemanticCacheScope
 from litellm.types.utils import Embedding, EmbeddingResponse, Usage
 
@@ -53,6 +56,29 @@ def test_cache_key_debug_log_does_not_include_prompt_material(caplog):
     assert any(cache_key in message for message in created_cache_key_logs)
 
 
+@pytest.mark.parametrize(
+    ("backend", "expected_level"),
+    [
+        pytest.param(MagicMock(spec=RedisCache), logging.DEBUG, id="redis_backend_is_throttled"),
+        pytest.param(MagicMock(), logging.ERROR, id="other_backend_logs_every_timeout"),
+    ],
+)
+def test_add_cache_timeout_only_joins_redis_throttle_for_redis_backends(backend, expected_level, caplog, monkeypatch):
+    throttle = _RedisTimeoutLogThrottle(interval=5.0, clock=MagicMock(return_value=1_000.0))
+    assert throttle.admit() == 0
+    monkeypatch.setattr(redis_cache_module, "_redis_timeout_log_throttle", throttle)
+
+    cache = Cache(type=LiteLLMCacheType.LOCAL)
+    backend.set_cache.side_effect = TimeoutError("lit7520 backend timed out")
+    cache.cache = backend
+
+    with caplog.at_level(logging.DEBUG, logger="LiteLLM"):
+        cache.add_cache("result", model="gpt-4.1-mini", messages=[{"role": "user", "content": "hi"}])
+
+    records = [r for r in caplog.records if "lit7520 backend timed out" in r.getMessage()]
+    assert [r.levelno for r in records] == [expected_level]
+
+
 def _embedding_response(prompt_tokens, num_items):
     return EmbeddingResponse(
         model="amazon.titan-embed-image-v1",
diff --git a/tests/test_litellm/caching/test_dual_cache.py b/tests/test_litellm/caching/test_dual_cache.py
index 4c9068722b8..95395878c25 100644
--- a/tests/test_litellm/caching/test_dual_cache.py
+++ b/tests/test_litellm/caching/test_dual_cache.py
@@ -704,3 +704,58 @@ async def test_open_breaker_keeps_async_batch_read_memory_hits_and_releases_rese
 
     assert list(await cache.async_batch_get_cache(["k1", "k2"])) == ["v1", None]
     assert "k2" not in cache.last_redis_batch_access_time
+
+
+@pytest.mark.asyncio
+async def test_redis_timeouts_falling_back_to_memory_log_once_per_interval(caplog, monkeypatch):
+    """The first fallback WARNING of a timeout streak logs, the rest stay at DEBUG until the summary."""
+    from redis.exceptions import TimeoutError as RedisTimeoutError
+
+    from litellm.caching import redis_cache as redis_cache_module
+    from litellm.caching.redis_cache import _RedisTimeoutLogThrottle
+
+    clock = MagicMock(return_value=1_000.0)
+    monkeypatch.setattr(
+        redis_cache_module, "_redis_timeout_log_throttle", _RedisTimeoutLogThrottle(interval=5.0, clock=clock)
+    )
+
+    class _TimingOutRedis:
+        async def async_increment_pipeline(self, increment_list, **kwargs):
+            raise RedisTimeoutError("Timeout reading from 127.0.0.1:6379")
+
+        async def async_increment(self, key, value, **kwargs):
+            raise RedisTimeoutError("Timeout reading from 127.0.0.1:6379")
+
+    cache = DualCache(
+        in_memory_cache=InMemoryCache(),
+        redis_cache=_TimingOutRedis(),  # pyright: ignore[reportArgumentType]  # duck-typed Redis double
+    )
+    increments = [RedisPipelineIncrementOperation(key="k", increment_value=1.0, ttl=60)]
+
+    with caplog.at_level(logging.DEBUG, logger="LiteLLM"):
+        for _ in range(100):
+            await cache.async_increment_cache_pipeline(increment_list=increments)
+            await cache.async_increment_cache("k", 1.0)
+
+    visible = [r for r in caplog.records if r.levelno >= logging.WARNING]
+    assert [(r.levelno, r.getMessage()) for r in visible] == [
+        (
+            logging.WARNING,
+            "Redis async_increment_cache_pipeline failed, falling back to in-memory result:"
+            " Timeout reading from 127.0.0.1:6379",
+        )
+    ]
+    assert visible[0].filename == "dual_cache.py"
+    assert sum("Timeout reading from" in r.getMessage() for r in caplog.records) == 200
+
+    caplog.clear()
+    clock.return_value += 5.0
+    with caplog.at_level(logging.DEBUG, logger="LiteLLM"):
+        await cache.async_increment_cache("k", 1.0)
+    assert [(r.levelno, r.getMessage()) for r in caplog.records] == [
+        (
+            logging.WARNING,
+            "Redis async_increment_cache failed, falling back to in-memory result: Timeout reading from 127.0.0.1:6379"
+            " (199 more Redis timeouts since the previous Redis timeout line were logged at DEBUG)",
+        )
+    ]
diff --git a/tests/test_litellm/caching/test_redis_cache.py b/tests/test_litellm/caching/test_redis_cache.py
index c1e3240adb7..19638c60b4b 100644
--- a/tests/test_litellm/caching/test_redis_cache.py
+++ b/tests/test_litellm/caching/test_redis_cache.py
@@ -978,17 +978,17 @@ async def test_stale_timeout_does_not_let_sub_threshold_hard_failures_open_the_b
     from redis.exceptions import ConnectionError as RedisConnectionError
     from redis.exceptions import TimeoutError as RedisTimeoutError
 
-    from litellm.caching.redis_cache import RedisCircuitBreaker, _is_redis_timeout_failure
+    from litellm.caching.redis_cache import RedisCircuitBreaker, is_redis_timeout_failure
 
     breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60, timeout_min_duration=0.05)
 
-    breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisTimeoutError("read timed out")))
+    breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisTimeoutError("read timed out")))
     await asyncio.sleep(0.06)
     for _ in range(breaker.failure_threshold - 1):
-        breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisConnectionError("refused")))
+        breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisConnectionError("refused")))
     assert breaker.is_open() is False, "2 hard failures and 1 stale timeout are below both thresholds"
 
-    breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisConnectionError("refused")))
+    breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisConnectionError("refused")))
     assert breaker.is_open() is True, "the threshold-th hard failure must still open it"
 
 
@@ -1000,19 +1000,19 @@ async def test_hard_failure_resets_timeout_streak_so_a_later_burst_must_earn_its
     from redis.exceptions import ConnectionError as RedisConnectionError
     from redis.exceptions import TimeoutError as RedisTimeoutError
 
-    from litellm.caching.redis_cache import RedisCircuitBreaker, _is_redis_timeout_failure
+    from litellm.caching.redis_cache import RedisCircuitBreaker, is_redis_timeout_failure
 
     breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60, timeout_min_duration=0.05)
 
-    breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisTimeoutError("read timed out")))
-    breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisConnectionError("refused")))
+    breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisTimeoutError("read timed out")))
+    breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisConnectionError("refused")))
     await asyncio.sleep(0.06)
     for _ in range(breaker.failure_threshold):
-        breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisTimeoutError("read timed out")))
+        breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisTimeoutError("read timed out")))
     assert breaker.is_open() is False, "the burst is instantaneous, so the duration gate must hold it closed"
 
     await asyncio.sleep(0.06)
-    breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisTimeoutError("read timed out")))
+    breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisTimeoutError("read timed out")))
     assert breaker.is_open() is True, "the same run of timeouts persisting past the duration must open it"
 
 
@@ -1023,7 +1023,7 @@ async def test_breaker_metrics_track_state_and_failure_class():
     from redis.exceptions import ConnectionError as RedisConnectionError
     from redis.exceptions import TimeoutError as RedisTimeoutError
 
-    from litellm.caching.redis_cache import RedisCircuitBreaker, _is_redis_timeout_failure
+    from litellm.caching.redis_cache import RedisCircuitBreaker, is_redis_timeout_failure
 
     def sample(name, labels=None):
         return REGISTRY.get_sample_value(name, labels) or 0.0
@@ -1035,9 +1035,9 @@ async def test_breaker_metrics_track_state_and_failure_class():
     closed_gauge_before = sample("litellm_redis_circuit_breaker_state", {"state": "closed"})
 
     breaker = RedisCircuitBreaker(failure_threshold=2, recovery_timeout=60, timeout_min_duration=5.0)
-    breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisTimeoutError("t")))
-    breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisConnectionError("refused")))
-    breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisConnectionError("refused")))
+    breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisTimeoutError("t")))
+    breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisConnectionError("refused")))
+    breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisConnectionError("refused")))
 
     assert sample("litellm_redis_circuit_breaker_failures_total", {"failure_class": "timeout"}) == timeout_before + 1
     assert sample("litellm_redis_circuit_breaker_failures_total", {"failure_class": "connectivity"}) == hard_before + 2
@@ -1205,6 +1205,117 @@ async def test_a_probe_overtaken_by_a_later_outage_leaves_the_breaker_to_the_new
     assert breaker._state == breaker.CLOSED
 
 
+def test_timeouts_during_a_blip_log_once_per_interval_not_once_per_call(sync_batch_redis_cache, caplog, monkeypatch):
+    """A timeout streak logs its first failure plus one summary per interval; other failures log per call."""
+    import logging
+
+    from redis.exceptions import TimeoutError as RedisTimeoutError
+
+    from litellm.caching import redis_cache as redis_cache_module
+    from litellm.caching.redis_cache import _RedisTimeoutLogThrottle
+
+    clock = MagicMock(return_value=1_000.0)
+    monkeypatch.setattr(
+        redis_cache_module, "_redis_timeout_log_throttle", _RedisTimeoutLogThrottle(interval=5.0, clock=clock)
+    )
+    sync_batch_redis_cache.redis_client.get.side_effect = RedisTimeoutError("Timeout reading from 127.0.0.1:6379")
+    sync_batch_redis_cache.redis_client.mget.side_effect = RedisTimeoutError("Timeout reading from 127.0.0.1:6379")
+
+    with caplog.at_level(logging.DEBUG, logger="LiteLLM"):
+        for _ in range(200):
+            assert sync_batch_redis_cache.get_cache("lit7520") is None
+        assert sync_batch_redis_cache.batch_get_cache(key_list=["lit7520"]) == {}
+
+    timeout_records = [r for r in caplog.records if "Timeout reading from" in r.getMessage()]
+    assert len(timeout_records) == 201, "every timeout must stay visible at DEBUG"
+    assert [r.getMessage() for r in timeout_records if r.levelno >= logging.WARNING] == [
+        "litellm.caching.caching: get() - Got exception from REDIS: Timeout reading from 127.0.0.1:6379"
+    ]
+    assert timeout_records[0].levelno == logging.ERROR
+    assert timeout_records[0].filename == "redis_cache.py"
+    assert timeout_records[0].lineno != timeout_records[-1].lineno, "the record must point at the cache operation"
+
+    caplog.clear()
+    clock.return_value += 5.0
+    with caplog.at_level(logging.DEBUG, logger="LiteLLM"):
+        assert sync_batch_redis_cache.batch_get_cache(key_list=["lit7520"]) == {}
+    assert [(r.levelno, r.getMessage()) for r in caplog.records] == [
+        (
+            logging.ERROR,
+            "Error occurred in batch get cache: Timeout reading from 127.0.0.1:6379"
+            " (200 more Redis timeouts since the previous Redis timeout line were logged at DEBUG)",
+        )
+    ]
+
+    caplog.clear()
+    sync_batch_redis_cache.redis_client.get.side_effect = OSError("redis unavailable")
+    with caplog.at_level(logging.DEBUG, logger="LiteLLM"):
+        for _ in range(3):
+            assert sync_batch_redis_cache.get_cache("lit7520") is None
+    assert [r.levelno for r in caplog.records if "redis unavailable" in r.getMessage()] == [logging.ERROR] * 3
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "call_method",
+    [
+        pytest.param(lambda c: c.async_set_cache_pipeline([("lit7520", "v")]), id="async_set_cache_pipeline"),
+        pytest.param(
+            lambda c: c.async_set_cache_pipeline_with_ttls([("lit7520", "v", 60.0)]),
+            id="async_set_cache_pipeline_with_ttls",
+        ),
+        pytest.param(lambda c: c.async_set_cache_sadd("lit7520", ["v"], ttl=None), id="async_set_cache_sadd"),
+        pytest.param(lambda c: c.async_increment("lit7520", 1.0), id="async_increment"),
+        pytest.param(
+            lambda c: c.async_increment_pipeline([{"key": "lit7520", "increment_value": 1.0, "ttl": 60}]),
+            id="async_increment_pipeline",
+        ),
+        pytest.param(lambda c: c.async_rpush("lit7520", ["v"]), id="async_rpush"),
+        pytest.param(
+            lambda c: c.async_rpush_pipeline([{"key": "lit7520", "values": ["v"]}]), id="async_rpush_pipeline"
+        ),
+        pytest.param(lambda c: c.async_lpop("lit7520"), id="async_lpop"),
+        pytest.param(lambda c: c.async_lpop_pipeline([{"key": "lit7520", "count": 1}]), id="async_lpop_pipeline"),
+    ],
+)
+async def test_write_path_timeouts_inside_the_interval_stay_at_debug(call_method, caplog, monkeypatch, redis_no_ping):
+    """A write or list operation timing out mid-streak is counted by the throttle instead of logging its own ERROR."""
+    import contextlib
+    import logging
+
+    from redis.exceptions import TimeoutError as RedisTimeoutError
+
+    from litellm.caching import redis_cache as redis_cache_module
+    from litellm.caching.redis_cache import _RedisTimeoutLogThrottle
+
+    clock = MagicMock(return_value=1_000.0)
+    throttle = _RedisTimeoutLogThrottle(interval=5.0, clock=clock)
+    assert throttle.admit() == 0
+    monkeypatch.setattr(redis_cache_module, "_redis_timeout_log_throttle", throttle)
+
+    timeout = RedisTimeoutError("Timeout reading from 127.0.0.1:6379")
+    client = MagicMock()
+    client.pipeline.return_value.__aenter__.side_effect = timeout
+    client.sadd = AsyncMock(side_effect=timeout)
+    client.incrbyfloat = AsyncMock(side_effect=timeout)
+    client.rpush = AsyncMock(side_effect=timeout)
+    client.lpop = AsyncMock(side_effect=timeout)
+    monkeypatch.setenv("REDIS_HOST", "https://my-test-host")
+    cache = RedisCache()
+
+    with (
+        patch.object(cache, "init_async_client", return_value=client),
+        caplog.at_level(logging.DEBUG, logger="LiteLLM"),
+    ):
+        with contextlib.suppress(RedisTimeoutError):
+            await call_method(cache)
+
+    timeout_records = [r for r in caplog.records if "Timeout reading from" in r.getMessage()]
+    assert [(r.levelno, r.filename) for r in timeout_records] == [(logging.DEBUG, "redis_cache.py")]
+    clock.return_value += 5.0
+    assert throttle.admit() == 1
+
+
 @pytest.mark.asyncio
 async def test_pool_wait_timeout_is_a_timeout_failure_not_hard_connectivity():
     """A saturated blocking pool must not open the breaker before the timeout minimum duration.
@@ -1241,7 +1352,7 @@ async def test_pool_wait_timeout_is_a_timeout_failure_not_hard_connectivity():
 def test_timeout_classification_follows_the_explicit_cause_chain_only():
     from redis.exceptions import ConnectionError as RedisConnectionError
 
-    from litellm.caching.redis_cache import _is_redis_timeout_failure
+    from litellm.caching.redis_cache import is_redis_timeout_failure
 
     def raise_chained_from_timeout() -> None:
         try:
@@ -1260,9 +1371,9 @@ def test_timeout_classification_follows_the_explicit_cause_chain_only():
     with pytest.raises(RedisConnectionError) as contextual:
         raise_while_handling_timeout()
 
-    assert _is_redis_timeout_failure(chained.value) is True
-    assert _is_redis_timeout_failure(contextual.value) is False
-    assert _is_redis_timeout_failure(RedisConnectionError("refused")) is False
+    assert is_redis_timeout_failure(chained.value) is True
+    assert is_redis_timeout_failure(contextual.value) is False
+    assert is_redis_timeout_failure(RedisConnectionError("refused")) is False
 
 
 class _RoundTripCountingRedis:
diff --git a/tests/test_litellm/proxy/proxy_server/test_spend_counters.py b/tests/test_litellm/proxy/proxy_server/test_spend_counters.py
index 6eac53df645..68462065393 100644
--- a/tests/test_litellm/proxy/proxy_server/test_spend_counters.py
+++ b/tests/test_litellm/proxy/proxy_server/test_spend_counters.py
@@ -1180,6 +1180,24 @@ async def test_apply_spend_counter_increments_open_breaker_invalidates_and_retur
     fake_cache.in_memory_cache.set_cache.assert_not_called()
 
 
+@pytest.mark.asyncio
+async def test_apply_spend_counter_increments_redis_timeout_invalidates_and_returns(monkeypatch):
+    """A Redis timeout invalidates the counters and returns without reaching the cost callback's error path."""
+    from redis.exceptions import TimeoutError as RedisTimeoutError
+
+    fake_cache = _make_spend_counter_cache()
+    fake_cache.redis_cache.async_increment_pipeline = AsyncMock(
+        side_effect=RedisTimeoutError("Timeout reading from 127.0.0.1:6379")
+    )
+    monkeypatch.setattr(ps, "spend_counter_cache", fake_cache)
+
+    await ps._apply_spend_counter_increments(_two_pending_increments())
+
+    deleted_keys = sorted(call.kwargs["key"] for call in fake_cache.in_memory_cache.delete_cache.call_args_list)
+    assert deleted_keys == ["spend:key:k", "spend:team:t"]
+    fake_cache.in_memory_cache.set_cache.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_apply_spend_counter_increments_other_redis_error_invalidates_and_raises(monkeypatch):
     fake_cache = _make_spend_counter_cache()