Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.

Implementierung eines semantischen Caches mit ElastiCache for Valkey

Die folgende exemplarische Vorgehensweise zeigt, wie Sie einen semantischen Lesecache mithilfe von ElastiCache for Valkey mit Amazon Bedrock implementieren.

Schritt 1: Erstellen Sie einen For-Valkey-Cluster ElastiCache

Erstellen Sie einen ElastiCache For Valkey-Cluster mit Version 8.2 oder höher mit dem: 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

Schritt 2: Connect zum Cluster her und konfigurieren Sie Einbettungen

Stellen Sie von Ihrem Anwendungscode aus, der auf Ihrer Amazon EC2 EC2-Instance ausgeführt wird, eine Verbindung zum ElastiCache Cluster her und richten Sie das Einbettungsmodell ein:

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"
)

Ersetzen Sie den Host-Wert durch den ElastiCache Konfigurationsendpunkt Ihres Clusters. Anweisungen zur Suche nach Ihrem Cluster-Endpunkt finden Sie unter Zugriff auf Ihren ElastiCache Cluster.

Schritt 3: Erstellen Sie den Vektorindex für den semantischen Cache

Konfigurieren Sie eine ValkeyStore , die Abfragen mithilfe eines HNSW-Index mit COSINE-Abstand für die Vektorsuche automatisch einbettet:

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()

Schritt 4: Implementieren Sie die Cache-Such- und Aktualisierungsfunktionen

Erstellen Sie Funktionen, um den Cache nach semantisch ähnlichen Abfragen zu durchsuchen und neue Abfrage-Antwort-Paare zu speichern:

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
        }
    )

Schritt 5: Implementieren Sie das Lese-Through-Cache-Muster

Integrieren Sie den Cache in die Anforderungsverarbeitung Ihrer Anwendung:

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),
    }

Zugrundeliegende Valkey-Befehle

Die folgende Tabelle zeigt die Valkey-Befehle, die zur Implementierung des semantischen Caches verwendet wurden: