feat(context): Phase 6 - Caching Layer #84

New Issue

cardosofelipe · 2026-01-04T00:52:21Z

cardosofelipe commented

2026-01-04 00:52:21 +00:00

Overview

Implement Redis-based caching for assembled contexts and scoring results.

Parent Issue

#61: Context Management Engine

Implementation Tasks

1. Create `cache/context_cache.py`

Create ContextCache class
Implement Redis connection handling
Implement get_assembled() method
Implement set_assembled() method
Implement compute_fingerprint() for cache keys
Handle Redis unavailability gracefully

class ContextCache:
    """Redis-based caching for context operations."""

    def __init__(self, redis: Redis, settings: ContextSettings):
        self.redis = redis
        self.prefix = settings.cache_prefix
        self.ttl = settings.cache_ttl_seconds

    def _cache_key(self, *parts: str) -> str:
        return f"{self.prefix}:{':'.join(parts)}"

    async def get_assembled(
        self,
        fingerprint: str
    ) -> AssembledContext | None:
        """Get cached assembled context by fingerprint."""
        key = self._cache_key("assembled", fingerprint)
        data = await self.redis.get(key)
        if data:
            return AssembledContext.from_json(data)
        return None

    async def set_assembled(
        self,
        fingerprint: str,
        context: AssembledContext
    ) -> None:
        """Cache assembled context."""
        key = self._cache_key("assembled", fingerprint)
        await self.redis.setex(key, self.ttl, context.to_json())

    def compute_fingerprint(
        self,
        contexts: list[BaseContext],
        query: str,
        model: str
    ) -> str:
        """Compute cache key from inputs."""
        content = json.dumps({
            "contexts": [c.to_dict() for c in contexts],
            "query": query,
            "model": model
        }, sort_keys=True)
        return hashlib.sha256(content.encode()).hexdigest()[:32]

2. Implement Score Caching

Cache relevance scores per query-context pair
Use shorter TTL for scores (5 minutes)
Implement cache invalidation

3. Implement Cache Metrics

Track cache hit/miss ratio
Log cache operations for debugging
Add cache stats to health check

Files to Create

backend/app/services/context/cache/
├── __init__.py
└── context_cache.py

Acceptance Criteria

Assembled contexts are cached in Redis
Cache fingerprinting is deterministic
TTL-based expiry works correctly
Graceful degradation when Redis unavailable
Cache hit improves response time significantly
Unit tests with mocked Redis

Dependencies

#79 (Phase 1 - Foundation)
#82 (Phase 4 - Assembly Pipeline)

Labels

phase-2, context, backend

## Overview Implement Redis-based caching for assembled contexts and scoring results. ## Parent Issue - #61: Context Management Engine --- ## Implementation Tasks ### 1. Create `cache/context_cache.py` - [ ] Create `ContextCache` class - [ ] Implement Redis connection handling - [ ] Implement `get_assembled()` method - [ ] Implement `set_assembled()` method - [ ] Implement `compute_fingerprint()` for cache keys - [ ] Handle Redis unavailability gracefully ```python class ContextCache: """Redis-based caching for context operations.""" def __init__(self, redis: Redis, settings: ContextSettings): self.redis = redis self.prefix = settings.cache_prefix self.ttl = settings.cache_ttl_seconds def _cache_key(self, *parts: str) -> str: return f"{self.prefix}:{':'.join(parts)}" async def get_assembled( self, fingerprint: str ) -> AssembledContext | None: """Get cached assembled context by fingerprint.""" key = self._cache_key("assembled", fingerprint) data = await self.redis.get(key) if data: return AssembledContext.from_json(data) return None async def set_assembled( self, fingerprint: str, context: AssembledContext ) -> None: """Cache assembled context.""" key = self._cache_key("assembled", fingerprint) await self.redis.setex(key, self.ttl, context.to_json()) def compute_fingerprint( self, contexts: list[BaseContext], query: str, model: str ) -> str: """Compute cache key from inputs.""" content = json.dumps({ "contexts": [c.to_dict() for c in contexts], "query": query, "model": model }, sort_keys=True) return hashlib.sha256(content.encode()).hexdigest()[:32] ``` ### 2. Implement Score Caching - [ ] Cache relevance scores per query-context pair - [ ] Use shorter TTL for scores (5 minutes) - [ ] Implement cache invalidation ### 3. Implement Cache Metrics - [ ] Track cache hit/miss ratio - [ ] Log cache operations for debugging - [ ] Add cache stats to health check --- ## Files to Create ``` backend/app/services/context/cache/ ├── __init__.py └── context_cache.py ``` --- ## Acceptance Criteria - [ ] Assembled contexts are cached in Redis - [ ] Cache fingerprinting is deterministic - [ ] TTL-based expiry works correctly - [ ] Graceful degradation when Redis unavailable - [ ] Cache hit improves response time significantly - [ ] Unit tests with mocked Redis --- ## Dependencies - #79 (Phase 1 - Foundation) - #82 (Phase 4 - Assembly Pipeline) ## Labels `phase-2`, `context`, `backend`

cardosofelipe referenced this issue

2026-01-04 00:52:21 +00:00

feat(context): Phase 7 - Main Engine & Integration #85

cardosofelipe referenced this issue

2026-01-04 00:54:11 +00:00

feat(mcp): Context Management Engine #61

cardosofelipe referenced this issue from a commit

2026-01-04 14:23:32 +00:00

feat(context): implement Redis-based caching layer (#84)

cardosofelipe closed this issue

2026-01-04 18:53:44 +00:00

Sign in to join this conversation.