jpmschweitzerandClaude Opus 4.5
2a9449bc81
Remove the Redis-backed performance benchmarking in favor of the new lightweight file-based tracing system which provides better debugging capabilities for local development. - Delete src/core/benchmarks.py - Remove ENABLE_BENCHMARKS, REDIS_BENCHMARK_DB, redis_url from config - Update memory_cache comment (now uses DB 1) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
391 lines
11 KiB
Python
391 lines
11 KiB
Python
"""
|
|
Redis-backed memory cache for session context.
|
|
|
|
Provides short-term memory storage with TTL:
|
|
- Session context (24h TTL)
|
|
- Recent entities mentioned in conversation
|
|
- User-scoped with conversation isolation
|
|
|
|
Uses Redis DB 1.
|
|
"""
|
|
import json
|
|
from typing import Any
|
|
|
|
import redis.asyncio as redis
|
|
|
|
from .config import config
|
|
from .logging_config import get_logger
|
|
from .multi_tenancy import get_session_key, get_entities_key
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class MemoryCache:
|
|
"""
|
|
Redis-backed cache for session memory.
|
|
|
|
Stores ephemeral context that doesn't need vector search:
|
|
- Session context (recent topics, user state)
|
|
- Recent entities (people, places, things mentioned)
|
|
- Conversation metadata
|
|
|
|
All data expires after REDIS_MEMORY_TTL_HOURS (default 24h).
|
|
|
|
Usage:
|
|
cache = MemoryCache()
|
|
await cache.set_session_context(
|
|
user="jpmschweitzer",
|
|
conversation_id="conv_123",
|
|
context={"topic": "docker", "mood": "curious"}
|
|
)
|
|
context = await cache.get_session_context("jpmschweitzer", "conv_123")
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
redis_url: str | None = None,
|
|
ttl_hours: int | None = None,
|
|
):
|
|
"""
|
|
Initialize memory cache.
|
|
|
|
Args:
|
|
redis_url: Redis connection URL (defaults to config.redis_memory_url)
|
|
ttl_hours: TTL for cached data (defaults to config.REDIS_MEMORY_TTL_HOURS)
|
|
"""
|
|
self._redis_url = redis_url or config.redis_memory_url
|
|
self._ttl_seconds = (ttl_hours or config.REDIS_MEMORY_TTL_HOURS) * 3600
|
|
self._client: redis.Redis | None = None
|
|
|
|
logger.info(
|
|
"memory_cache_initialized",
|
|
redis_url=self._redis_url,
|
|
ttl_hours=ttl_hours or config.REDIS_MEMORY_TTL_HOURS,
|
|
)
|
|
|
|
async def _get_client(self) -> redis.Redis:
|
|
"""Get or create Redis client."""
|
|
if self._client is None:
|
|
self._client = redis.from_url(
|
|
self._redis_url,
|
|
encoding="utf-8",
|
|
decode_responses=True,
|
|
socket_timeout=config.REDIS_TIMEOUT,
|
|
socket_connect_timeout=config.REDIS_TIMEOUT,
|
|
)
|
|
return self._client
|
|
|
|
async def close(self) -> None:
|
|
"""Close Redis connection."""
|
|
if self._client is not None:
|
|
await self._client.aclose()
|
|
self._client = None
|
|
|
|
# =========================================================================
|
|
# Session Context
|
|
# =========================================================================
|
|
|
|
async def get_session_context(
|
|
self,
|
|
user: str,
|
|
conversation_id: str,
|
|
) -> dict[str, Any] | None:
|
|
"""
|
|
Get session context for a conversation.
|
|
|
|
Args:
|
|
user: User identifier
|
|
conversation_id: Conversation identifier
|
|
|
|
Returns:
|
|
Session context dict or None if not found
|
|
|
|
Example:
|
|
>>> context = await cache.get_session_context("jpmschweitzer", "conv_123")
|
|
>>> context
|
|
{"topic": "docker", "mood": "curious", "last_tool": "librarian"}
|
|
"""
|
|
try:
|
|
client = await self._get_client()
|
|
key = get_session_key(user, conversation_id)
|
|
|
|
data = await client.get(key)
|
|
if data is None:
|
|
return None
|
|
|
|
return json.loads(data)
|
|
|
|
except Exception as e:
|
|
logger.warning(
|
|
"memory_cache_get_session_failed",
|
|
user=user,
|
|
conversation_id=conversation_id,
|
|
error=str(e),
|
|
)
|
|
return None
|
|
|
|
async def set_session_context(
|
|
self,
|
|
user: str,
|
|
conversation_id: str,
|
|
context: dict[str, Any],
|
|
) -> bool:
|
|
"""
|
|
Set session context for a conversation.
|
|
|
|
Args:
|
|
user: User identifier
|
|
conversation_id: Conversation identifier
|
|
context: Context data to store
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
|
|
Example:
|
|
>>> await cache.set_session_context(
|
|
... "jpmschweitzer",
|
|
... "conv_123",
|
|
... {"topic": "docker", "mood": "curious"}
|
|
... )
|
|
True
|
|
"""
|
|
try:
|
|
client = await self._get_client()
|
|
key = get_session_key(user, conversation_id)
|
|
|
|
await client.setex(
|
|
key,
|
|
self._ttl_seconds,
|
|
json.dumps(context),
|
|
)
|
|
|
|
logger.debug(
|
|
"memory_cache_set_session",
|
|
user=user,
|
|
conversation_id=conversation_id,
|
|
context_keys=list(context.keys()),
|
|
)
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.warning(
|
|
"memory_cache_set_session_failed",
|
|
user=user,
|
|
conversation_id=conversation_id,
|
|
error=str(e),
|
|
)
|
|
return False
|
|
|
|
async def update_session_context(
|
|
self,
|
|
user: str,
|
|
conversation_id: str,
|
|
updates: dict[str, Any],
|
|
) -> bool:
|
|
"""
|
|
Update session context (merge with existing).
|
|
|
|
Args:
|
|
user: User identifier
|
|
conversation_id: Conversation identifier
|
|
updates: Fields to update/add
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
existing = await self.get_session_context(user, conversation_id) or {}
|
|
existing.update(updates)
|
|
return await self.set_session_context(user, conversation_id, existing)
|
|
|
|
async def delete_session_context(
|
|
self,
|
|
user: str,
|
|
conversation_id: str,
|
|
) -> bool:
|
|
"""
|
|
Delete session context for a conversation.
|
|
|
|
Args:
|
|
user: User identifier
|
|
conversation_id: Conversation identifier
|
|
|
|
Returns:
|
|
True if deleted, False otherwise
|
|
"""
|
|
try:
|
|
client = await self._get_client()
|
|
key = get_session_key(user, conversation_id)
|
|
await client.delete(key)
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.warning(
|
|
"memory_cache_delete_session_failed",
|
|
user=user,
|
|
conversation_id=conversation_id,
|
|
error=str(e),
|
|
)
|
|
return False
|
|
|
|
# =========================================================================
|
|
# Recent Entities
|
|
# =========================================================================
|
|
|
|
async def get_recent_entities(
|
|
self,
|
|
user: str,
|
|
conversation_id: str,
|
|
) -> list[str]:
|
|
"""
|
|
Get recently mentioned entities in a conversation.
|
|
|
|
Args:
|
|
user: User identifier
|
|
conversation_id: Conversation identifier
|
|
|
|
Returns:
|
|
List of entity names/identifiers
|
|
|
|
Example:
|
|
>>> entities = await cache.get_recent_entities("jpmschweitzer", "conv_123")
|
|
>>> entities
|
|
["Docker", "Kubernetes", "nginx"]
|
|
"""
|
|
try:
|
|
client = await self._get_client()
|
|
key = get_entities_key(user, conversation_id)
|
|
|
|
# Get all members of the set
|
|
entities = await client.smembers(key)
|
|
return list(entities)
|
|
|
|
except Exception as e:
|
|
logger.warning(
|
|
"memory_cache_get_entities_failed",
|
|
user=user,
|
|
conversation_id=conversation_id,
|
|
error=str(e),
|
|
)
|
|
return []
|
|
|
|
async def add_recent_entities(
|
|
self,
|
|
user: str,
|
|
conversation_id: str,
|
|
entities: list[str],
|
|
) -> bool:
|
|
"""
|
|
Add entities to the recent entities set.
|
|
|
|
Args:
|
|
user: User identifier
|
|
conversation_id: Conversation identifier
|
|
entities: Entity names to add
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
|
|
Example:
|
|
>>> await cache.add_recent_entities(
|
|
... "jpmschweitzer",
|
|
... "conv_123",
|
|
... ["Docker", "Kubernetes"]
|
|
... )
|
|
True
|
|
"""
|
|
if not entities:
|
|
return True
|
|
|
|
try:
|
|
client = await self._get_client()
|
|
key = get_entities_key(user, conversation_id)
|
|
|
|
# Add to set
|
|
await client.sadd(key, *entities)
|
|
|
|
# Refresh TTL
|
|
await client.expire(key, self._ttl_seconds)
|
|
|
|
logger.debug(
|
|
"memory_cache_add_entities",
|
|
user=user,
|
|
conversation_id=conversation_id,
|
|
entities=entities,
|
|
)
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.warning(
|
|
"memory_cache_add_entities_failed",
|
|
user=user,
|
|
conversation_id=conversation_id,
|
|
error=str(e),
|
|
)
|
|
return False
|
|
|
|
async def clear_recent_entities(
|
|
self,
|
|
user: str,
|
|
conversation_id: str,
|
|
) -> bool:
|
|
"""
|
|
Clear all recent entities for a conversation.
|
|
|
|
Args:
|
|
user: User identifier
|
|
conversation_id: Conversation identifier
|
|
|
|
Returns:
|
|
True if cleared, False otherwise
|
|
"""
|
|
try:
|
|
client = await self._get_client()
|
|
key = get_entities_key(user, conversation_id)
|
|
await client.delete(key)
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.warning(
|
|
"memory_cache_clear_entities_failed",
|
|
user=user,
|
|
conversation_id=conversation_id,
|
|
error=str(e),
|
|
)
|
|
return False
|
|
|
|
# =========================================================================
|
|
# Health Check
|
|
# =========================================================================
|
|
|
|
async def health_check(self) -> bool:
|
|
"""
|
|
Check if Redis is reachable.
|
|
|
|
Returns:
|
|
True if healthy, False otherwise
|
|
"""
|
|
try:
|
|
client = await self._get_client()
|
|
await client.ping()
|
|
return True
|
|
except Exception as e:
|
|
logger.error("memory_cache_health_check_failed", error=str(e))
|
|
return False
|
|
|
|
|
|
# Global cache instance (lazy initialization)
|
|
_memory_cache: MemoryCache | None = None
|
|
|
|
|
|
def get_memory_cache() -> MemoryCache:
|
|
"""
|
|
Get global memory cache instance.
|
|
|
|
Returns:
|
|
MemoryCache instance
|
|
"""
|
|
global _memory_cache
|
|
if _memory_cache is None:
|
|
_memory_cache = MemoryCache()
|
|
return _memory_cache
|