Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Implementación de una caché semántica con ElastiCache for Valkey

En el siguiente tutorial se muestra cómo implementar una caché semántica de lectura completa utilizando ElastiCache Valkey con Amazon Bedrock.

Paso 1: Cree un clúster de Form Valkey ElastiCache

Cree un clúster ElastiCache para Valkey con la versión 8.2 o posterior mediante: AWS CLI

aws elasticache create-replication-group \
  --replication-group-id "valkey-semantic-cache" \
  --cache-node-type cache.r7g.large \
  --engine valkey \
  --engine-version 8.2 \
  --num-node-groups 1 \
  --replicas-per-node-group 1

Paso 2: Conectarse al clúster y configurar las incrustaciones

Desde el código de la aplicación que se ejecuta en la instancia de Amazon EC2, conéctese al ElastiCache clúster y configure el modelo de incrustación:

from valkey.cluster import ValkeyCluster
from langchain_aws import BedrockEmbeddings

# Connect to ElastiCache for Valkey
valkey_client = ValkeyCluster(
    host="mycluster.xxxxxx.clustercfg.use1.cache.amazonaws.com",  # Your cluster endpoint
    port=6379,
    decode_responses=False
)

# Set up Amazon Bedrock Titan embeddings
embeddings = BedrockEmbeddings(
    model_id="amazon.titan-embed-text-v2:0",
    region_name="us-east-1"
)

Sustituya el valor del host por el punto final de configuración del ElastiCache clúster. Para obtener instrucciones sobre cómo encontrar el punto final del clúster, consulte Acceder al ElastiCache clúster.

Paso 3: Crea el índice vectorial para la caché semántica

Configure una ValkeyStore que incorpore automáticamente las consultas utilizando un índice HNSW con la distancia COSINE para la búsqueda vectorial:

from langgraph_checkpoint_aws import ValkeyStore
from hashlib import md5

store = ValkeyStore(
    client=valkey_client,
    index={
        "collection_name": "semantic_cache",
        "embed": embeddings,
        "fields": ["query"],           # Fields to vectorize
        "index_type": "HNSW",          # Vector search algorithm
        "distance_metric": "COSINE",   # Similarity metric
        "dims": 1024                   # Titan V2 produces 1024-d vectors
    }
)
store.setup()

def cache_key_for_query(query: str):
    """Generate a deterministic cache key for a query."""
    return md5(query.encode("utf-8")).hexdigest()

Paso 4: Implemente las funciones de búsqueda y actualización de la memoria caché

Cree funciones para buscar en la caché consultas semánticamente similares y almacenar nuevos pares de consultas y respuestas:

def search_cache(user_message: str, k: int = 3, min_similarity: float = 0.8):
    """Look up a semantically similar cached response from ElastiCache."""
    hits = store.search(
        namespace="semantic-cache",
        query=user_message,
        limit=k
    )
    if not hits:
        return None

    # Sort by similarity score (highest first)
    hits = sorted(hits, key=lambda h: h["score"], reverse=True)
    top_hit = hits[0]
    score = top_hit["score"]

    if score < min_similarity:
        return None  # Below similarity threshold

    return top_hit["value"]["answer"]  # Return cached answer


def store_cache(user_message: str, result_message: str):
    """Store a new query-response pair in the semantic cache."""
    key = cache_key_for_query(user_message)
    store.put(
        namespace="semantic-cache",
        key=key,
        value={
            "query": user_message,
            "answer": result_message
        }
    )

Paso 5: Implemente el patrón de caché de lectura completa

Integre la memoria caché en el manejo de solicitudes de su aplicación:

import time

def handle_query(user_message: str) -> dict:
    """Handle a user query with read-through semantic cache."""
    start = time.time()

    # Step 1: Search the semantic cache
    cached_response = search_cache(user_message, min_similarity=0.8)

    if cached_response:
        # Cache hit - return cached response
        elapsed = (time.time() - start) * 1000
        return {
            "response": cached_response,
            "source": "cache",
            "latency_ms": round(elapsed, 1),
        }

    # Step 2: Cache miss - invoke LLM
    llm_response = invoke_llm(user_message)  # Your LLM invocation function

    # Step 3: Store the response in cache for future reuse
    store_cache(user_message, llm_response)

    elapsed = (time.time() - start) * 1000
    return {
        "response": llm_response,
        "source": "llm",
        "latency_ms": round(elapsed, 1),
    }

Comandos de Valkey subyacentes

En la siguiente tabla se muestran los comandos de Valkey que se utilizan para implementar la caché semántica: