기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

ElastiCache for Valkey를 사용하여 의미 체계 캐시 구현

다음 연습에서는 Amazon Bedrock과 함께 ElastiCache for Valkey를 사용하여 읽기-스루 의미 체계 캐시를 구현하는 방법을 보여줍니다.

1단계: Valkey용 ElastiCache 클러스터 생성

AWS CLI다음을 사용하여 버전 8.2 이상의 ElastiCache for Valkey 클러스터를 생성합니다.

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

2단계: 클러스터에 연결하고 임베딩 구성

Amazon EC2 인스턴스에서 실행되는 애플리케이션 코드에서 ElastiCache 클러스터에 연결하고 임베딩 모델을 설정합니다.

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

호스트 값을 ElastiCache 클러스터의 구성 엔드포인트로 바꿉니다. 클러스터 엔드포인트를 찾는 방법에 대한 지침은 ElastiCache 클러스터 액세스를 참조하세요.

3단계: 의미 체계 캐시에 대한 벡터 인덱스 생성

벡터 검색을 위해 COSINE 거리와 함께 HNSW 인덱스를 사용하여 쿼리를 자동으로 포함하는 ValkeyStore를 구성합니다.

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

4단계: 캐시 검색 및 업데이트 함수 구현

캐시에서 의미상 유사한 쿼리를 검색하고 새 쿼리-응답 페어를 저장하는 함수를 생성합니다.

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

5단계: 읽기-스루 캐시 패턴 구현

캐시를 애플리케이션의 요청 처리에 통합합니다.

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

기본 Valkey 명령

다음 표에는 의미 체계 캐시를 구현하는 데 사용되는 Valkey 명령이 나와 있습니다.