ValkeyDocumentStore

Valkey를 백엔드로 쓰는 document store예요. Valkey는 고성능 인메모리 데이터 구조 서버라서 RAG 같은 벡터 유사도 검색 워크로드에 잘 맞아요.

API 레퍼런스: Valkey GitHub 링크: https://github.com/deepset-ai/haystack-core-integrations/tree/main/integrations/valkey

출처: 문서

본문

Valkey는 고성능 인메모리 데이터 구조 저장소예요. ValkeyDocumentStore로 Haystack 파이프라인에서 쓸 수 있어요. Valkey는 기본적으로 최대 성능을 위해 인메모리로 동작하지만, 데이터 영속성을 위해 지속성(persistence) 옵션으로 구성할 수도 있어요.

ValkeyDocumentStore는 검색 모듈이 실행되는 Valkey 서버에 연결해요. 그리고 RAG와 다른 검색 유스케이스용 벡터 유사도 검색을 지원해요. 사용 가능한 모든 메서드와 설정에 대한 자세한 개요는 API Reference를 참고하세요.

설치 (Installation)

Valkey Haystack 통합은 이렇게 설치해요:

pip install valkey-haystack

이 페이지의 예시들은 sentence-transformers-haystack 패키지로 옮겨간 Sentence Transformers embedder를 사용해요. 예시를 실행하려면 설치하세요:

pip install sentence-transformers-haystack

초기화 (Initialization)

Haystack 파이프라인의 데이터 저장소로 Valkey를 쓰려면 검색 모듈이 실행되는 Valkey 서버가 필요해요. ValkeyDocumentStore는 이렇게 초기화해요:

from haystack_integrations.document_stores.valkey import ValkeyDocumentStore

document_store = ValkeyDocumentStore(
    nodes_list=[("localhost", 6379)],
    index_name="my_documents",
    embedding_dim=768,
    distance_metric="cosine",
)

Valkey 로컬 실행 (Running Valkey locally)

개발과 테스트를 위해 Docker로 Valkey 서버를 시작할 수 있어요:

docker run -d -p 6379:6379 valkey/valkey-bundle:latest

그런 다음 위와 같은 초기화 코드로 nodes_list=[("localhost", 6379)]를 사용해 연결하면 돼요.

더 고급 설정과 클러스터링 구성은 Valkey 문서를 참고하세요.

문서 쓰기 (Writing documents)

ValkeyDocumentStore에 문서를 쓰려면 인덱싱 파이프라인을 만들거나 write_documents() 메서드를 사용하면 돼요. 데이터를 가져오고 준비하는 데는 Converter, PreProcessor 같은 통합을 쓸 수 있어요. 아래는 Markdown 파일을 Valkey에 인덱싱하는 예시예요.

인덱싱 파이프라인 (Indexing pipeline)

from haystack import Pipeline
from haystack.components.converters import MarkdownToDocument
from haystack.components.writers import DocumentWriter
from haystack_integrations.components.embedders.sentence_transformers import (
    SentenceTransformersDocumentEmbedder,
)
from haystack.components.preprocessors import DocumentSplitter
from haystack_integrations.document_stores.valkey import ValkeyDocumentStore

document_store = ValkeyDocumentStore(
    nodes_list=[("localhost", 6379)],
    index_name="my_documents",
    embedding_dim=768,
    distance_metric="cosine",
)
indexing = Pipeline()
indexing.add_component("converter", MarkdownToDocument())
indexing.add_component(
    "splitter",
    DocumentSplitter(split_by="sentence", split_length=2),
)
indexing.add_component("embedder", SentenceTransformersDocumentEmbedder())
indexing.add_component("writer", DocumentWriter(document_store))
indexing.connect("converter", "splitter")
indexing.connect("splitter", "embedder")
indexing.connect("embedder", "writer")
indexing.run({"converter": {"sources": ["filename.md"]}})

RAG 파이프라인에서 Valkey 사용하기 (Using Valkey in a RAG pipeline)

문서가 ValkeyDocumentStore에 들어가면 ValkeyEmbeddingRetriever로 검색할 수 있어요. 아래 예시는 커스텀 프롬프트로 RAG 파이프라인을 만드는 경우지요:

from haystack import Pipeline
from haystack.utils import Secret
from haystack.dataclasses import ChatMessage
from haystack_integrations.components.embedders.sentence_transformers import (
    SentenceTransformersTextEmbedder,
)
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack_integrations.document_stores.valkey import ValkeyDocumentStore
from haystack_integrations.components.retrievers.valkey import ValkeyEmbeddingRetriever

document_store = ValkeyDocumentStore(
    nodes_list=[("localhost", 6379)],
    index_name="my_documents",
    embedding_dim=768,
    distance_metric="cosine",
)
prompt_template = [
    ChatMessage.from_system(
        "Answer the question based on the provided context. If the context does not include an answer, reply with 'I don't know'.",
    ),
    ChatMessage.from_user(
        "Query: {{query}}\n"
        "Documents:\n{% for doc in documents %}{{ doc.content }}\n{% endfor %}\n"
        "Answer:",
    ),
]

query_pipeline = Pipeline()
query_pipeline.add_component("text_embedder", SentenceTransformersTextEmbedder())
query_pipeline.add_component(
    "retriever",
    ValkeyEmbeddingRetriever(document_store=document_store),
)
query_pipeline.add_component(
    "prompt_builder",
    ChatPromptBuilder(
        template=prompt_template,
        required_variables=["query", "documents"],
    ),
)
query_pipeline.add_component(
    "generator",
    OpenAIChatGenerator(
        api_key=Secret.from_token("YOUR_OPENAI_API_KEY"),
        model="gpt-4o",
    ),
)

query_pipeline.connect("text_embedder.embedding", "retriever.query_embedding")
query_pipeline.connect("retriever.documents", "prompt_builder.documents")
query_pipeline.connect("prompt_builder.prompt", "generator.messages")

query = "What is Valkey?"
results = query_pipeline.run(
    {
        "text_embedder": {"text": query},
        "prompt_builder": {"query": query},
    },
)

더 많은 예시는 리포지토리의 examples 폴더를 확인하세요.

성능 이점 (Performance benefits)

인메모리 저장: 읽기·쓰기 연산이 빠르다. 높은 처리량: 초당 많은 연산을 처리한다. 낮은 지연 시간: 문서 연산의 응답 시간이 최소화된다. 확장성: 수평 확장을 위한 클러스터링을 지원한다.

지원되는 Retriever (Supported Retrievers)

ValkeyEmbeddingRetriever: 쿼리와 문서 임베딩을 비교해서 ValkeyDocumentStore에서 쿼리와 가장 관련 있는 문서를 가져온다.

ValkeyDocumentStore