Amazon Elasticache

ElastiCache

A modular ElastiCache toolkit organized as a registry of sub-skills. Each sub-skill handles one domain of ElastiCache work. The router below matches user intent to the right sub-skill, then loads only the references needed for that sub-skill.

How this skill works

1. Match the user's request against the semantic categories in the registry below. Match on meaning, not exact wording ("help me figure out which data structures to use" matches data-modeling even without the word "pattern"). 2. Disambiguation: If the user's intent matches multiple sub-skills, apply these rules in order:

• If .elasticache/requirements.json exists with infrastructure.endpoint set, prefer monitoring or data-modeling (the user has an existing cache).
• If no cache exists (no requirements.json or no endpoint), prefer requirements.
• If still ambiguous, ask one clarifying question: "Are you looking to set up something new, or troubleshoot something existing?"

3. Check the Guardrails section before recommending an engine or deployment model. 4. Read references/{sub-skill-id}/instructions.md for the matched sub-skill. If the file is not found at a relative path, check your prompt or environment for the skill directory absolute path and retry with {skill-directory}/references/{sub-skill-id}/instructions.md. 5. If the request spans multiple sub-skills, execute them in pipeline order. 6. If a sub-skill requires upstream context (engine, deployment model, endpoint) not yet in session memory, route to the upstream sub-skill first. 7. If no sub-skill matches, activate requirements first. 8. If a script or CLI call fails, show the error to the user and suggest a specific fix before retrying.

Sub-skill registry

Each entry has: an ID (directory name under references/), a domain description, semantic categories for matching, and upstream/downstream dependencies.

Pipeline order

Sub-skills run independently, but common multi-step journeys follow these pipelines:

• requirements → setup → (data-modeling | genai) → monitoring
• migration → setup → monitoring
• monitoring → setup | migration (if metrics indicate)

State handoff: requirements.json

.elasticache/requirements.json is the single source of truth for cross-sub-skill state. Each sub-skill reads it at start and writes its section after completing work. Read before writing; merge, do not overwrite.

Ownership note: deployment_model is set by requirements during initial interview. migration may update it after an engine or deployment model switch (e.g., node-based to serverless).

requirements.json should include "schema_version": 1 and "last_updated": "<ISO timestamp>" at the top level. Every sub-skill that writes to requirements.json must update last_updated. If last_updated is older than 7 days, warn the user that cached state may be stale.

requirements.json tracks one active cache. If the user works with multiple caches in the same project, confirm which cache is active before reading or writing state.

When a sub-skill needs upstream context (engine, endpoint, auth model), check requirements.json first. If the field is null or the file does not exist, route to the upstream sub-skill.

Global rules (apply to every sub-skill)

1. Execution path. Use AWS CLI, SDK (boto3), CloudFormation, or CDK as the primary path for control-plane work. Use valkey-py as the primary path for data-plane work.

2. Response depth. Summary (2-3 sentences) for "should I" or "which" questions. Standard (recommendation + config + code + next steps) by default. Expert (full decision matrix with alternatives, cost, security caveats) for "why" or "compare all" questions. Escalate on user request; never downgrade unprompted.

3. Session memory. Track region, VPC, engine, deployment model, auth model, compute runtime, and language. Carry forward across sub-skills. Do not re-ask. If the user overrides a value, update it everywhere. Inferred values (from workspace scan or IaC) must be re-confirmed before high-risk decisions (engine, deployment model, security posture); low-risk inferences (language, framework, region) can be used as defaults silently.

4. Source priority. Always answer from skill-local files first (sub-skill references, then scripts/). Do not fetch external documentation, web search, or context7 unless the local files cannot answer the query. When local files are insufficient, fall back to official AWS docs: https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/ for features and https://aws.amazon.com/elasticache/pricing/ for pricing. Never invent price points or version constraints. If the user references a Valkey or Redis version, feature, or pricing tier not covered in local files, fall back to https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/ before answering. Do not extrapolate from local content that may be outdated.

5. Freshness disclaimer. When outputting pricing, version constraints, or feature availability, include a one-line disclaimer: "For current pricing see https://aws.amazon.com/elasticache/pricing/. For current feature availability see https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/."

6. Connection safety. Never create a Valkey/Redis client at module level (top of file, import time). Initialize connections inside a function or on first use. Module-level connections crash applications that import the module before the cache is reachable.

7. Code environment. Use python3 (not python) in all commands and shebangs. Do not generate pip install or pytest commands in application code. Prefer the agent's file-writing tool over shell redirection for writing files.

8. Input validation (write operations only). Before running any create, modify, or delete CLI command or SDK call that uses user-provided values (cache name, region, subnet IDs, security group IDs, VPC ID, KMS key, snapshot name), validate them with scripts/input_validator.py. Run python3 scripts/input_validator.py cache_name=<val> region=<val> ... or import validate_all as a library. Skip validation for read-only operations (describe, list) where bad input is harmless.

9. Resource tagging (always apply on resource creation). When creating any cache (create-serverless-cache, create-replication-group, create-cache-cluster), ALWAYS include these tags:

--tags Key=managed_by,Value=aws-skills Key=skill,Value=elasticache Key=skill_version,Value=1.0.0 Key=created_by,Value=elasticache-skill Key=generation_model,Value={your-model-id}

Example: --tags Key=managed_by,Value=aws-skills Key=skill,Value=elasticache Key=skill_version,Value=1.0.0 Key=created_by,Value=elasticache-skill Key=generation_model,Value=claude-sonnet-4-20250514

Set generation_model to your own model identifier. Include these tags even if the user does not mention tagging. If the user provides additional tags, append these to their tags rather than replacing them.

Reference loading

Load additional references only when the current turn's answer requires them.

On-demand pointers (not preloaded; fetch when the trigger applies):

• references/shared-ux/production-readiness.md — when the user asks if their cache is ready for production, or after setup completes and the user wants to go to production
• references/shared-ux/action-safety.md — before any destructive action (risk levels, never-auto-execute list)
• references/shared-ux/error-remediation.md — when the user hits a specific ElastiCache error code (MOVED, CROSSSLOT, CLUSTERDOWN, MULTI/EXEC+IAM, etc.)
• references/shared-foundation/boundary-doc.md — when the user asks what this skill covers
• references/shared-foundation/attribution.md — when generating CLI commands, SDK code, or IaC templates
• references/shared-foundation/architecture-diagrams.md — when the user asks for architecture diagrams or visual reference
• references/shared-runtime/lambda.md — when connecting from Lambda (cold start gotchas, IAM auth code, lazy init)
• references/shared-runtime/ecs.md — when connecting from ECS (SIGTERM shutdown, connection pool drain, task definition)
• references/shared-runtime/eks.md — when connecting from EKS (IRSA, service mesh bypass, SecurityGroupPolicy CRD)
• references/shared-runtime/api-gateway.md — when integrating with API Gateway (no direct path, caching layers comparison)
• references/shared-runtime/rds-acceleration.md — when caching RDS/Aurora queries (thundering herd, stampede protection, invalidation)
• references/shared-runtime/secret-injection.md — when the user asks about credential management per compute platform
• references/shared-security/encryption-defaults.md — when adding encryption to an existing unencrypted cluster (TLS two-step migration, at-rest immutability)
• references/shared-security/config-guardrails.md — when the user wants continuous compliance monitoring (AWS Config rules, custom Lambda rules)
• references/shared-security/vpc-patterns.md — when debugging port/security-group issues (port 6380 serverless reader, anti-patterns)

Folder convention: references/ contains 10 folders. 6 match the sub-skills (requirements, setup, data-modeling, genai, monitoring, migration) and are routing destinations. The 4 shared-* folders (shared-foundation, shared-ux, shared-security, shared-runtime) are cross-cutting material loaded on demand, not routing destinations.

Guardrails

Product truths

• ElastiCache Serverless deploys in under a minute and removes infrastructure management.
• Valkey serverless pricing is 33% lower than other supported engines; node-based Valkey pricing is 20% lower.
• Serverless caches have in-transit encryption always enabled (cannot be disabled).
• IAM auth is available for all ElastiCache Valkey versions (7.2 is the baseline Valkey version on ElastiCache) and Redis OSS 7.0+.
• Valkey version ladder: 7.2 (baseline), 8.0 (20% more data per node (capacity improvement), per-slot metrics), 8.1 (Bloom filters, COMMANDLOG, SET IFEQ, 20% less memory via new hash table (efficiency improvement)), 8.2 (vector search), 9.0 (recommended default for new clusters). Recommend Valkey 9.0 for new clusters unless a specific feature dictates otherwise.
• Vector search is available for Valkey 8.2 or above on node-based clusters (recommend 9.0).
• Global Datastore is available for node-based clusters only. It does not support IPv6 or Local Zones. Global Datastore supports AUTH and RBAC. Cross-region failover must be promoted manually (no autofailover across regions). At-rest encryption must be enabled on all clusters in the Global Datastore, but each cluster can use a separate KMS key per region.
• Online migration from self-managed Redis to ElastiCache requires: (source) AUTH must not be enabled, protected-mode set to no, replication and administrative commands must not be renamed (e.g., sync, psync, info, config, command, cluster); (target) encryption in-transit disabled, Multi-AZ enabled, engine version Redis OSS 5.0.6+ or Valkey 7.2+, not part of a Global Datastore, data tiering disabled. Shard counts must match between source and target. All source Redis instances must use the same port. Online migration is not supported for serverless caches (node-based targets only). See references/migration/topology-validation.md for the full checklist.

Command and Feature Availability by Engine Version

Quick reference for which commands and features are available in each ElastiCache engine version. Use this when recommending data structures or patterns to verify they are supported on the customer's current engine.

Command Availability Matrix

Deployment Model Restrictions

Some commands are further restricted by deployment model:

Common Scenarios

"I need per-field TTL on hashes": Available on ElastiCache Valkey 9.0. Per-field hash expiration (HEXPIRE, HTTL, HSETEX, HGETEX) was introduced in Valkey 9.0. On engine versions before 9.0, the workaround is to store each field as a separate key with its own EXPIRE, or use a sorted set with timestamps for expiration tracking.

"I need vector search / semantic similarity": Requires Valkey 8.2 or above on node-based clusters (recommend 9.0). Not available on serverless. Not available in Redis OSS on ElastiCache. Use FT.CREATE to define a vector index, FT.SEARCH to query it.

"I need server-side functions": Requires Redis OSS 7.x or Valkey 7.2+. FUNCTION LOAD is available from Redis OSS 7.x (node-based only; not available on serverless). FUNCTION LOAD replaces the older EVAL-only model (though EVAL still works). Functions persist across restarts, unlike EVALSHA-cached scripts.

"I need sharded Pub/Sub": Requires Redis OSS 7.x or Valkey 7.2+. Use SSUBSCRIBE instead of SUBSCRIBE for cluster-mode-enabled deployments. Sharded Pub/Sub routes messages to the shard that owns the channel, reducing cross-node traffic.

Common Patterns (Python & Node.js)

Canonical implementations for the three most requested patterns. Each includes connection setup, core operations, error handling, and TTL strategy. For advanced algorithms (XFetch, sliding-window rate limiting, tie-breaking leaderboards), see recipe-gallery.md.

All examples use lazy client initialization per the connection safety rule.

Cluster mode vs. standalone clients: ElastiCache Serverless operates in cluster mode enabled only; clients must support cluster mode (e.g., ValkeyCluster in Python, new Redis.Cluster() in ioredis) to handle slot-based routing. The standalone Valkey/new Redis() clients shown below work for cluster mode disabled node-based clusters. For serverless or cluster mode enabled deployments, swap to the cluster-aware client variant.

---

1. Cache-Aside (DB Query Cache)

Read from cache first. On miss, read from DB, write to cache. On DB write, invalidate cache.

Python (valkey-py)

import valkey
import json

_client = None

def get_client():
    global _client
    if _client is None:
        _client = valkey.Valkey(
            host="mycluster.xxxxxx.use1.cache.amazonaws.com",  # Replace with your endpoint; for clustercfg endpoints, use ValkeyCluster instead
            port=6379, ssl=True, decode_responses=True,
            username="appuser", password="<your-password-or-iam-token>",
            socket_connect_timeout=5, socket_timeout=2,
        )
    return _client

CACHE_TTL = 300  # 5 minutes; tune per staleness tolerance

def cache_get(client, entity: str, entity_id: str, db_fetch_fn) -> dict | None:
    key = f"cache:{entity}:{entity_id}"
    cached = client.get(key)
    if cached is not None:
        return json.loads(cached)
    result = db_fetch_fn(entity_id)
    if result is not None:
        client.setex(key, CACHE_TTL, json.dumps(result))
    return result

def cache_invalidate(client, entity: str, entity_id: str) -> None:
    client.delete(f"cache:{entity}:{entity_id}")

def cache_write_through(client, entity: str, entity_id: str, data: dict, db_write_fn) -> None:
    db_write_fn(entity_id, data)
    client.setex(f"cache:{entity}:{entity_id}", CACHE_TTL, json.dumps(data))

Node.js (ioredis)

const Redis = require("ioredis");

let client;
function getClient() {
  if (!client) {
    client = new Redis({
      host: "mycluster.xxxxxx.use1.cache.amazonaws.com", port: 6379, tls: {},  // For clustercfg or serverless endpoints, use new Redis.Cluster([{host, port}], {redisOptions: {tls: {}}}) instead of standalone new Redis()
      username: "appuser", password: "<your-password-or-iam-token>",
      connectTimeout: 5000, commandTimeout: 2000,
    });
  }
  return client;
}

const CACHE_TTL = 300;

async function cacheGet(entity, entityId, dbFetchFn) {
  const key = `cache:${entity}:${entityId}`;
  const cached = await getClient().get(key);
  if (cached !== null) return JSON.parse(cached);
  const result = await dbFetchFn(entityId);
  if (result !== null && result !== undefined) {
    await getClient().setex(key, CACHE_TTL, JSON.stringify(result));
  }
  return result;
}

async function cacheInvalidate(entity, entityId) {
  await getClient().del(`cache:${entity}:${entityId}`);
}

---

2. Session Store

Store user session data as a hash. TTL matches session timeout.

Python (valkey-py)

import valkey
import json
import time

_client = None

def get_client():
    global _client
    if _client is None:
        _client = valkey.ValkeyCluster(
            host="mycluster.xxxxxx.clustercfg.use1.cache.amazonaws.com",  # Replace with your endpoint
            port=6379, ssl=True, decode_responses=True,
            username="appuser", password="<your-password-or-iam-token>",
            socket_connect_timeout=5, socket_timeout=2,
        )
    return _client

SESSION_TTL = 1800  # 30 minutes

def save_session(client, session_id: str, data: dict) -> None:
    key = f"session:{session_id}"
    data["updated_at"] = str(time.time())
    client.hset(key, mapping=data)
    client.expire(key, SESSION_TTL)

def get_session(client, session_id: str) -> dict | None:
    key = f"session:{session_id}"
    data = client.hgetall(key)
    if not data:
        return None
    client.expire(key, SESSION_TTL)  # slide expiry on access
    return data

def delete_session(client, session_id: str) -> None:
    client.delete(f"session:{session_id}")

Node.js (ioredis)

const SESSION_TTL = 1800;

async function saveSession(sessionId, data) {
  const key = `session:${sessionId}`;
  data.updated_at = String(Date.now() / 1000);
  await getClient().hset(key, data);
  await getClient().expire(key, SESSION_TTL);
}

async function getSession(sessionId) {
  const key = `session:${sessionId}`;
  const data = await getClient().hgetall(key);
  if (!data || Object.keys(data).length === 0) return null;
  await getClient().expire(key, SESSION_TTL);
  return data;
}

async function deleteSession(sessionId) {
  await getClient().del(`session:${sessionId}`);
}

---

3. Rate Limiter (Fixed Window)

Simple INCR + EXPIRE pattern. For sliding-window with better accuracy at window boundaries, see recipe-gallery.md.

Python (valkey-py)

import valkey

_client = None

def get_client():
    global _client
    if _client is None:
        _client = valkey.ValkeyCluster(
            host="mycluster.xxxxxx.clustercfg.use1.cache.amazonaws.com",  # Replace with your endpoint
            port=6379, ssl=True, decode_responses=True,
            username="appuser", password="<your-password-or-iam-token>",
            socket_connect_timeout=5, socket_timeout=2,
        )
    return _client

# Lua script: atomic INCR + conditional EXPIRE.
# Returns [count, ttl]. Sets TTL only when the key was just created (count == 1)
# or when TTL is missing (race condition recovery).
#
# NOTE: ElastiCache Serverless requires all Lua scripts to have at least one
# KEY parameter (scripts with 0 keys will fail). This script uses KEYS[1].
# In cluster mode (including Serverless), all keys used in a Lua script must
# hash to the same slot. If you extend this script to use multiple keys,
# use hash tags (e.g., {client_id}) to ensure they share a slot.
_RATE_LIMIT_SCRIPT = """
local count = redis.call('INCR', KEYS[1])
local ttl = redis.call('TTL', KEYS[1])
if count == 1 or ttl == -1 then
    redis.call('EXPIRE', KEYS[1], ARGV[1])
end
return {count, ttl}
"""

def is_rate_limited(client, client_id: str, max_requests: int,
                    window_seconds: int) -> tuple[bool, int]:
    """Returns (is_limited, remaining_requests).

    Uses a Lua script for atomicity: INCR and EXPIRE run in a single
    server-side call, eliminating the race where a key could persist
    without a TTL if EXPIRE failed or another request slipped in
    between INCR and EXPIRE.
    """
    key = f"ratelimit:{client_id}:{window_seconds}"
    count, _ = client.eval(_RATE_LIMIT_SCRIPT, 1, key, window_seconds)

    remaining = max(0, max_requests - count)
    return count > max_requests, remaining

Node.js (ioredis)

// Lua script: atomic INCR + conditional EXPIRE.
// Sets TTL when key is new (count == 1) or TTL is missing (race recovery).
const RATE_LIMIT_SCRIPT = `
local count = redis.call('INCR', KEYS[1])
local ttl = redis.call('TTL', KEYS[1])
if count == 1 or ttl == -1 then
    redis.call('EXPIRE', KEYS[1], ARGV[1])
end
return {count, ttl}
`;

async function isRateLimited(clientId, maxRequests, windowSeconds) {
  const key = `ratelimit:${clientId}:${windowSeconds}`;
  const [count] = await getClient().eval(RATE_LIMIT_SCRIPT, 1, key, windowSeconds);
  const remaining = Math.max(0, maxRequests - count);
  return { limited: count > maxRequests, remaining };
}

---

Error Handling Pattern

Cache is an optimization, not a dependency. Wrap all cache calls so failures fall through to the data source.

def safe_cache_get(client, entity, entity_id, db_fetch_fn):
    try:
        return cache_get(client, entity, entity_id, db_fetch_fn)
    except Exception:
        return db_fetch_fn(entity_id)

async function safeCacheGet(entity, entityId, dbFetchFn) {
  try {
    return await cacheGet(entity, entityId, dbFetchFn);
  } catch {
    return dbFetchFn(entityId);
  }
}

Reconnection strategy: When a client disconnects due to a timeout or failover, retry with exponential backoff and jitter to avoid a thundering herd of reconnections that can overwhelm the server and cause prolonged outages. See the AWS best practices for cluster client discovery and exponential backoff.

Data Modeling & Patterns

When to use: The user needs help designing key schemas, choosing data structures (strings, hashes, sorted sets, streams, etc.), implementing caching patterns (cache-aside, session store, leaderboard, rate limiting), or generating application code for a specific data access pattern. When not needed: The user is setting up a new cache, choosing an engine or deployment model, configuring authentication, or working on monitoring and cost optimization.

Design key schemas and implement Valkey/Redis data structure patterns for application use cases.

Loading

Read this file first. Other references in this folder load on demand when the current answer requires them. Scripts in scripts/ run on demand.

Routing

• For semantic cache, conversational memory, RAG, or vector-based recommendation, route to the genai sub-skill.
• If the user's engine is Memcached, load memcached-recipe.md for Memcached-specific patterns and constraints.
• If the user asks whether a specific command or feature is available on their engine version, load command-availability.md for the version compatibility matrix.

Reference File Loading

Load recipe-gallery.md when the user asks about ANY of these patterns (do not rely on the Pattern Quick Reference table alone; the gallery has full working code):

Load command-availability.md when the user asks about command support on a specific engine or version, or when generating code that uses commands beyond basic GET/SET/HSET.

Load memcached-recipe.md when engine is Memcached or the user mentions Memcached-specific constraints.

Check for existing context

Before starting, read .elasticache/requirements.json if it exists. Use the values (engine, runtime, deployment_model, patterns) as inputs rather than re-asking. If engine is memcached, load memcached-recipe.md. If patterns already lists the user's use case, skip pattern selection.

Workflow

1. Identify the user's use case and match to a pattern below 2. Recommend the right data structures and key design 3. Provide working application code in the user's language 4. Include TTL strategy and invalidation approach

After implementation

Update .elasticache/requirements.json: add or confirm the implemented pattern(s) in the patterns array. Read the existing file first, merge your updates, then write it back. Do not overwrite fields owned by other sub-skills.

Handoff to monitoring

After implementing a pattern, prompt the user: "Want me to set up CloudWatch dashboards and alarms for this cache?" If yes, hand off to the monitoring sub-skill. For cache-aside patterns, the key metrics to monitor are hit rate, evictions, and latency. For rate limiters, monitor CPU and connection count.

Pattern Quick Reference

Key Design Principles

• Use colons as namespace separators: {entity}:{id}:{attribute}
• Include the entity type in the key: user:u100:profile, not just u100
• Keep keys short but readable
• Use TTL on everything that has a natural expiry. On serverless, Valkey uses a volatile-lru eviction policy and auto-scales storage up to the configured CacheUsageLimits maximum. OOM errors only occur when the maximum is reached AND no keys with a TTL are eligible for eviction. Always set a TTL on serverless caches.
• Prefer atomic operations (INCR, HINCRBY, ZADD) over read-modify-write

Multi-Tenant Key Namespacing

In multi-tenant systems, prefix every key with the tenant identifier to prevent collisions:

{tenant_id}:{resource}:{id}

Example: tenant42:session:abc123, tenant42:cart:user7

Benefits of consistent tenant prefixing:

• Prevents data leaks between tenants; each tenant's keys are isolated by prefix.
• Enables per-tenant SCAN. Use SCAN 0 MATCH tenant42:* to enumerate only one tenant's keys.
• Supports per-tenant eviction or cleanup by scanning and deleting a single prefix.

Hash tags for multi-key operations: If you use MGET, pipelines, or transactions across multiple keys for the same tenant on cluster-mode-enabled caches (including serverless), wrap the tenant ID in curly braces to ensure all keys hash to the same slot: {tenant42}:session:abc123. Without hash tags, multi-key operations across different slots will fail with CROSSSLOT errors. Hot-slot risk: Hash tags concentrate all keys for a tenant onto a single shard. For large tenants with high key counts or throughput, this can create a hot slot that overloads one shard while others remain idle. Monitor per-shard CPU and memory metrics, and consider splitting very large tenants across multiple hash tags if hot-spotting occurs.

• Simplifies capacity analysis; count keys per tenant with SCAN + prefix match.

Dangerous Commands -- Never Use KEYS in Production

The `KEYS` command must never be used in production. KEYS * (or any KEYS pattern) blocks the entire Redis/Valkey server while it scans every key in the database. On caches with millions of keys, this can block the server for seconds or longer, causing timeouts and cascading failures for all connected clients. This applies to both node-based and serverless ElastiCache deployments.

Use `SCAN` with cursor-based iteration instead. SCAN performs the same work incrementally without blocking the server. Each SCAN call returns a small batch of keys and a cursor to continue from.

Safe SCAN pattern in Python:

# Using valkey-py (pip install valkey)
import valkey
r = valkey.Valkey(host="your-endpoint", port=6379, ssl=True, decode_responses=True)

# Or using redis-py (pip install redis) -- the AWS-validated Python client
# import redis
# r = redis.Redis(host="your-endpoint", port=6379, ssl=True, decode_responses=True)

def scan_keys(pattern="*", count=100):
    """Iterate over keys matching a pattern without blocking the server.

    Uses SCAN with a cursor to retrieve keys in small batches.
    The count parameter is a hint (not a hard limit) for batch size.
    """
    cursor = 0
    while True:
        cursor, keys = r.scan(cursor=cursor, match=pattern, count=count)
        for key in keys:
            yield key
        if cursor == 0:
            break

# Example: find all session keys
for key in scan_keys("session:*"):
    print(key)

If a team member or external reference suggests using KEYS, always redirect to SCAN. The same principle applies to other blocking commands like SMEMBERS on very large sets; prefer SSCAN, HSCAN, and ZSCAN for large collections.

Code Generation

When generating application code:

• Always use TLS (ssl=True / tls:{}) for ElastiCache endpoints
• Prefer connection pooling; avoid creating a new connection per request. Use a module-level singleton pattern (initialize on first use via a get_client() function) to reuse connections across requests.
• Use pipelining for multi-command operations (reduces round trips)
• Wrap cache calls in try/except; cache is an optimization, not a dependency
• Reuse connections across requests (connection pooling). Create the connection pool once at application startup and share it; creating a new connection per request is ~13x slower (2.82 ms vs 0.21 ms per AWS benchmarks)
• Include the relevant SDK: redis-py or valkey-py (Python), ioredis (Node.js), Lettuce or Jedis (Java), go-redis (Go), or Valkey Glide (multi-language)
• Never use restricted ElastiCache commands in generated code. The following commands are unavailable on ElastiCache for clusters running Redis OSS or Valkey: BGREWRITEAOF, BGSAVE, CONFIG, DEBUG, MIGRATE, REPLICAOF, SAVE, SLAVEOF, SHUTDOWN, SYNC. Some restrictions may vary by engine version; check command-availability.md for details.

Freshness disclaimer

When your response includes pricing, version constraints, or feature availability, include the freshness disclaimer per SKILL.md Global Rule #5: "For current pricing see https://aws.amazon.com/elasticache/pricing/. For current feature availability see https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/."

Memcached Scale-Out Recipe

Guidance for using Amazon ElastiCache for Memcached: when it fits, how it scales, and what it lacks.

When to Use Memcached

Memcached is the right choice when all of the following are true:

• You need simple get/set ephemeral caching (string key-value pairs)
• Your workload is read-heavy and benefits from multi-threaded read performance
• You do not need data persistence (data loss on restart is acceptable)
• You do not need replication or automatic failover
• You do not need advanced data structures (sorted sets, streams, hashes, lists)
• You do not need pub/sub, RBAC, IAM authentication, or vector search

Typical use case: Caching serialized database query results, HTML fragments, API responses, or computed objects where the source of truth is always the database and the cache is purely an acceleration layer.

Serverless Memcached vs Node-Based Memcached

Serverless Memcached

• Requires Memcached engine version 1.6.22 or higher
• Transparent Multi-AZ: data is automatically replicated across availability zones
• Supports automatic daily backups for recovery (must be explicitly enabled)
• No capacity planning or node selection required
• Automatic scaling within configured limits
• TLS always enabled: clients must support TLS connectivity to connect to serverless Memcached
• Simpler operational model: no node monitoring, no manual shard management

Node-Based Memcached

• Manual scaling: you choose instance types and number of nodes
• No replication: each node holds a unique slice of data
• Node failure means partial data loss (client detects and routes to other nodes, but data is gone)
• Lower per-unit cost at steady-state if you can accurately predict load

Recommendation: Start with serverless Memcached unless you have a specific reason to manage nodes (e.g., cost optimization at very large steady-state scale).

Auto Discovery

ElastiCache Memcached supports Auto Discovery, which lets clients automatically detect when nodes are added or removed. The client connects to the cluster's configuration endpoint (not individual node endpoints), polls for changes once per minute by default (this interval can be adjusted), and updates its node list without any application redeployment. See the ElastiCache Auto Discovery documentation for implementation details.

Consistent Hashing

Memcached clients use consistent hashing to distribute keys across nodes:

• Each node is assigned positions on a hash ring
• A key is hashed and placed on the ring; it maps to the next node clockwise
• When a node is added or removed, only a fraction of keys need to remap (roughly 1/N where N is the number of nodes), rather than all keys

This minimizes cache misses during scaling events.

Scale-Out Pattern

Adding capacity to a Memcached cluster:

1. Add nodes via the console, AWS CLI, or SDK 2. Auto Discovery detects the new nodes within the polling interval 3. Consistent hashing redistributes a fraction of the key space to new nodes 4. New nodes start cold: cache misses for remapped keys will hit the database until the cache warms up

For node-based, scaling is done via ModifyCacheCluster (CLI: aws elasticache modify-cache-cluster). For serverless Memcached, scaling is automatic within the configured CacheUsageLimits; adjust limits via modify-serverless-cache with --cache-usage-limits. See AWS docs for full parameter reference.

Node Failure Behavior

• Data on the failed node is lost. Node-based Memcached has no replication, so there is no replica to promote.
• The client removes the failed node from its hash ring; affected keys become cache misses that fall through to the database.
• Use serverless Memcached for transparent Multi-AZ redundancy, or design the application to tolerate cache misses gracefully.

When NOT to Use Memcached

Switch to Valkey (or Redis OSS) if you need any of the following:

Rule of thumb: If your use case goes beyond simple get/set/delete with TTL, use Valkey.

Memcached Constraints

• Maximum item size: 1 MB
• Maximum 60 nodes per node-based Memcached cluster (default quota; can be increased via Service Quotas; serverless scales differently)
• No persistence (node-based); daily backups only (serverless)
• No pub/sub, no Lua scripting, no server-side logic
• No RBAC or IAM authentication
• Data model limited to string key-value pairs

Recipe Gallery

Non-obvious algorithms that require specific implementation details. Standard patterns (session stores, counters, pub/sub, shopping carts, job queues, activity feeds) are omitted because the model generates those from constraints alone. These recipes contain fractional score encoding, probabilistic math, or cleanup semantics that must be implemented precisely.

# Shared setup for all recipes
# Uses valkey-py (pip install valkey). Alternatively, redis-py 4.1.2+ is the
# AWS-validated client (pip install redis) and is API-compatible; replace
# `import valkey` with `import redis` and `valkey.Valkey` with `redis.Redis`.
import valkey, time, json, math, random, uuid

_client = None

def get_client():
    """Lazy client initialization. Never create connections at module level.
    Uses ValkeyCluster because serverless is cluster-mode-enabled only.
    For cluster-mode-disabled node-based clusters, use valkey.Valkey instead.
    """
    global _client
    if _client is None:
        # Serverless requires cluster-mode client
        _client = valkey.ValkeyCluster(
            host="your-cache-endpoint.serverless.use1.cache.amazonaws.com",
            port=6379, ssl=True, decode_responses=True,
            socket_connect_timeout=5, socket_timeout=2,
        )
    return _client

---

1. Leaderboard Tie-Breaking

Sorted sets order equal scores lexicographically, which is rarely desired. To break ties by time (earliest score wins), encode an inverse timestamp into the fractional part of the score.

Data structure: Sorted set | Key: leaderboard:{game_id} Score format: points.inverse_timestamp (e.g., 1500.300000000) | Member: Player ID

def add_score_with_tiebreak(game_id: str, player_id: str, points: int) -> None:
    """Score = points + fractional inverse-timestamp.
    Example: 1500 pts at epoch 1700000000 -> 1500.300000000
    Higher fractional = earlier timestamp, so ZREVRANGE ranks earlier achievers first.
    """
    max_ts = 9999999999  # far-future ceiling
    fractional = (max_ts - int(time.time())) / (max_ts + 1)
    get_client().zadd(f"leaderboard:{game_id}", {player_id: points + fractional})

def get_top_n(game_id: str, n: int = 10) -> list[tuple[str, float]]:
    return get_client().zrevrange(f"leaderboard:{game_id}", 0, n - 1, withscores=True)

def get_player_rank(game_id: str, player_id: str) -> int | None:
    return get_client().zrevrank(f"leaderboard:{game_id}", player_id)

Why this works: The fractional part is always < 1, so it never changes integer-point ranking. Within the same points, an earlier player has a larger fractional part (max_ts - earlier > max_ts - later), so ZREVRANGE places them higher. No secondary data structure or Lua script needed.

---

2. Cache-Aside with XFetch (Probabilistic Early Refresh)

Probabilistic early refresh prevents thundering herd on popular keys. As TTL approaches zero, random callers refresh before actual expiry so the key never truly expires under load.

Data structure: String | Key: cache:{entity}:{id} TTL: Varies by staleness tolerance (e.g., 300s for product data, 60s for inventory)

CACHE_TTL = 300  # 5 minutes

def cache_aside_get(entity: str, entity_id: str, db_fetch_fn, ttl: int = CACHE_TTL) -> dict:
    """Cache-aside with XFetch. db_fetch_fn(entity_id) -> dict reads from the DB."""
    key = f"cache:{entity}:{entity_id}"
    c = get_client()
    cached = c.get(key)
    if cached is not None:
        data = json.loads(cached)
        remaining_ttl = c.ttl(key)
        if remaining_ttl > 0 and _should_early_refresh(remaining_ttl, ttl):
            result = db_fetch_fn(entity_id)
            if result is not None:
                c.setex(key, ttl, json.dumps(result))
                return result
        return data
    # Cache miss
    result = db_fetch_fn(entity_id)
    if result is not None:
        c.setex(key, ttl, json.dumps(result))
    return result

def _should_early_refresh(remaining_ttl: int, total_ttl: int) -> bool:
    """XFetch: returns True with increasing probability as TTL approaches 0.
    Only activates in the last 20% of TTL. beta controls aggressiveness."""
    beta = 1.0
    if total_ttl <= 0:
        return False
    threshold = total_ttl * 0.2
    if remaining_ttl >= threshold:
        return False
    return random.random() < math.exp(-remaining_ttl * beta / threshold)

Why probabilistic, not deterministic? A fixed threshold (e.g., "refresh at 10% TTL remaining") causes all concurrent readers to hit the DB at once. The exponential probability curve means on average exactly one caller refreshes early while the rest serve the cached value.

When NOT to use: Do not use XFetch for consistency-critical data (inventory counts, account balances) where serving a stale value during the refresh window is a correctness bug, not just a performance tradeoff.

---

3. Sliding-Window Rate Limiter

Each request is a unique sorted set member scored by timestamp. The window slides continuously, and denied requests are cleaned up to avoid polluting the count.

Data structure: Sorted set | Key: ratelimit:{client_id}:{window} Score: Epoch seconds | Member: {timestamp}:{uuid} (UUID prevents same-ms collisions)

def sliding_window_rate_limit(
    client_id: str, window_name: str, max_requests: int, window_seconds: int,
) -> tuple[bool, int]:
    """Returns (allowed, remaining)."""
    key = f"ratelimit:{client_id}:{window_name}"
    now = time.time()
    member = f"{now}:{uuid.uuid4().hex[:8]}"
    c = get_client()

    pipe = c.pipeline()
    pipe.zremrangebyscore(key, "-inf", now - window_seconds)  # prune old
    pipe.zadd(key, {member: now})                             # optimistic add
    pipe.zcard(key)                                           # count window
    pipe.expire(key, window_seconds + 1)                      # auto-cleanup
    results = pipe.execute()

    current_count = results[2]
    allowed = current_count <= max_requests
    if not allowed:
        c.zrem(key, member)  # remove denied request to keep count accurate
    return allowed, max(0, max_requests - current_count)

Cluster mode note: In cluster mode (including serverless), all keys in a pipeline must hash to the same slot. This recipe uses a single key per pipeline call, so it works as-is. If you extend pipelines to operate on multiple keys, use hash tags (e.g., {prefix}:key1, {prefix}:key2) to ensure slot co-location.

Why UUID members? Timestamp-only members silently drop concurrent requests (ZADD updates the score of an existing member identified by member name, so duplicate timestamps would collapse into one entry). Using a UUID suffix as part of the member name ensures each request is a distinct entry.

Why remove denied requests? Without cleanup, denied requests inflate the count, making the limiter progressively stricter under burst traffic.

---

4. Cache Invalidation (Write-Through + Event-Driven + TTL Hybrid)

No single invalidation method is reliable on its own: write-through misses changes from other services, event-driven delivery can lag or fail, and TTL alone allows stale reads. Combining all three bounds staleness even when individual layers fail.

• Write-through: Update cache on every DB write in the same code path.
• Event-driven: Subscribe to change events (SNS/SQS, DynamoDB Streams) to invalidate keys modified by other services.
• TTL safety net: Always set a TTL as backstop, so stale keys self-heal if both other layers miss.

CACHE_TTL = 300  # 5-minute backstop

def write_through(entity: str, entity_id: str, data: dict, db_write_fn) -> None:
    """Write to DB then cache. TTL acts as safety net."""
    db_write_fn(entity_id, data)
    get_client().setex(f"cache:{entity}:{entity_id}", CACHE_TTL, json.dumps(data))

def handle_change_event(entity: str, entity_id: str) -> None:
    """Called by SNS/SQS consumer when another service modifies the entity."""
    get_client().delete(f"cache:{entity}:{entity_id}")

Why all three layers? Write-through keeps the cache fresh for the owning service. Event-driven catches external writes. TTL guarantees bounded staleness even if events are lost or write-through fails. TTL is the only layer that requires zero operational trust.

Semantic Agent Memory Implementation Guide

User preferences, learned facts, and cross-session recall via vector similarity search. Requires node-based Valkey 8.2 or above (recommend 9.0).

For conversation persistence (session state, message history, resumable sessions), see session-store.md instead.

---

1. Semantic Memory with mem0 + Valkey

mem0 provides a memory layer that extracts facts from conversations, deduplicates them, and retrieves by semantic similarity. It has a first-class Valkey connector (provider: "valkey") that uses native FT.CREATE/FT.SEARCH.

Configuration

from mem0 import Memory
from utils.embeddings import VECTOR_DIM  # Your project's embedding dimension constant (e.g., 1024 for Titan V2)

config = {
    "vector_store": {
        "provider": "valkey",
        "config": {
            "index_name": "app_memory",
            "embedding_model_dims": VECTOR_DIM,
            "valkey_url": "valkeys://your-endpoint.cache.amazonaws.com:6379",
            "index_type": "hnsw",
            "hnsw_m": 16,
            "hnsw_ef_construction": 200,
            "hnsw_ef_runtime": 10,
        }
    },
    "embedder": {
        # Provider-specific. See embedding-providers.md for mem0 embedder
        # configs for Bedrock Titan, Cohere, fastembed, sentence-transformers.
    },
}

memory = Memory.from_config(config)

For HNSW parameter tuning guidance, see elasticache-search.md.

Short-Term vs Long-Term Memory

Use memory_type in metadata to distinguish:

from datetime import datetime, timedelta

def add_short_term(messages, session_id, user_id=None):
    metadata = {"memory_type": "short_term",
                "expires_at": (datetime.now() + timedelta(hours=720)).timestamp()}
    if user_id:
        memory.add(messages, user_id=user_id, agent_id="myapp",
                   run_id=session_id, metadata=metadata)
    else:
        memory.add(messages, user_id=session_id, agent_id="myapp",
                   run_id="global", metadata=metadata)

def add_long_term(messages, user_id):
    metadata = {"memory_type": "long_term",
                "expires_at": (datetime.now() + timedelta(days=90)).timestamp()}
    memory.add(messages, user_id=user_id, agent_id="myapp",
               run_id="global", metadata=metadata)

Note: The expires_at field is stored as metadata only. It does not automatically expire keys in Valkey. Your application must handle key expiration separately (e.g., by calling EXPIRE on the underlying keys, or by running a periodic cleanup job that filters on expires_at and deletes expired entries).

Add and Search

messages = [
    {"role": "user", "content": "I prefer size 12 wide width shoes"},
    {"role": "assistant", "content": "Noted, I'll filter for size 12 wide."},
]
memory.add(messages, user_id="user_001", agent_id="myapp", run_id="global")

results = memory.search(
    query="What shoe size does this customer wear?",
    user_id="user_001",
    limit=5,
)
# mem0's search() returns a similarity score (higher = more similar),
# which is the inverse of raw FT.SEARCH COSINE distance (lower = more similar).
for entry in results["results"]:
    if entry.get("score", 0) >= 0.7:
        print(f"Memory: {entry['memory']}  (score: {entry['score']:.2f})")

Important: Always pass a truthy run_id (default to "global").

---

2. Semantic Memory Without mem0

For teams that don't want the mem0 dependency, use the Valkey search commands directly. The building blocks (FT.CREATE schema, binary encoding, FT.SEARCH with TAG filters, result parsing) are all in elasticache-search.md.

Key differences from the generic patterns there:

• Prefix: memory:{uuid} to scope the index
• Schema fields: user_id TAG (isolation), memory_type TAG (short/long), created_at NUMERIC, memory TAG.

Note: ElastiCache supports three field types for FT.CREATE: TAG, NUMERIC, and VECTOR. TAG fields support exact-match filtering. The TEXT field type (which supports full-text search) is only available on MemoryDB, not ElastiCache.

• Query pattern: always pre-filter by user_id before KNN: (@user_id:{user123})=>[KNN 5 @embedding $vec AS score]
• TTL: 90 days for long-term, 30 days for short-term, applied via EXPIRE after HSET

Use generate_embedding() and embedding_to_bytes() from the shared embedding utility. See embedding-providers.md.

---

3. Key Design Principles

Namespace by user_id. Every memory query should filter by user_id to isolate data across users. In mem0, this is automatic (pass user_id= to every call). In raw Valkey, use a TAG filter: @user_id:{user123}.

Scope by agent_id. In multi-agent systems, use agent_id to separate memories per agent. One agent's learned facts should not bleed into another's context.

Scope by run_id within a session. Use run_id to associate memories with a specific session or run. This lets you query "what did the agent learn in this session?" separately from "what does the agent know about this user overall?"

Search threshold: 0.7. A cosine similarity of 0.7 is a good default for memory relevance. Below that, results tend to be tangentially related rather than genuinely useful. Tune based on your embedding model and use case.

---

Cross-References

• Conversation persistence (session state, no vector search): see session-store.md
• Embedding model selection and configuration: see embedding-providers.md
• ElastiCache platform constraints, FT.SEARCH encoding, HNSW tuning: see elasticache-search.md
• LangChain/LlamaIndex framework integration: see framework-guide.md

ElastiCache Vector Search Constraints

This file contains all ElastiCache-specific constraints for vector search. Loaded on-demand when any genai pattern needs vector search (Mode 2 server-side or Mode 3).

---

Platform Gate

This is the single most common mistake. Serverless ElastiCache cannot run vector search.

Version Detection (MUST run before any FT.* command)

Before emitting any FT.CREATE, FT.SEARCH, or other FT.* command, detect the server version and confirm vector search support. Vanilla Valkey 8.0.x does NOT have FT.CREATE. Only Valkey 8.2 or above on node-based ElastiCache supports it.

from valkey.exceptions import ResponseError

def get_valkey_version(client) -> tuple[int, int]:
    """Return (major, minor) version of the connected Valkey/Redis server."""
    info = client.info("server")
    ver = info.get("valkey_version", info.get("redis_version", "0.0"))
    major, minor = int(ver.split(".")[0]), int(ver.split(".")[1])
    return major, minor

def supports_ft_search(client) -> bool:
    """True if the server supports FT.* vector search commands.

    Uses FT._LIST probing instead of version comparison because
    ElastiCache v8.2 reports valkey_version as 8.1.x, making
    version-based detection unreliable.
    """
    try:
        client.execute_command("FT._LIST")
        return True
    except ResponseError:
        return False

Usage pattern: Call supports_ft_search(client) once at startup. If it returns False, use the Python-side fallback below instead of FT.* commands. Never assume FT.CREATE is available without checking.

Note: Because ElastiCache v8.2 reports valkey_version as 8.1.x, the get_valkey_version helper above should NOT be used for vector search detection. The supports_ft_search function probes with FT._LIST which is reliable regardless of reported version.

---

Python-Side Vector Search Fallback

When the server does not support FT.* commands (Valkey < 8.2, serverless, or any environment without search), use application-side cosine similarity. This approach stores vectors as binary HASH fields and performs brute-force KNN in Python. Suitable for datasets under ~50K vectors.

import struct
import math

def python_cosine_similarity(a: list[float], b: list[float]) -> float:
    """Cosine similarity between two vectors using only stdlib."""
    dot = sum(x * y for x, y in zip(a, b))
    norm_a = math.sqrt(sum(x * x for x in a))
    norm_b = math.sqrt(sum(x * x for x in b))
    if norm_a == 0 or norm_b == 0:
        return 0.0
    return dot / (norm_a * norm_b)

def python_knn_search(client, prefix: str, query_vec: list[float],
                      vector_field: str = "embedding", k: int = 5,
                      filter_field: str = None, filter_value: str = None) -> list[dict]:
    """Brute-force KNN search using Python-side cosine similarity.

    Scans all keys matching prefix, decodes stored FLOAT32 vectors,
    computes cosine similarity, returns top-k results sorted by similarity.
    """
    cursor = 0
    results = []
    dim = len(query_vec)

    while True:
        cursor, keys = client.scan(cursor=cursor, match=f"{prefix}*", count=200)
        for key in keys:
            data = client.hgetall(key)
            if filter_field and filter_value:
                field_key = filter_field.encode() if isinstance(filter_field, str) else filter_field
                stored = data.get(field_key, b"").decode()
                if stored != filter_value:
                    continue
            vec_bytes = data.get(vector_field.encode() if isinstance(vector_field, str) else vector_field)
            if not vec_bytes or len(vec_bytes) != dim * 4:
                continue
            stored_vec = list(struct.unpack(f"{dim}f", vec_bytes))
            score = python_cosine_similarity(query_vec, stored_vec)
            key_str = key.decode() if isinstance(key, bytes) else key
            results.append({"key": key_str, "similarity": score, "data": data})

        if cursor == 0:
            break

    results.sort(key=lambda x: -x["similarity"])
    return results[:k]

Integration pattern: Use supports_ft_search to pick the right path at startup:

client = get_client()
if supports_ft_search(client):
    ensure_index(client)
    hits = search_similar(client, query_embedding, top_k=5)
else:
    hits = python_knn_search(client, PREFIX, query_embedding, k=5)

---

Command Boundary

⚠️ FT.AGGREGATE is NOT available on ElastiCache. FT.AGGREGATE is supported on Amazon MemoryDB but is not supported on ElastiCache. If you need server-side aggregation over vector search results, use MemoryDB or perform aggregation client-side after FT.SEARCH.

---

Hard Limits

Transaction restriction: FT.CREATE, FT.DROPINDEX, and alias commands CANNOT run inside MULTI/EXEC, Lua scripts, or functions.

---

Index Lifecycle / Backfill

FT.CREATE triggers a background backfill for all existing keys matching the PREFIX. Query operations attempted while an index is undergoing backfill are not allowed and are terminated with an error.

Backfill types: During initial index creation (FT.CREATE), queries against the index are blocked and return an error until backfill completes. However, during scaling events (e.g., adding shards), the index may undergo backfill with reduced recall for search queries — queries are allowed but may return incomplete results.

Check readiness with FT.INFO. Key fields to monitor:

Wait for state=ready before issuing FT.SEARCH.

---

Hash Slot Constraint (Cluster Mode)

All keys queried by a single FT.SEARCH must reside in the same hash slot. Use hash tags to guarantee slot co-location:

doc:{myprefix}:chunk_001
doc:{myprefix}:chunk_002

The {myprefix} portion determines the slot. Without hash tags on multi-shard clusters, FT.SEARCH only returns results from the shard it executes on.

---

Client Connection for Vector Operations

When using valkey-py or redis-py for vector operations, set decode_responses=False. Vector data is binary (FLOAT32 bytes) and must not be decoded as UTF-8.

import valkey

client = valkey.Valkey(
    host="endpoint", port=6379,
    ssl=True, ssl_cert_reqs="required",  # validate the server cert (production default)
    # ssl_ca_certs="/path/to/ca-bundle.pem",  # only if your OS lacks a system CA store
    # Dev/tunnel ONLY (e.g. SSH tunnel to localhost, where the cert name won't match):
    # ssl_cert_reqs="none",  # INSECURE: disables cert validation; never in production
    decode_responses=False,  # CRITICAL for vector operations
)

If you need decode_responses=True for non-vector operations (session store, counters), use a separate client instance.

Do NOT use the high-level valkey.commands.search or redis.commands.search Python wrappers for FT.* commands. Their parameter signatures vary across library versions and produce hard-to-debug failures. Use client.execute_command() for all FT.CREATE, FT.SEARCH, FT.INFO, and FT.DROPINDEX calls.

Do NOT create the Valkey/Redis client at module level (top of file, import time). Always initialize connections inside a function or on first use. Module-level connections crash applications that import the module before the cache is reachable.

Do NOT call FT.CREATE at module level or import time. Create indexes lazily in an explicit setup function, guarded by an "already exists" check.

---

Vector Binary Encoding

HASH vectors must be stored as binary little-endian IEEE 754 FLOAT32.

Python encoding:

import struct
binary = struct.pack(f"{len(vec)}f", *vec)

• JSON vectors are stored as arrays (no binary encoding needed).
• Query vectors passed via PARAMS are also FLOAT32 bytes.

Do NOT use numpy for vector byte packing. Use only struct.pack from the Python standard library. numpy may not be available in all deployment environments.

---

Distance to Similarity Conversion

FT.SEARCH returns cosine DISTANCE, not similarity.

Conversion: similarity = 1.0 - (distance / 2.0) # Normalized to [0,1]; standard cosine similarity = 1 - distance

Cosine distance ranges from 0 (identical) to 2 (opposite). This formula normalizes to [0, 1].

Many models and tutorials use similarity (0=opposite, 1=identical). Mixing these up inverts threshold logic.

---

FT.CREATE Schema Reference

FT.CREATE idx ON HASH PREFIX 1 prefix:
  SCHEMA
  embedding VECTOR HNSW 6 TYPE FLOAT32 DIM <VECTOR_DIM> DISTANCE_METRIC COSINE
  request_id TAG
  state_tags TAG SEPARATOR ,
  slot_budget_usd NUMERIC

Key details:

• 6 after HNSW means 3 key-value pairs follow (TYPE, DIM, DISTANCE_METRIC). The number is the total count of arguments, not the number of pairs.
• PREFIX 1 prefix: scopes which hashes get indexed. The 1 is the number of prefixes.
• TAG SEPARATOR , is optional. The default separator is ,. Only specify SEPARATOR if you need a different delimiter character.

---

FT.SEARCH Query Reference

Hybrid query with pre-filters:

FT.SEARCH idx "(@tag:{value})=>[KNN 1 @embedding $vec AS score]"
  PARAMS 2 vec <binary_vector>
  RETURN 2 request_id score
  DIALECT 2

• DIALECT 2 is required for KNN queries.
• PARAMS 2 means 1 key-value pair follows (param name, param value). The number is the total count of arguments.
• Pre-filter goes before =>. Only documents matching the filter enter the KNN stage.

Parsing FT.SEARCH results (valkey-py / redis-py with decode_responses=False):

# result = [total_count, key_name_bytes, [field_bytes, value_bytes, ...], ...]
total = int(result[0])
if total == 0:
    return None

key_name = result[1]  # bytes
fields = result[2]    # flat list: [b"field1", b"value1", b"field2", b"value2", ...]
doc = {}
for i in range(0, len(fields), 2):
    k = fields[i].decode() if isinstance(fields[i], bytes) else fields[i]
    v = fields[i+1].decode() if isinstance(fields[i+1], bytes) else fields[i+1]
    doc[k] = v

distance = float(doc["score"])
similarity = 1.0 - (distance / 2.0)  # Normalized to [0,1]; standard cosine similarity = 1 - distance

The result format is a flat list, not a dict. Field names and values alternate. With decode_responses=False, all values are bytes and must be decoded manually.

---

FLAT vs HNSW

Use FLAT for small datasets (<10K vectors). Exact brute-force search, no graph overhead, zero tuning. Switch to HNSW when dataset grows and query latency matters.

---

HNSW Tuning Defaults

---

TAG Field Escaping

Hyphens in TAG values must be escaped in queries:

escaped = value.replace("-", "\\-")

Spaces should be replaced with underscores before storing.

---

Complete Vector Search Recipe

Self-contained working example. Copy and adapt. Uses only stdlib + valkey-py, lazy connection, execute_command for all FT.* calls.

import struct
import valkey

VECTOR_DIM = 384  # match your embedding model's output dimensions
INDEX_NAME = "idx:items"
PREFIX = "item:"

def get_client():
    """Lazy connection. Never call at module level."""
    return valkey.Valkey(
        host="your-endpoint", port=6379,
        ssl=True, ssl_cert_reqs="required",  # validate the server cert (production default)
        # ssl_ca_certs="/path/to/ca-bundle.pem",  # only if your OS lacks a system CA store
        # Dev/tunnel ONLY (e.g. SSH tunnel to localhost, where the cert name won't match):
        # ssl_cert_reqs="none",  # INSECURE: disables cert validation; never in production
        decode_responses=False,
    )

def ensure_index(client):
    """Create index if it doesn't exist. Call once at app startup, not at import time."""
    try:
        client.execute_command(
            "FT.CREATE", INDEX_NAME,
            "ON", "HASH",
            "PREFIX", "1", PREFIX,
            "SCHEMA",
            "embedding", "VECTOR", "HNSW", "6",
                "TYPE", "FLOAT32", "DIM", str(VECTOR_DIM), "DISTANCE_METRIC", "COSINE",
            "category", "TAG",
        )
    except Exception as e:
        if "already exists" not in str(e).lower():
            raise

def store_vector(client, item_id, embedding, category=""):
    """Store a vector. embedding is a list of floats."""
    key = f"{PREFIX}{item_id}"
    client.hset(key, mapping={
        "embedding": struct.pack(f"{VECTOR_DIM}f", *embedding),
        "category": category,
    })

def search_similar(client, query_embedding, top_k=5):
    """KNN search. Returns list of {key, similarity} dicts."""
    query_bytes = struct.pack(f"{VECTOR_DIM}f", *query_embedding)

    result = client.execute_command(
        "FT.SEARCH", INDEX_NAME,
        f"*=>[KNN {top_k} @embedding $vec AS score]",
        "PARAMS", "2", "vec", query_bytes,
        "RETURN", "1", "score",
        "LIMIT", "0", str(top_k),
        "DIALECT", "2",
    )

    if int(result[0]) == 0:
        return []

    hits = []
    for i in range(1, len(result), 2):
        key = result[i].decode() if isinstance(result[i], bytes) else result[i]
        fields = result[i + 1]
        field_dict = {}
        for j in range(0, len(fields), 2):
            k = fields[j].decode() if isinstance(fields[j], bytes) else fields[j]
            v = fields[j+1].decode() if isinstance(fields[j+1], bytes) else fields[j+1]
            field_dict[k] = v
        distance = float(field_dict["score"])
        similarity = 1.0 - (distance / 2.0)  # Normalized to [0,1]; standard cosine similarity = 1 - distance
        hits.append({"key": key, "similarity": similarity})

    return hits

---

Error Recovery

Common failures and how to fix them.

App crashes on import / startup: Move all valkey.Valkey() calls and FT.CREATE calls out of module scope. Wrap in a get_client() function called inside route handlers or on first use.

`ImportError: No module named 'numpy'`: Replace numpy.array(...).tobytes() with struct.pack(f"{len(vec)}f", *vec). Never depend on numpy for vector byte packing.

`FT.CREATE` returns an error: Verify the command matches the exact syntax: FT.CREATE <name> ON HASH PREFIX 1 <prefix> SCHEMA <fields>. The most common mistakes: missing SCHEMA keyword, wrong argument count after HNSW (must be 6 for 3 key-value pairs: TYPE, DIM, DISTANCE_METRIC), or using INDEX instead of CREATE.

`FT.SEARCH` returns 0 results when data exists:

1. Check index state: FT.INFO <index_name>, look for state: ready. If backfill is in progress, wait. 2. Check DIALECT 2 is present in the query. KNN queries require it. 3. Check the query vector is FLOAT32 bytes (struct.pack), not a string or list. 4. Check the key prefix matches what FT.CREATE was given.

`ResponseError` mentioning `valkey.commands.search`: Do not use the high-level Python search wrapper. Replace with client.execute_command("FT.SEARCH", ...) as shown in the recipe above.

Score/similarity values seem inverted: COSINE distance is 0 (identical) to 2 (opposite). Convert with similarity = 1.0 - (distance / 2.0). If your thresholds aren't working, verify you're comparing similarity (not distance) against the threshold.

Embedding Provider Chain for GenAI Sub-Skill

Load this reference when the user needs to choose or configure an embedding provider for vector search with ElastiCache (Valkey).

Important: Vector search (FT.CREATE / FT.SEARCH) requires node-based ElastiCache Valkey 8.2 or later (recommend 9.0). It is not available on ElastiCache Serverless. If using serverless, see the application-side comparison approach in semantic-cache.md.

Provider Chain (Decision Order)

First, check `.elasticache/requirements.json`. If infrastructure.embedding_provider, infrastructure.embedding_model, and infrastructure.embedding_dim are already set, use those values. Do not re-ask.

If not set, ask the user: "Do you have a preferred embedding model or provider?"

• User names a specific model (OpenAI, Cohere, Titan, etc.) -> Use that model. Match the DIM in FT.CREATE to its output dimensions.
• User says Bedrock / AWS -> Use Bedrock Titan Embed v2 (Option 1 below)
• User has no preference, or no API access -> Default to open-source fastembed (Option 3 below). Zero setup, zero cost, works anywhere.

After selection:

A. Persist the choice to requirements.json

{
  "infrastructure": {
    "embedding_provider": "bedrock",
    "embedding_model": "amazon.titan-embed-text-v2:0",
    "embedding_dim": 1024,
    "embedding_module": "utils/embeddings.py"
  }
}

B. Generate a reusable embedding utility in the user's project

Create a file (default: utils/embeddings.py, or wherever fits the user's project structure) that exports:

• generate_embedding(text: str) -> list[float]
• embedding_to_bytes(embedding: list[float]) -> bytes
• VECTOR_DIM: int

Use the provider-specific code from the Standard Functions section below. This file is generated ONCE. Every subsequent file the model generates imports from it:

from utils.embeddings import generate_embedding, embedding_to_bytes, VECTOR_DIM

Save the file path in requirements.json as infrastructure.embedding_module so the model never regenerates it.

On Return Visits

If infrastructure.embedding_module is set in requirements.json, read that file to confirm it exists. If it exists, import from it. Never regenerate. If the file was deleted, regenerate it from the stored provider/model/dim values.

---

Provider Options

Option 1: Amazon Bedrock Titan Embed Text v2 (Recommended for production)

• Model ID: amazon.titan-embed-text-v2:0
• Dimensions: 256, 512, or 1024 (recommend 1024 for best accuracy; 256 for cost-sensitive workloads)
• IAM permissions needed: bedrock:InvokeModel on the Titan Embed model ARN

Option 2: Amazon Bedrock Cohere Embed

• Model ID: cohere.embed-english-v3 or cohere.embed-multilingual-v3
• Dimensions: 1024
• IAM permissions needed: bedrock:InvokeModel on the Cohere Embed model ARN
• Note: Cohere uses texts (list) instead of inputText (string), and requires input_type. Use "search_document" when storing and "search_query" when querying.

Option 3: fastembed (Open-Source, No API Key Needed)

• Install: pip install fastembed
• Default model: BAAI/bge-small-en-v1.5 (384 dims) or BAAI/bge-base-en-v1.5 (768 dims)
• Tradeoff: Lower accuracy than Titan, but zero cost and zero setup
• Good for: Prototyping, development, CI/CD tests, users without AWS accounts

Option 4: Amazon Bedrock Cohere Embed v4 (Multimodal)

• Model ID: cohere.embed-v4:0
• Dimensions: 256, 512, 1024, or 1536 (default 1536)
• Context: Up to 128k tokens
• Multimodal: Supports interleaved text + image inputs via inputs field
• IAM permissions needed: bedrock:InvokeModel on the Cohere Embed v4 model ARN
• Note: Uses a different request format from v3. See AWS Bedrock docs for the inputs field schema.

Option 5: sentence-transformers (Open-Source, More Model Choices)

• Install: pip install sentence-transformers
• Popular model: all-MiniLM-L6-v2 (384 dims)
• Tradeoff: Wider model selection, but heavier dependency (pulls in PyTorch)

---

mem0 Embedder Configs

Use these in the embedder block of mem0's config dict. See agent-memory.md for the full mem0 configuration.

Bedrock Titan:

{
  "embedder": {
    "provider": "aws_bedrock",
    "config": {
      "model": "amazon.titan-embed-text-v2:0",
      "aws_region": "us-east-1"
    }
  }
}

Bedrock Cohere:

{
  "embedder": {
    "provider": "aws_bedrock",
    "config": {
      "model": "cohere.embed-english-v3",
      "aws_region": "us-east-1"
    }
  }
}

fastembed / sentence-transformers:

{
  "embedder": {
    "provider": "huggingface",
    "config": {
      "model": "BAAI/bge-small-en-v1.5"
    }
  }
}

---

Standard Functions (Utility File Templates)

These are the canonical implementations for the reusable embedding utility file. Use the one matching the user's chosen provider.

Bedrock Titan:

import boto3, json, struct

_bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")
VECTOR_DIM = 1024

def generate_embedding(text: str) -> list[float]:
    response = _bedrock.invoke_model(
        modelId="amazon.titan-embed-text-v2:0",
        # "embeddingTypes" is optional; float is the default. Include only if you need
        # a specific type (e.g., "binary"). Omitting it returns float embeddings.
        body=json.dumps({"inputText": text, "dimensions": VECTOR_DIM}),
    )
    return json.loads(response["body"].read())["embedding"]

def embedding_to_bytes(embedding: list[float]) -> bytes:
    return struct.pack(f"{VECTOR_DIM}f", *embedding)

Bedrock Cohere:

import boto3, json, struct

_bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")
VECTOR_DIM = 1024

def generate_embedding(text: str, query: bool = False) -> list[float]:
    response = _bedrock.invoke_model(
        modelId="cohere.embed-english-v3",
        body=json.dumps({
            "texts": [text],
            "input_type": "search_query" if query else "search_document",
            "truncate": "END",
        }),
    )
    return json.loads(response["body"].read())["embeddings"][0]

def embedding_to_bytes(embedding: list[float]) -> bytes:
    return struct.pack(f"{VECTOR_DIM}f", *embedding)

Cohere note: Pass query=True when embedding a search query (retrieval), query=False (default) when embedding documents for storage. Other providers ignore this parameter.

fastembed:

import struct
from fastembed import TextEmbedding

_model = TextEmbedding("BAAI/bge-small-en-v1.5")
VECTOR_DIM = 384

def generate_embedding(text: str) -> list[float]:
    return list(_model.embed([text]))[0].tolist()

def embedding_to_bytes(embedding: list[float]) -> bytes:
    return struct.pack(f"{VECTOR_DIM}f", *embedding)

sentence-transformers:

import struct
from sentence_transformers import SentenceTransformer

_model = SentenceTransformer("all-MiniLM-L6-v2")
VECTOR_DIM = 384

def generate_embedding(text: str) -> list[float]:
    return _model.encode(text).tolist()

def embedding_to_bytes(embedding: list[float]) -> bytes:
    return struct.pack(f"{VECTOR_DIM}f", *embedding)

---

Batch Embedding for Bulk Ingestion

For bulk ingestion (>1K documents), batch embedding calls to avoid per-request overhead. Titan accepts sequential calls (add exponential backoff for throttling). Cohere natively supports batching via texts: [list] with up to 96 texts per call.

---

Dimension Compatibility Warning

The FT.CREATE index DIM must match your embedding model's output dimension exactly. If you change embedding providers, you must:

1. Drop the existing index: FT.DROPINDEX <index_name> 2. Delete all existing vector keys: SCAN + DEL by prefix 3. Recreate the index with the new DIM 4. Re-embed and re-ingest all data

This is destructive. Choose your embedding model before ingesting production data.

Backfill warning: After recreating an index with FT.CREATE, queries (FT.SEARCH) are not allowed while the index is backfilling and will return an error. Use FT.INFO <index_name> and check the state field -- wait until it reports ready before issuing queries.

Framework Integration Guide

How to connect popular AI/ML frameworks to ElastiCache Valkey. This file covers framework-specific wiring only. For full implementation patterns, see the dedicated guides linked in each section.

---

1. Strands Agents

Package: strands-valkey-session-manager (community package, v0.1.0+ — MIT license, maintained by jeromevdl)

Import: from strands_valkey_session_manager import ValkeySessionManager

Implements Strands' SessionManager interface. Persists conversation messages, agent state, and session metadata to Valkey automatically. Serverless OK.

For full setup code and key design patterns, see session-store.md.

Strands does not include a built-in semantic cache. Wrap the agent call with cache check/store logic using the approach in semantic-cache.md.

---

2. mem0

Package: mem0

Native Valkey vector store provider (provider: "valkey"). Handles index creation, embedding storage, and similarity search internally. Requires node-based Valkey 8.2 or later (recommend 9.0).

Key wiring points:

• Use valkeys:// URL scheme (the s enables TLS). Port 6379 for node-based.
• The llm block is required for mem0's fact extraction. Use Bedrock:

  "llm": {
      "provider": "aws_bedrock",
      "config": {
          "model": "us.anthropic.claude-sonnet-4-6-v1:0",
          "max_tokens": 512,
      }
  }

• Always pass a user_id to memory.add() and memory.search() for user-scoped memory isolation.
• Key config fields: embedding_model_dims (e.g., 1024 for Titan V2) and index_type (flat or hnsw).

For full mem0 config, HNSW parameters, short/long-term memory patterns, and identity model, see agent-memory.md. For mem0 embedder configs per provider, see embedding-providers.md.

---

3. LangChain / LangGraph

Package: langgraph-checkpoint-aws (install with pip install 'langgraph-checkpoint-aws[valkey]')

Checkpointing (ValkeySaver)

Persist LangGraph agent state across invocations.

from langgraph_checkpoint_aws import ValkeySaver

with ValkeySaver.from_conn_string(
    "valkeys://your-cluster.serverless.use1.cache.amazonaws.com:6379",
    ttl_seconds=3600,
) as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "session-1"}}
    result = graph.invoke({"messages": [HumanMessage(content="Hello")]}, config)

LLM Caching (ValkeyCache)

Exact-match caching of LLM responses (no vector search needed, works on serverless).

from langgraph_checkpoint_aws import ValkeyCache

cache = ValkeyCache.from_conn_string(
    "valkeys://your-cluster.serverless.use1.cache.amazonaws.com:6379",
    prefix="llm_cache:",
    ttl=3600,
)

Use valkeys:// URL scheme for TLS.

Semantic Caching (ValkeyStore)

Vector-based semantic caching of LLM responses (requires node-based Valkey 8.2 or later; recommend 9.0).

from langgraph_checkpoint_aws import ValkeyStore, ValkeyIndexConfig

index_config = ValkeyIndexConfig(
    collection_name="semantic_cache",
    embed=embeddings,
    fields=["query"],
    index_type="HNSW",
    dims=1024,
)

store = ValkeyStore.from_conn_string(
    "valkeys://your-cluster.cache.amazonaws.com:6379",
    index=index_config,
)
store.setup()

Unlike ValkeyCache (exact-match), ValkeyStore uses vector search to match semantically similar queries. For full implementation, see semantic-cache.md.

---

4. ElastiCache TLS Connection Reference

All frameworks must use TLS when connecting to ElastiCache.

Port Reference

Security group note: For serverless caches, your VPC security group must allow inbound TCP on both port 6379 (primary) and port 6380 (reader). If you only open 6379, reader-endpoint connections will fail silently.

For raw valkey-py and valkey-glide connection examples, see elasticache-search.md. For IAM authentication setup, see the setup sub-skill (references/setup/auth-model-selector.md).

GenAI, Search & Vector Workloads

When to use: The user wants to implement semantic caching, conversational memory, RAG, vector search, recommendation engines, or any GenAI/LLM pattern with ElastiCache. Also useful when the user asks about caching LLM responses (exact or semantic match). When not needed: The user is working on traditional caching patterns (session store, leaderboard, rate limiting), general setup, monitoring, or migration without a GenAI component.

Loading

Read this file first. Load files listed below on demand based on the routing decision.

Check for existing context

Before starting, read .elasticache/requirements.json if it exists. If the genai section is set (mode, mode_2_path, framework), use those values instead of re-asking. If infrastructure.embedding_module is set and the file exists, import from it.

Three-Way GenAI Routing

Before recommending a pattern, classify the user's need into one of these three modes. Many teams asking for "RAG cache" actually need semantic response reuse (Mode 2), not full vector search (Mode 3). Ask a clarifying question before jumping to the heaviest option.

Mode 1: Plain Cache

Standard caching of LLM or API responses by exact key match. No vector index needed.

When to use: the user wants to avoid repeated identical LLM calls. Responses are looked up by an exact key (prompt hash, request fingerprint, or deterministic cache key).

Deployment: serverless Valkey. This is just regular SET/GET. Route to data-modeling sub-skill for key schema and TTL guidance.

Mode 2: Semantic Response Cache

Cache LLM responses with semantic similarity matching so "nearly identical" prompts return a cached response.

When to use: the user wants fuzzy cache hits, where semantically close prompts share a cached response.

Two implementation paths:

• Application-side embedding comparison: app computes embeddings, stores as plain keys, does similarity math client-side. Serverless OK. Works on ANY Valkey/Redis version.
• Server-side vector similarity: Valkey performs the similarity search via FT.SEARCH. Requires node-based Valkey 8.2 or above (recommend 9.0).

Ask the user which approach they prefer. Default to application-side unless they have high query volume AND confirmed Valkey 8.2 or above node-based. If the user's Valkey version is < 8.2, application-side is the ONLY option. Always load elasticache-search.md and use the supports_ft_search() version check before generating any FT.* code.

Deployment: serverless Valkey for application-side; node-based Valkey 8.2 or above (recommend 9.0) for server-side. Load: semantic-cache.md, elasticache-search.md (if server-side), embedding-providers.md

Mode 3: Full Vector Search

Vector indexing, KNN/ANN queries, RAG retrieval, recommendation via embeddings, agent memory with vector recall, or any workload that requires Valkey to maintain and query a vector index.

When to use: the user needs to store many embeddings and retrieve the top-K most similar. Typical for RAG knowledge bases, agent memory, recommendation engines, semantic search, catalog search.

Deployment: node-based Valkey 8.2 or above (mandatory; recommend 9.0). No serverless. Load: elasticache-search.md plus the relevant pattern file (rag-retrieval.md, agent-memory.md, or semantic-cache.md), embedding-providers.md, framework-guide.md (if framework mentioned)

How to classify

1. Does the user need similarity matching at all? If no, use Mode 1. 2. Is the similarity matching scoped to caching LLM responses? If yes, use Mode 2 and ask application-side vs server-side. 3. Does the user need to index, store, and query a corpus of embeddings? If yes, use Mode 3.

Hard Routing Rules (non-negotiable)

1. Vector search MUST use node-based Valkey 8.2 or above (recommend 9.0). Serverless does NOT support vector search. Never recommend serverless for vector search workloads, not even as a temporary or future option.

2. Global Datastore MUST use node-based clusters. If the GenAI workload requires multi-Region replication, it must be node-based.

3. Vector search is NOT available on data tiering node types (r6gd family). Do not recommend r6gd instances for any workload requiring vector search.

4. Changing embedding providers requires full re-indexing. If the user wants to switch embedding models, warn that ALL existing vectors must be deleted and re-embedded. This is destructive. Confirm before proceeding.

Common Mistakes to Avoid

1. Do NOT generate FT.CREATE or FT.SEARCH code without first loading elasticache-search.md and including the supports_ft_search() guard. 2. Do NOT use numpy for vector byte packing. Use struct.pack only. 3. When using raw valkey-py/redis-py client directly, do NOT use the high-level redis.commands.search or valkey.commands.search Python wrappers for FT.* commands. Use execute_command() instead. When using supported frameworks like langgraph-checkpoint-aws (ValkeyStore) or Mem0, their built-in abstractions (e.g., store.search(), store.put(), m.add(), m.search()) are acceptable. 4. Do NOT recommend serverless for any Mode 3 workload or Mode 2 server-side. 5. Do NOT generate inline embedding code. Always import from the shared utility file (infrastructure.embedding_module). 6. Do NOT assume the user's Valkey version supports vector search. Always check with supports_ft_search(). 7. Do NOT skip the embedding provider selection step. The FT.CREATE DIM must match the embedding model's output dimensions exactly. 8. Be aware of an inconsistency in AWS docs regarding the HNSW M parameter: the FT.CREATE command doc states the maximum is 512, while the vector search limits page (vector-search-features-limits.md) states the maximum is 2,000,000. Treat the limits page as authoritative for parametric restrictions.

Data-Plane Access

Mode 2 (server-side) and Mode 3 workloads need data-plane access to run FT.CREATE, FT.SEARCH, JSON operations, and other Valkey commands. The primary path is the valkey-py client (pip install valkey). Connect through the cache endpoint (via the jump host or SSM tunnel covered in setup). The agent writes Python that runs each command via execute_command(), consistent with the guidance in "Common Mistakes to Avoid" rule 3.

Engine Requirement

All GenAI patterns require Valkey. Vector search patterns specifically require Valkey 8.2 or above on node-based clusters (recommend 9.0).

Vector search is available with Valkey 8.2 or above on node-based clusters in all AWS Regions at no additional cost. Not supported on data-tiering instances (r6gd) or serverless caches.

If the user doesn't have a cache yet, hand off to setup and ensure node-based Valkey 8.2 or above is selected (recommend 9.0) when vector search is needed.

Workflow

1. Classify the user's need (Mode 1/2/3). If genai.mode is already set in requirements.json, skip classification. 2. Persist the mode to requirements.json under genai.mode. If Mode 2, also persist genai.mode_2_path ("app-side" or "server-side"). 3. If Mode 2 (server-side) or Mode 3: check requirements.json for infrastructure.embedding_module. If set and the file exists, import from it. If not set, load embedding-providers.md, ask the user for their preferred model, generate the reusable utility file in their project, and persist the choice. 4. Load the relevant reference files per the table above 5. If vector search is needed, verify node-based Valkey 8.2 or above (recommend 9.0; load elasticache-search.md for constraints) 6. Walk through the pattern-specific implementation guide. All generated code imports from the embedding utility created in step 3 rather than generating inline embedding code. 7. If the user mentions a framework, load framework-guide.md and persist the choice to genai.framework in requirements.json.

After implementation

Update requirements artifact

After the pattern is implemented, update .elasticache/requirements.json. GenAI owns the genai section. Read the existing file first, merge your updates, then write it back. Do not overwrite fields owned by setup or requirements.

Freshness disclaimer

When your response includes pricing, version constraints, or feature availability, include the freshness disclaimer per SKILL.md Global Rule #5: "For current pricing see https://aws.amazon.com/elasticache/pricing/. For current feature availability see https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/."

RAG Retrieval with ElastiCache Valkey

When to Use

RAG retrieval from ElastiCache Valkey. Best for:

• Real-time knowledge bases: Valkey indexes update automatically when matching keys are mutated. No batch re-indexing required.
• Low-latency agentic AI retrieval: Sub-millisecond vector search vs hundreds of ms from dedicated vector DBs.
• Colocated vector + cache workloads: The vector store lives alongside your application cache, eliminating a separate service hop.

Deployment Requirement

Node-based Valkey 8.2 or above clusters only (recommend 9.0). Vector search is NOT available on serverless. See elasticache-search.md for full platform constraints and limits.

Cluster mode: If running on a multi-shard cluster, use hash tags to ensure all document chunks for a single index land on the same shard:

doc:{corpus}:{doc_id}:{chunk_id}

The {corpus} hash tag forces slot co-location. Without this, FT.SEARCH returns partial results (it only queries the shard it executes on). See elasticache-search.md Hash Slot Constraint for details.

Note: The code examples below use the simplified key pattern doc:{doc_id}:{chunk_id} which assumes a single-shard cluster. For multi-shard clusters, adapt the key pattern to include a hash tag, e.g., doc:{corpus}:{doc_id}:{chunk_id}, and update the index prefix accordingly.

---

Step 1: Design the Document Schema

Use HASH keys with a consistent prefix. All keys matching the prefix are auto-indexed.

Key pattern: doc:{doc_id}:{chunk_id}

Fields:
  embedding    - FLOAT32 bytes (match your embedding model's dimensions)
  text         - the chunk text (plain string, not indexed for search)
  source       - source document name/path (TAG)
  category     - document category (TAG)
  created_at   - unix timestamp (NUMERIC)
  chunk_index  - position in source document (NUMERIC)

Prefix-based scoping: When the index uses PREFIX 1 doc:, every HASH key starting with doc: is automatically indexed. New keys, updated keys, and deleted keys are reflected in the index without manual intervention.

---

Step 2: Create the Index

from utils.embeddings import VECTOR_DIM

# Note: Maximum of 10 indexes can be created per cluster.
# For multi-tenant RAG, use TAG pre-filters on a shared index rather than per-tenant indexes.
client.execute_command(
    "FT.CREATE", "idx:docs",
    "ON", "HASH",
    "PREFIX", "1", "doc:",
    "SCHEMA",
    "embedding", "VECTOR", "HNSW", "6",
        "TYPE", "FLOAT32", "DIM", str(VECTOR_DIM), "DISTANCE_METRIC", "COSINE",
    "source", "TAG",
    "category", "TAG",
    "created_at", "NUMERIC",
    "chunk_index", "NUMERIC",
)

After creation, if matching keys already exist, Valkey backfills the index in the background. Query operations attempted while an index is undergoing backfill are not allowed and are terminated with an error. Wait for readiness before querying:

Backfill types: During initial index creation (FT.CREATE), queries against the index are blocked and return an error until backfill completes. However, during scaling events (e.g., adding shards), the index may undergo backfill with reduced recall for search queries — queries are allowed but may return incomplete results.

import time

def wait_for_index_ready(client, index_name, timeout=60):
    """Poll FT.INFO until index state is 'ready'.

    Note: FT.INFO returns a flat array with nested sub-arrays for some fields.
    This simple zip approach works for top-level fields like 'state' but is
    fragile if the response structure changes. For production use, iterate
    through the list looking for the 'state' key explicitly.
    """
    start = time.time()
    while time.time() - start < timeout:
        info = client.execute_command("FT.INFO", index_name)
        info_dict = dict(zip(info[::2], info[1::2]))
        state = info_dict.get("state") or info_dict.get(b"state")
        if state in ("ready", b"ready"):
            return True
        time.sleep(0.5)
    raise TimeoutError(f"Index {index_name} not ready after {timeout}s")

Replica backfill caveat: Backfill completion is not synchronized between primary and replicas. If your application reads from replicas, verify backfill completion on all replicas before issuing search queries. See elasticache-search.md for details.

---

Step 3: Ingest Documents

Chunking, embedding, and storing in a pipeline batch:

from utils.embeddings import generate_embedding, embedding_to_bytes

def ingest_documents(client, documents, batch_size=50):
    pipe = client.pipeline(transaction=False)
    count = 0

    for doc in documents:
        chunks = chunk_document(doc["text"])
        for i, chunk_text in enumerate(chunks):
            embedding = generate_embedding(chunk_text)

            key = f"doc:{doc['id']}:{i}"
            pipe.hset(key, mapping={
                "embedding": embedding_to_bytes(embedding),
                "text": chunk_text,
                "source": doc.get("source", ""),
                "category": doc.get("category", ""),
                "created_at": str(doc.get("created_at", -1)),
                "chunk_index": str(i),
            })
            count += 1

            if count % batch_size == 0:
                pipe.execute()
                pipe = client.pipeline(transaction=False)

    pipe.execute()
    return count

Chunking guidance:

Prefer chunking by semantic boundaries (paragraphs, sections, headers) over fixed-size splits when the source has structure. Fixed-size is acceptable for unstructured text.

Overlap prevents information loss at chunk boundaries. A sentence split across two chunks without overlap is lost to both. 10-20% overlap is the standard range.

Note: Bedrock Titan Embed v2 supports up to 8192 input tokens. For retrieval tasks, AWS recommends segmenting documents into logical segments such as paragraphs or sections rather than embedding at the maximum token length.

Real-time updates: To update a document chunk, just HSET the same key with new field values. Valkey re-indexes automatically. No rebuild needed.

---

Step 4: Retrieve

Supports pure vector search or hybrid (vector + metadata pre-filters).

The filter expression is a pre-filter: it narrows the candidate set before KNN runs.

from utils.embeddings import generate_embedding, embedding_to_bytes

def retrieve_chunks(client, query_text, category=None,
                    min_timestamp=None, top_k=5):
    query_bytes = embedding_to_bytes(generate_embedding(query_text))

    filter_parts = []
    if category:
        safe_cat = category.replace("-", "\\-")
        filter_parts.append(f"@category:{{{safe_cat}}}")
    if min_timestamp:
        filter_parts.append(f"@created_at:[{min_timestamp} +inf]")

    if filter_parts:
        pre_filter = "(" + " ".join(filter_parts) + ")"
        query_str = f"{pre_filter}=>[KNN {top_k} @embedding $vec AS score]"
    else:
        query_str = f"*=>[KNN {top_k} @embedding $vec AS score]"

    results = client.execute_command(
        "FT.SEARCH", "idx:docs", query_str,
        "PARAMS", "2", "vec", query_bytes,
        "RETURN", "3", "text", "source", "score",
        "LIMIT", "0", str(top_k),
        "DIALECT", "2",
    )

    chunks = []
    if results[0] == 0:
        return chunks

    for i in range(1, len(results), 2):
        fields = results[i + 1]
        field_dict = dict(zip(fields[::2], fields[1::2]))
        text = field_dict.get(b"text", b"").decode()
        source = field_dict.get(b"source", b"").decode()
        score = float(field_dict.get(b"score", b"0"))
        # COSINE distance is [0, 2] (0=identical, 2=opposite). Convert to [0, 1] similarity.
        similarity = 1.0 - (score / 2.0)
        chunks.append({"text": text, "source": source, "similarity": similarity})

    return chunks

---

Step 5: Delete Documents

When source documents are deleted or fully replaced:

def delete_document_chunks(client, doc_id):
    """Remove all chunks for a document. Index removes them automatically."""
    cursor = "0"
    prefix = f"doc:{doc_id}:"
    while True:
        cursor, keys = client.scan(cursor=cursor, match=f"{prefix}*", count=100)
        if keys:
            client.delete(*keys)
        if cursor == 0:
            break

For bulk refresh: delete existing chunks, then re-ingest the updated document.

---

Cross-References

• Embedding providers and utility setup: see embedding-providers.md
• Platform constraints, FT.SEARCH encoding, HNSW tuning: see elasticache-search.md
• Framework integration (LangChain, Strands): see framework-guide.md

Semantic Cache Implementation Guide

When to use

Semantic cache for LLM/API responses. Avoids redundant inference calls when prompts are semantically similar (not just exact match). Two deployment options:

• Application-side comparison (serverless OK): Generate embeddings in app code, compare locally. No vector index needed in Valkey, just store/retrieve by key. Works with ElastiCache Serverless.
• Server-side vector similarity (node-based Valkey 8.2 or above required; recommend 9.0): FT.SEARCH with KNN finds the nearest cached prompt. Sub-millisecond lookup. Requires ElastiCache node-based with search enabled.

This guide covers the server-side approach. For the application-side approach (serverless OK, any Valkey version), use the python_knn_search fallback in elasticache-search.md with semantic cache key patterns below.

*Before using any FT. code below**, call supports_ft_search(client) from elasticache-search.md. If it returns False, use the Python-side fallback instead.

Key design

Use a dual-key pattern. Separate the vector index hash from the response payload:

semcache:vec:{request_id}   # HASH: embedding, request_id, timestamp, filter fields
semcache:rr:{request_id}    # HASH: request_text, response_text, created_at

Why separate keys: the vector index only scans the vec: prefix. Response payloads can be large (full LLM output) and do not need indexing. Keeping them out of the index reduces memory pressure on HNSW graph traversal.

Cluster mode: If running on a multi-shard cluster, use hash tags to ensure all keys for a single index land on the same shard:

semcache:vec:{myapp}:{request_id}
semcache:rr:{myapp}:{request_id}

The {myapp} hash tag forces slot co-location. Without this, FT.SEARCH returns partial results (it only queries the shard it executes on). See elasticache-search.md Hash Slot Constraint for details. For single-shard clusters, hash tags are optional since all keys land on the same shard.

PREFIX_VECTOR = "semcache:vec:"
PREFIX_RR     = "semcache:rr:"
INDEX_NAME    = "idx:semcache"

Step 1: Create the index

import time
import uuid
from utils.embeddings import VECTOR_DIM

def create_index(client):
    """Create HNSW COSINE vector index. Idempotent."""
    try:
        client.execute_command(
            "FT.CREATE", INDEX_NAME,
            "ON", "HASH",
            "PREFIX", "1", PREFIX_VECTOR,
            "SCHEMA",
            "embedding", "VECTOR", "HNSW", "6",
                "TYPE", "FLOAT32",
                "DIM", str(VECTOR_DIM),
                "DISTANCE_METRIC", "COSINE",
            "request_id", "TAG",
            "scope", "TAG", "SEPARATOR", ",",
            "timestamp", "NUMERIC",
        )
    except Exception as e:
        if "already exists" not in str(e).lower():
            raise

Index limit: A maximum of 10 indexes can be created per cluster. For multi-tenant designs, use TAG-based filtering within a single index rather than creating per-tenant indexes.

HNSW tuning parameters: The index above uses HNSW defaults (M=16, EF_CONSTRUCTION=200, EF_RUNTIME=10). For semantic cache workloads, the default EF_RUNTIME of 10 may yield suboptimal recall; consider increasing it (e.g., 50-200) via the EF_RUNTIME query modifier on FT.SEARCH to improve cache hit detection. Higher M values (e.g., 32+) improve recall at the cost of memory. EF_CONSTRUCTION values of 200-400 are generally sufficient. See the vector search overview for detailed guidance on tuning these parameters.

Step 2: Cache lookup (FT.SEARCH)

All code below uses the shared embedding utility:

from utils.embeddings import generate_embedding, embedding_to_bytes, VECTOR_DIM

SIMILARITY_THRESHOLD = 0.90  # default cosine similarity; tune per use case

def cache_lookup(client, query_vec: bytes, threshold: float = SIMILARITY_THRESHOLD) -> dict | None:
    """Search for a semantically similar cached prompt using a precomputed query
    embedding (bytes from embedding_to_bytes). Returns hit dict or None."""

    result = client.execute_command(
        "FT.SEARCH", INDEX_NAME,
        "*=>[KNN 1 @embedding $vec AS score]",
        "PARAMS", "2", "vec", query_vec,
        "RETURN", "2", "request_id", "score",
        "DIALECT", "2",
    )

    if not result or int(result[0]) == 0:
        return None

    # Parse result: [total_hits, key_name, [field, value, ...]]
    # With decode_responses=False, all values are bytes
    fields = result[2]
    doc = {}
    for i in range(0, len(fields), 2):
        k = fields[i].decode() if isinstance(fields[i], bytes) else fields[i]
        v = fields[i+1].decode() if isinstance(fields[i+1], bytes) else fields[i+1]
        doc[k] = v

    # Cosine distance to similarity: distance 0=identical, 2=opposite.
    similarity = 1.0 - (float(doc["score"]) / 2.0)

    if similarity < threshold:
        return None

    # Fetch the cached response from the rr key
    request_id = doc["request_id"]
    rr_key = f"{PREFIX_RR}{request_id}"
    rr_data = client.hgetall(rr_key)
    if not rr_data:
        return None

    # Decode bytes keys/values from hgetall
    response_text = rr_data.get(b"response_text", b"").decode()

    return {
        "response": response_text,
        "similarity": similarity,
        "request_id": request_id,
    }

Step 3: Cache store (HSET)

def cache_store(client, prompt: str, response: str, embedding_bytes: bytes,
                scope: str = "", ttl: int = 3600):
    """Store prompt+response pair. Sets TTL on both keys."""
    request_id = str(uuid.uuid4())
    now = time.time()

    # Vector key (indexed)
    vec_key = f"{PREFIX_VECTOR}{request_id}"
    client.hset(vec_key, mapping={
        "embedding": embedding_bytes,
        "request_id": request_id,
        "timestamp": str(now),
        "scope": scope if scope else "",
    })

    # Response key (not indexed)
    rr_key = f"{PREFIX_RR}{request_id}"
    client.hset(rr_key, mapping={
        "request_text": prompt,
        "response_text": response,
        "created_at": str(now),
    })

    if ttl > 0:
        # Add random jitter to spread out cache invalidations and prevent
        # thundering herd when many entries expire simultaneously.
        import random
        jitter = random.randint(0, max(1, ttl // 10))  # up to 10% jitter
        client.expire(vec_key, ttl + jitter)
        client.expire(rr_key, ttl + jitter)

Step 4: Full flow

def semantic_cache_query(client, prompt: str, llm_fn, threshold: float = 0.90,
                         scope: str = "", ttl: int = 3600) -> dict:
    """
    Complete semantic cache flow.
    llm_fn: callable that takes a prompt string and returns response string.
    """
    # Compute the query embedding once and reuse it for lookup and store.
    embedding_bytes = embedding_to_bytes(generate_embedding(prompt))

    # Lookup
    hit = cache_lookup(client, embedding_bytes, threshold=threshold)
    if hit:
        return {"response": hit["response"], "source": "cache", "similarity": hit["similarity"]}

    # Miss: call LLM
    response = llm_fn(prompt)

    # Store (reuses the embedding already computed above)
    cache_store(client, prompt, response, embedding_bytes, scope=scope, ttl=ttl)

    return {"response": response, "source": "llm", "similarity": 0.0}

Similarity thresholds

Note: ElastiCache vector search uses cosine distance (1 - cosine_similarity), where 0 = identical and 1 = orthogonal. A distance threshold of 0.10 corresponds to cosine similarity ≥ 0.90.

Starting recommendations:

Lower threshold = more cache hits but higher risk of returning a semantically incorrect answer. Start strict (0.92) and lower based on observed false-hit rate.

Advanced: hybrid filtering

Add TAG or NUMERIC pre-filters to scope cache hits before vector similarity runs. This narrows the candidate set so KNN only compares within a relevant subset.

# Pre-filter by model version and user segment, then KNN
pre_filter = "(@scope:{bedrock_claude_v4} @timestamp:[1700000000 +inf])"
query = f"{pre_filter}=>[KNN 1 @embedding $vec AS score]"

result = client.execute_command(
    "FT.SEARCH", INDEX_NAME, query,
    "PARAMS", "2", "vec", query_vec,
    "RETURN", "2", "request_id", "score",
    "DIALECT", "2",
)

Filter patterns:

• @scope:{model_v2} filters TAG field to exact token
• @timestamp:[{cutoff} +inf] filters NUMERIC to recent entries only
• Multiple filters combine with implicit AND inside parentheses
• Escape hyphens in TAG values: my\\-value

Session Store Implementation Guide

Session state, message history, and resumable conversations with ElastiCache Valkey. Plain data structures, no vector search, no embedding model. Works on serverless.

Serverless eviction policy: Serverless caches use volatile-lru (not configurable), which only evicts keys that have a TTL set. Always set a TTL on every key. Keys without a TTL will never be evicted and can cause OOM errors. The allkeys-lru policy recommended in some best-practice guides applies only to node-based clusters.

Key Design

{conv:session_id}:messages     # JSON-encoded list of message dicts
{conv:session_id}:metadata     # JSON-encoded conversation metadata
conv:user:{user_id}:sessions   # sorted set of session_ids, score = timestamp

Hash tag note: The {conv:session_id} hash tag ensures messages and metadata keys hash to the same slot on cluster-mode-enabled caches (including serverless). Without a hash tag, pipeline operations on these keys are non-atomic across slots and a mid-pipeline failure could leave partial state. The conv:user:{user_id}:sessions index key does not share a hash tag with the message/metadata keys, so it hashes to a different slot. This is acceptable because the index is advisory (listing sessions), not transactional.

Store and Load Pattern (Custom)

Use this when you're not on the Strands framework and need direct control over session storage.

# pip install valkey
# For serverless or cluster-mode-enabled node-based clusters:
#   from valkey.cluster import ValkeyCluster
# For cluster-mode-disabled node-based clusters:
#   import valkey  (use valkey.Valkey)
import json
import time
import valkey

CONVERSATION_TTL = 30 * 24 * 60 * 60  # 30 days (adjust to your retention policy)

_client = None

def get_client():
    """Lazy client initialization. Never create connections at module level.

    For serverless or cluster-mode-enabled node-based clusters, use
    valkey.cluster.ValkeyCluster instead of valkey.Valkey.
    Serverless is always cluster-mode-enabled. The single serverless
    endpoint abstracts slots to one virtual node, so ValkeyCluster
    works transparently. For cluster-mode-disabled node-based clusters,
    valkey.Valkey is correct.
    """
    global _client
    if _client is None:
        _client = valkey.ValkeyCluster(
            host="your-endpoint.serverless.use1.cache.amazonaws.com",  # serverless format; node-based differs
            port=6379, ssl=True, ssl_cert_reqs="required",  # verify server cert against trusted CAs (secure default)
            # ssl_cert_reqs="none",  # tunnel/dev ONLY (e.g. local SSH tunnel); never in production
            decode_responses=True, socket_timeout=5,
        )
    return _client

def save_session(session_id: str, messages: list[dict], metadata: dict,
                 user_id: str = None):
    """Persist conversation state. Call after each turn."""
    msg_key = f"{{conv:{session_id}}}:messages"
    meta_key = f"{{conv:{session_id}}}:metadata"
    client = get_client()

    pipe = client.pipeline()
    pipe.set(msg_key, json.dumps(messages, default=str))
    pipe.expire(msg_key, CONVERSATION_TTL)
    pipe.set(meta_key, json.dumps(metadata, default=str))
    pipe.expire(meta_key, CONVERSATION_TTL)

    if user_id:
        # Note: idx_key hashes to a different slot than {conv:session_id} keys.
        # On cluster-mode-enabled caches (including serverless), this pipeline
        # becomes a multi-slot pipeline. ValkeyCluster handles this transparently
        # by splitting commands across slots, but atomicity is only guaranteed
        # within a single slot. If you need atomic user-index updates, issue
        # the ZADD/EXPIRE separately outside the pipeline.
        idx_key = f"conv:user:{user_id}:sessions"
        pipe.zadd(idx_key, {session_id: time.time()})
        pipe.expire(idx_key, CONVERSATION_TTL)

    pipe.execute()

def load_session(session_id: str) -> dict | None:
    """Load persisted conversation. Returns {messages, metadata} or None."""
    client = get_client()
    raw_messages = client.get(f"{{conv:{session_id}}}:messages")
    if not raw_messages:
        return None
    raw_metadata = client.get(f"{{conv:{session_id}}}:metadata")
    return {
        "messages": json.loads(raw_messages),
        "metadata": json.loads(raw_metadata) if raw_metadata else {},
    }

def list_user_sessions(user_id: str, limit: int = 20) -> list[str]:
    """Most recent sessions first."""
    return get_client().zrevrange(f"conv:user:{user_id}:sessions", 0, limit - 1)

Strands Integration

The strands-valkey-session-manager package (community package, v0.1.0+ — MIT license, maintained by jeromevdl) provides a drop-in ValkeySessionManager. Five lines of setup:

import valkey
from strands import Agent
from strands_valkey_session_manager import ValkeySessionManager

_strands_client = None

def get_strands_client():
    """Lazy client initialization. Never create connections at module level.
    Uses ValkeyCluster for serverless (cluster-mode-enabled).
    For cluster-mode-disabled node-based clusters, use valkey.Valkey instead.
    """
    global _strands_client
    if _strands_client is None:
        _strands_client = valkey.ValkeyCluster(
            host="your-endpoint.serverless.use1.cache.amazonaws.com", port=6379, ssl=True,
            ssl_cert_reqs="required", decode_responses=True,  # in-VPC EKS with real cert verification
            # tunnel/dev alternative: ssl_cert_reqs="none" (skips cert verification; not for production)
        )
    return _strands_client

def create_agent(session_id: str = "user-42") -> Agent:
    client = get_strands_client()
    session_mgr = ValkeySessionManager(session_id=session_id, client=client)
    return Agent(system_prompt="You are a helpful assistant.", session_manager=session_mgr)

# Usage:
# agent = create_agent("user-42")
# response = agent("My name is Alex and I'm building a RAG pipeline.")

The session manager stores three key types in Valkey: session:{id} (session record), session:{id}:agent:{agent_id} (agent state), and session:{id}:agent:{agent_id}:message:{msg_id} (individual messages).

TTL Recommendations

---

Cross-References

• Semantic memory (vector-based recall across sessions): see agent-memory.md
• LangChain/LlamaIndex/Strands framework integration: see framework-guide.md