diff --git a/docs-website/docs/pipeline-components/caching/cachechecker.mdx b/docs-website/docs/pipeline-components/caching/cachechecker.mdx index ae74b8efded..057079c1bfa 100644 --- a/docs-website/docs/pipeline-components/caching/cachechecker.mdx +++ b/docs-website/docs/pipeline-components/caching/cachechecker.mdx @@ -63,6 +63,35 @@ print( ) # Values that were not found in the cache, like: ["ABCDE"] ``` +### Cache expiration + +By default, `CacheChecker` considers a matching document to be a cache hit regardless of when it was cached. +To enable cache expiration, provide a `ttl` and optionally specify the metadata field containing the cache timestamp. The default timestamp field is `cached_at`. + +```python +from datetime import timedelta + +cache_checker = CacheChecker( + document_store=my_doc_store, + cache_field="url", + ttl=timedelta(hours=24), +) +``` + +When `ttl` is configured, a matching document is considered a cache hit only if the timestamp in `cached_at` is within the specified TTL. Documents with an expired or missing `cached_at` timestamp are treated as cache misses. + +To use a different metadata field for the cache timestamp, specify `time_field`: + +```python +cache_checker = CacheChecker( + document_store=my_doc_store, + cache_field="url", + ttl=timedelta(hours=24), + time_field="created_at", +) +``` +When `ttl` is `None` (the default), `CacheChecker` retains its existing behavior and does not check the timestamp. + ### In a pipeline ```python diff --git a/haystack/components/caching/cache_checker.py b/haystack/components/caching/cache_checker.py index 2da90c84786..0d33e6cc2b1 100644 --- a/haystack/components/caching/cache_checker.py +++ b/haystack/components/caching/cache_checker.py @@ -2,6 +2,7 @@ # # SPDX-License-Identifier: Apache-2.0 +from datetime import datetime, timedelta from typing import Any from haystack import Document, component, default_from_dict, default_to_dict @@ -16,6 +17,11 @@ class CacheChecker: If matching documents are found, they are returned as "hits". If not found in the cache, the items are returned as "misses". + If `ttl` is provided, matching documents are considered cache hits only if their + timestamp is within the specified TTL. The timestamp is read from the `time_field` + metadata field, which defaults to `"cached_at"`. If `ttl` is `None`, documents are + considered cache hits based only on the presence of a matching `cache_field`. + ### Usage example ```python @@ -37,7 +43,13 @@ class CacheChecker: ``` """ - def __init__(self, document_store: DocumentStore, cache_field: str) -> None: + def __init__( + self, + document_store: DocumentStore, + cache_field: str, + ttl: timedelta | None = None, + time_field: str = "cached_at", + ) -> None: """ Creates a CacheChecker component. @@ -46,9 +58,17 @@ def __init__(self, document_store: DocumentStore, cache_field: str) -> None: :param cache_field: Name of the document's metadata field to check for cache hits. + :param ttl: + Maximum age of a cached document before it is considered expired. If `None`, + matching documents are considered cache hits regardless of their age. + :param time_field: + Name of the document's metadata field containing the cache timestamp. + Defaults to `"cached_at"`. """ self.document_store = document_store self.cache_field = cache_field + self.ttl = ttl + self.time_field = time_field def to_dict(self) -> dict[str, Any]: """ @@ -57,7 +77,15 @@ def to_dict(self) -> dict[str, Any]: :returns: Dictionary with serialized data. """ - return default_to_dict(self, document_store=self.document_store, cache_field=self.cache_field) + init_parameters = {"document_store": self.document_store, "cache_field": self.cache_field} + + if self.ttl is not None: + init_parameters["ttl"] = self.ttl + + if self.time_field != "cached_at": + init_parameters["time_field"] = self.time_field + + return default_to_dict(self, **init_parameters) @classmethod def from_dict(cls, data: dict[str, Any]) -> "CacheChecker": @@ -71,6 +99,18 @@ def from_dict(cls, data: dict[str, Any]) -> "CacheChecker": """ return default_from_dict(cls, data) + def _filter_valid_documents(self, documents: list[Document]) -> list[Document]: + if self.ttl is None: + return documents + + now = datetime.now() + + return [ + document + for document in documents + if self.time_field in document.meta and now - document.meta[self.time_field] < self.ttl + ] + @component.output_types(hits=list[Document], misses=list) def run(self, items: list[Any]) -> dict[str, Any]: """ @@ -89,8 +129,9 @@ def run(self, items: list[Any]) -> dict[str, Any]: for item in items: filters = {"field": self.cache_field, "operator": "==", "value": item} found = self.document_store.filter_documents(filters=filters) - if found: - found_documents.extend(found) + valid_documents = self._filter_valid_documents(found) + if valid_documents: + found_documents.extend(valid_documents) else: misses.append(item) return {"hits": found_documents, "misses": misses} @@ -116,8 +157,9 @@ async def run_async(self, items: list[Any]) -> dict[str, Any]: for item in items: filters = {"field": self.cache_field, "operator": "==", "value": item} found = await self.document_store.filter_documents_async(filters=filters) - if found: - found_documents.extend(found) + valid_documents = self._filter_valid_documents(found) + if valid_documents: + found_documents.extend(valid_documents) else: misses.append(item) return {"hits": found_documents, "misses": misses} diff --git a/releasenotes/notes/add-cache-checker-ttl-ae5a3c32c080cb89.yaml b/releasenotes/notes/add-cache-checker-ttl-ae5a3c32c080cb89.yaml new file mode 100644 index 00000000000..d0f6c5dcac3 --- /dev/null +++ b/releasenotes/notes/add-cache-checker-ttl-ae5a3c32c080cb89.yaml @@ -0,0 +1,4 @@ +--- +enhancements: + - | + Added optional TTL support to the ``CacheChecker`` component, allowing cached documents to expire after a configurable duration. The cache timestamp can be read from a custom metadata field using ``time_field``. diff --git a/test/components/caching/test_cache_checker.py b/test/components/caching/test_cache_checker.py index 4ff3ff5bec7..ed3c795fd2e 100644 --- a/test/components/caching/test_cache_checker.py +++ b/test/components/caching/test_cache_checker.py @@ -2,6 +2,7 @@ # # SPDX-License-Identifier: Apache-2.0 +from datetime import datetime, timedelta from unittest.mock import Mock, patch import pytest @@ -73,6 +74,30 @@ def test_from_dict_nonexisting_docstore(self): ): CacheChecker.from_dict(data) + def test_to_dict_with_ttl(self): + ttl = timedelta(hours=1) + checker = CacheChecker( + document_store=InMemoryDocumentStore(), cache_field="url", ttl=ttl, time_field="created_at" + ) + + result = checker.to_dict() + + assert result["init_parameters"]["ttl"] == ttl + assert result["init_parameters"]["time_field"] == "created_at" + + def test_from_dict_with_ttl(self): + ttl = timedelta(hours=1) + checker = CacheChecker( + document_store=InMemoryDocumentStore(), cache_field="url", ttl=ttl, time_field="created_at" + ) + + data = checker.to_dict() + reconstructed = CacheChecker.from_dict(data) + + assert reconstructed.cache_field == "url" + assert reconstructed.ttl == ttl + assert reconstructed.time_field == "created_at" + def test_run(self, in_memory_doc_store): documents = [ Document(content="doc1", meta={"url": "https://example.com/1"}), @@ -103,3 +128,53 @@ def test_close(self): checker = CacheChecker(document_store=nonclosable_document_store, cache_field="url") checker.close() assert nonclosable_document_store.mock_calls == [] + + def test_run_with_missing_time_field(self, in_memory_doc_store): + document = Document(content="doc1", meta={"url": "https://example.com/1"}) + + in_memory_doc_store.write_documents([document]) + checker = CacheChecker(in_memory_doc_store, cache_field="url", ttl=timedelta(hours=1)) + + results = checker.run(items=["https://example.com/1"]) + + assert results == {"hits": [], "misses": ["https://example.com/1"]} + + def test_run_with_fresh_document(self, in_memory_doc_store): + document = Document(content="doc1", meta={"url": "https://example.com/1", "cached_at": datetime.now()}) + + in_memory_doc_store.write_documents([document]) + checker = CacheChecker(in_memory_doc_store, cache_field="url", ttl=timedelta(hours=1)) + results = checker.run(items=["https://example.com/1"]) + + assert results == {"hits": [document], "misses": []} + + def test_run_with_expired_document(self, in_memory_doc_store): + document = Document( + content="doc1", meta={"url": "https://example.com/1", "cached_at": datetime.now() - timedelta(hours=2)} + ) + + in_memory_doc_store.write_documents([document]) + checker = CacheChecker(in_memory_doc_store, cache_field="url", ttl=timedelta(hours=1)) + results = checker.run(items=["https://example.com/1"]) + + assert results == {"hits": [], "misses": ["https://example.com/1"]} + + def test_run_with_custom_time_field(self, in_memory_doc_store): + document = Document(content="doc1", meta={"url": "https://example.com/1", "created_at": datetime.now()}) + + in_memory_doc_store.write_documents([document]) + checker = CacheChecker(in_memory_doc_store, cache_field="url", ttl=timedelta(hours=1), time_field="created_at") + results = checker.run(items=["https://example.com/1", "https://example.com/5"]) + + assert results == {"hits": [document], "misses": ["https://example.com/5"]} + + def test_run_with_custom_time_field_and_expired_document(self, in_memory_doc_store): + document = Document( + content="doc1", meta={"url": "https://example.com/1", "created_at": datetime.now() - timedelta(hours=2)} + ) + + in_memory_doc_store.write_documents([document]) + checker = CacheChecker(in_memory_doc_store, cache_field="url", ttl=timedelta(hours=1), time_field="created_at") + results = checker.run(items=["https://example.com/1"]) + + assert results == {"hits": [], "misses": ["https://example.com/1"]} diff --git a/test/components/caching/test_cache_checker_async.py b/test/components/caching/test_cache_checker_async.py index d33307af875..14034393628 100644 --- a/test/components/caching/test_cache_checker_async.py +++ b/test/components/caching/test_cache_checker_async.py @@ -2,6 +2,7 @@ # # SPDX-License-Identifier: Apache-2.0 +from datetime import datetime, timedelta from unittest.mock import AsyncMock, MagicMock, Mock import pytest @@ -73,3 +74,29 @@ async def test_close_async(self): checker = CacheChecker(document_store=nonclosable_document_store, cache_field="url") await checker.close_async() assert nonclosable_document_store.mock_calls == [] + + @pytest.mark.asyncio + async def test_run_async_with_fresh_document(self, in_memory_doc_store): + document = Document(content="doc1", meta={"url": "https://example.com/1", "cached_at": datetime.now()}) + in_memory_doc_store.write_documents([document]) + checker = CacheChecker(in_memory_doc_store, cache_field="url", ttl=timedelta(hours=1)) + results = await checker.run_async(items=["https://example.com/1"]) + assert results == {"hits": [document], "misses": []} + + @pytest.mark.asyncio + async def test_run_async_with_expired_document(self, in_memory_doc_store): + document = Document( + content="doc1", meta={"url": "https://example.com/1", "cached_at": datetime.now() - timedelta(hours=2)} + ) + in_memory_doc_store.write_documents([document]) + checker = CacheChecker(in_memory_doc_store, cache_field="url", ttl=timedelta(hours=1)) + results = await checker.run_async(items=["https://example.com/1"]) + assert results == {"hits": [], "misses": ["https://example.com/1"]} + + @pytest.mark.asyncio + async def test_run_async_with_missing_time_field(self, in_memory_doc_store): + document = Document(content="doc1", meta={"url": "https://example.com/1"}) + in_memory_doc_store.write_documents([document]) + checker = CacheChecker(in_memory_doc_store, cache_field="url", ttl=timedelta(hours=1)) + results = await checker.run_async(items=["https://example.com/1"]) + assert results == {"hits": [], "misses": ["https://example.com/1"]}