diff --git a/backend/package/yuxi/channel/infrastructure/persistence/repository/caching_binding_repository.py b/backend/package/yuxi/channel/infrastructure/persistence/repository/caching_binding_repository.py
new file mode 100644
index 00000000..06ffac69
--- /dev/null
+++ b/backend/package/yuxi/channel/infrastructure/persistence/repository/caching_binding_repository.py
@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+import json
+import logging
+
+from yuxi.channel.domain.exception.cache_error import CacheError
+from yuxi.channel.domain.model.binding.channel_binding import ChannelBinding
+from yuxi.channel.domain.port import CachePort
+from yuxi.channel.domain.repository.binding_repository import BindingRepositoryPort
+
+logger = logging.getLogger(__name__)
+
+_CACHE_TTL = 60
+_NULL_CACHE_TTL = 10
+_NULL_MARKER = json.dumps({"__null__": True})
+_CACHE_VERSION = "v2"
+
+
+class CachingBindingRepository(BindingRepositoryPort):
+    def __init__(self, delegate: BindingRepositoryPort, cache_port: CachePort):
+        self._delegate = delegate
+        self._cache = cache_port
+
+    def _cache_key(self, channel_type: str, account_id: str, group_id: str) -> str:
+        return f"channel:binding:{_CACHE_VERSION}:{channel_type}:{account_id}:{group_id}"
+
+    async def find_active_binding(self, *, channel_type: str, account_id: str, group_id: str) -> ChannelBinding | None:
+        try:
+            cache_key = self._cache_key(channel_type, account_id, group_id)
+            cached = await self._cache.get(cache_key)
+            if cached:
+                data = json.loads(cached)
+                if data.get("__null__"):
+                    return None
+                return ChannelBinding(
+                    id=data.get("id", 0),
+                    channel_type=data.get("channel_type", ""),
+                    account_id=data.get("account_id", ""),
+                    group_id=data.get("group_id", ""),
+                    agent_config_id=data.get("agent_config_id", 0),
+                    session_key_strategy=data.get("session_key_strategy", "auto"),
+                    is_enabled=data.get("is_enabled", True),
+                    created_by=data.get("created_by"),
+                    updated_by=data.get("updated_by"),
+                    created_at=data.get("created_at"),
+                    updated_at=data.get("updated_at"),
+                )
+        except CacheError:
+            logger.warning("Redis cache read failed, falling back to DB")
+        except (json.JSONDecodeError, KeyError, TypeError):
+            logger.warning("Cache data corrupted for key %s:%s:%s, falling back to DB", channel_type, account_id, group_id)
+
+        result = await self._delegate.find_active_binding(
+            channel_type=channel_type, account_id=account_id, group_id=group_id
+        )
+
+        try:
+            cache_key = self._cache_key(channel_type, account_id, group_id)
+            if result is None:
+                await self._cache.set(cache_key, _NULL_MARKER, ex=_NULL_CACHE_TTL)
+            else:
+                await self._cache.set(
+                    cache_key,
+                    json.dumps(result.__dict__, ensure_ascii=False, default=str),
+                    ex=_CACHE_TTL,
+                )
+        except CacheError:
+            logger.warning("Redis cache write failed, skipping cache")
+
+        return result
+
+    async def create_binding(
+        self,
+        *,
+        channel_type: str,
+        account_id: str,
+        group_id: str,
+        agent_config_id: int,
+        created_by: str | None = None,
+    ) -> ChannelBinding:
+        result = await self._delegate.create_binding(
+            channel_type=channel_type,
+            account_id=account_id,
+            group_id=group_id,
+            agent_config_id=agent_config_id,
+            created_by=created_by,
+        )
+        await self._invalidate(channel_type, account_id, group_id)
+        return result
+
+    async def update_binding(
+        self,
+        binding_id: int,
+        *,
+        agent_config_id: int | None = None,
+        is_enabled: bool | None = None,
+        updated_by: str | None = None,
+    ) -> ChannelBinding | None:
+        result = await self._delegate.update_binding(
+            binding_id,
+            agent_config_id=agent_config_id,
+            is_enabled=is_enabled,
+            updated_by=updated_by,
+        )
+        if result:
+            await self._invalidate(result.channel_type, result.account_id, result.group_id)
+        return result
+
+    async def delete_binding(self, binding_id: int, *, updated_by: str | None = None) -> ChannelBinding | None:
+        result = await self._delegate.delete_binding(binding_id, updated_by=updated_by)
+        if result:
+            await self._invalidate(result.channel_type, result.account_id, result.group_id)
+        return result
+
+    async def get_binding(self, binding_id: int) -> ChannelBinding | None:
+        return await self._delegate.get_binding(binding_id)
+
+    async def list_bindings(
+        self, *, channel_type: str | None = None, offset: int = 0, limit: int = 50
+    ) -> tuple[list[ChannelBinding], int]:
+        return await self._delegate.list_bindings(channel_type=channel_type, offset=offset, limit=limit)
+
+    async def _invalidate(self, channel_type: str, account_id: str, group_id: str) -> None:
+        try:
+            cache_key = self._cache_key(channel_type, account_id, group_id)
+            await self._cache.delete(cache_key)
+        except CacheError:
+            logger.warning("Redis cache invalidation failed, cache will expire via TTL")