CacheChecker

CacheChecker

CacheChecker는 지정한 캐시 필드(cache field)를 기준으로 Document Store 안에 문서가 존재하는지 확인하는 컴포넌트예요. 캐싱을 구현할 때 이미 처리한 문서를 다시 처리하지 않도록 건너뛰는 데 유용합니다.

출처: 문서

본문

Overview

CacheChecker는 Document Store 안에 items 입력 변수로 받은 값들 중 하나라도 cache_field에 해당하는 값을 가진 문서가 있는지 확인해요. 결과로 두 개의 키를 가진 딕셔너리를 반환합니다: "hits"(캐시에서 찾은 문서 리스트)와 "misses"(캐시에서 찾지 못한 값들의 리스트).

Usage

On its own

from haystack.components.caching import CacheChecker
from haystack.document_stores.in_memory import InMemoryDocumentStore

my_doc_store = InMemoryDocumentStore()

# For URL-based caching
cache_checker = CacheChecker(document_store=my_doc_store, cache_field="url")
cache_check_results = cache_checker.run(
    items=[
        "https://example.com/resource",
        "https://another_example.com/other_resources",
    ],
)
print(
    cache_check_results["hits"],
)  # List of Documents that were found in the cache: all of these have 'url': <one of the above> in the metadata
print(
    cache_check_results["misses"],
)  # URLs that were not found in the cache, like ["https://example.com/resource"]

# For caching based on a custom identifier
cache_checker = CacheChecker(document_store=my_doc_store, cache_field="metadata_field")
cache_check_results = cache_checker.run(items=["12345", "ABCDE"])
print(
    cache_check_results["hits"],
)  # Documents that were found in the cache: all of these have 'metadata_field': <one of the above> in the metadata
print(
    cache_check_results["misses"],
)  # Values that were not found in the cache, like: ["ABCDE"]

In a pipeline

from haystack import Pipeline
from haystack.components.converters import TextFileToDocument
from haystack.components.preprocessors import DocumentCleaner, DocumentSplitter
from haystack.components.writers import DocumentWriter
from haystack.components.caching import CacheChecker
from haystack.document_stores.in_memory import InMemoryDocumentStore

pipeline = Pipeline()
document_store = InMemoryDocumentStore()
pipeline.add_component(
    instance=CacheChecker(document_store, cache_field="meta.file_path"),
    name="cache_checker",
)
pipeline.add_component(instance=TextFileToDocument(), name="text_file_converter")
pipeline.add_component(instance=DocumentCleaner(), name="cleaner")
pipeline.add_component(
    instance=DocumentSplitter(split_by="sentence", split_length=250, split_overlap=30),
    name="splitter",
)
pipeline.add_component(
    instance=DocumentWriter(document_store=document_store),
    name="writer",
)
pipeline.connect("cache_checker.misses", "text_file_converter.sources")
pipeline.connect("text_file_converter.documents", "cleaner.documents")
pipeline.connect("cleaner.documents", "splitter.documents")
pipeline.connect("splitter.documents", "writer.documents")

pipeline.draw("pipeline.png")
# Take the current directory as input and run the pipeline
result = pipeline.run({"cache_checker": {"items": ["code_of_conduct_1.txt"]}})
print(result)
# The second execution skips the files that were already processed
result = pipeline.run({"cache_checker": {"items": ["code_of_conduct_1.txt"]}})
print(result)

위 예시처럼 파이프라인에서 CacheChecker를 맨 앞에 두고 misses(캐시에 없는 값들)만 다음 단계로 연결하면, 두 번째 실행부터는 이미 처리한 파일을 건너뛰는 캐싱 파이프라인을 만들 수 있어요.

더 알아보기 (Learn more)

• CacheChecker — Haystack 공식 문서