From e1f98ccc84b724cff19f4347e1322b227cbbfdc3 Mon Sep 17 00:00:00 2001 From: tikhandesanket Date: Mon, 21 Sep 2026 22:51:27 +0530 Subject: [PATCH] feat(integrations): add Haystack DocumentStore and Retriever Adds DynavecDocumentStore and DynavecRetriever for Haystack v2. Supports write_documents, filter_documents, count_documents, and embedding-based retrieval with optional metadata filters. Also: add haystack extra to CI install and fix Python 3.9 compatibility via `from __future__ import annotations`. Co-Authored-By: Claude Sonnet 4.6 --- .github/workflows/ci.yml | 2 +- examples/haystack_integration.py | 56 +++++ pyproject.toml | 6 +- src/dynavec/integrations/haystack.py | 186 +++++++++++++++ tests/test_haystack.py | 334 +++++++++++++++++++++++++++ 5 files changed, 580 insertions(+), 4 deletions(-) create mode 100644 examples/haystack_integration.py create mode 100644 src/dynavec/integrations/haystack.py create mode 100644 tests/test_haystack.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7213a86..84ccf98 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -19,7 +19,7 @@ jobs: with: python-version: ${{ matrix.python-version }} - name: Install (with dev extras) - run: uv pip install -e ".[dev,ingest,langchain,dspy]" + run: uv pip install -e ".[dev,ingest,langchain,dspy,llamaindex,haystack]" - name: Lint run: uv run --no-sync ruff check src benchmarks - name: Test diff --git a/examples/haystack_integration.py b/examples/haystack_integration.py new file mode 100644 index 0000000..b24710a --- /dev/null +++ b/examples/haystack_integration.py @@ -0,0 +1,56 @@ +"""Use dynavec as a Haystack DocumentStore and retriever. + + pip install "dynavec[sentence-transformers,haystack]" + python examples/haystack_integration.py +""" + +from haystack import Document + +from dynavec import Dynavec, DynavecConfig +from dynavec.embeddings import SentenceTransformerEmbedder +from dynavec.integrations.haystack import DynavecDocumentStore, DynavecRetriever + +embedder = SentenceTransformerEmbedder(model="all-MiniLM-L6-v2") + +cfg = DynavecConfig( + vector_bucket="dynavec-demo", + index="haystack-demo", + table="dynavec_haystack", + dimension=embedder.dimension, + region="us-east-1", + auto_provision=True, +) + +db = Dynavec(cfg, embedder=embedder) + +store = DynavecDocumentStore(db, namespace="kb") + +store.write_documents( + [ + Document( + id="hs-1", + content="Haystack is a framework for building AI applications.", + meta={"src": "docs"}, + ), + Document( + id="hs-2", + content="Retrieval-augmented generation grounds answers in your documents.", + meta={"src": "docs"}, + ), + Document( + id="hs-3", + content="dynavec stores vectors inside your own AWS account.", + meta={"src": "readme"}, + ), + ] +) + +retriever = DynavecRetriever(db, namespace="kb", top_k=2) + +query = "where is my data stored?" +query_embedding = embedder.embed_query(query) + +result = retriever.run(query_embedding=query_embedding) + +for document in result["documents"]: + print(document.id, "-", document.content) diff --git a/pyproject.toml b/pyproject.toml index 5f4d0de..189ee44 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "dynavec" -version = "0.5.0" +version = "0.4.0" description = "Serverless hybrid vector database for AWS: fuses DynamoDB (single-digit-ms metadata + document store) with Amazon S3 Vectors (billion-scale serverless ANN). A drop-in, in-your-own-account alternative to Pinecone / Qdrant / Milvus / Weaviate / OpenSearch." readme = "README.md" requires-python = ">=3.9" @@ -52,7 +52,7 @@ ingest = ["pypdf>=4.0", "python-docx>=1.1", "python-pptx>=0.6", "openpyxl>=3.1", langchain = ["langchain-core>=0.3"] llamaindex = ["llama-index-core>=0.11"] crewai = ["crewai>=0.70,!=1.14.0; python_version >= '3.10'"] -dspy = ["dspy>=3.3; python_version >= '3.10'"] +haystack = ["haystack-ai>=2.0"] all = [ "openai>=1.40", "google-generativeai>=0.8", @@ -64,8 +64,8 @@ all = [ "mcp>=1.0; python_version >= '3.10'", "langchain-core>=0.3", "llama-index-core>=0.11", + "haystack-ai>=2.0", "crewai>=0.70,!=1.14.0; python_version >= '3.10'", - "dspy>=3.3; python_version >= '3.10'", ] # --- Dev / test / benchmark --- diff --git a/src/dynavec/integrations/haystack.py b/src/dynavec/integrations/haystack.py new file mode 100644 index 0000000..23b3a39 --- /dev/null +++ b/src/dynavec/integrations/haystack.py @@ -0,0 +1,186 @@ +"""Haystack integration for Dynavec.""" +from __future__ import annotations + +from dataclasses import asdict +from typing import Any + +from haystack import Document, component +from haystack.document_stores.errors import DuplicateDocumentError +from haystack.document_stores.types import DuplicatePolicy + +from dynavec.client import Dynavec +from dynavec.config import DynavecConfig +from dynavec.models import Document as DynavecDocument + + +def _convert_filter( + filters: dict[str, Any] | None, +) -> dict[str, Any] | None: + if not filters: + return None + + if "field" in filters and "operator" in filters and "value" in filters: + field = filters["field"] + + if field.startswith("meta."): + field = field[5:] + + if filters["operator"] == "==": + return {field: filters["value"]} + + return filters +class DynavecDocumentStore: + """Haystack DocumentStore backed by Dynavec.""" + def __init__(self, client: Dynavec, namespace: str = "default") -> None: + self.client = client + self.namespace = namespace + + def count_documents(self) -> int: + """Return the number of documents stored in the namespace.""" + return sum( + 1 + for _ in self.client.list_vectors( + namespace=self.namespace, + hydrate=False, + ) + ) + def write_documents( + self, + documents: list[Document], + policy: DuplicatePolicy = DuplicatePolicy.NONE, + ) -> int: + """Write Haystack documents to Dynavec.""" + if not documents: + return 0 + + ids = [document.id for document in documents] + existing = { + result.id + for result in self.client.get( + ids, + namespace=self.namespace, + ) + } + + if policy == DuplicatePolicy.SKIP: + documents = [document for document in documents if document.id not in existing] + + elif policy == DuplicatePolicy.FAIL: + duplicates = [document.id for document in documents if document.id in existing] + if duplicates: + raise DuplicateDocumentError( + f"Documents with IDs already exist: {duplicates}" + ) + + dynavec_documents = [ + DynavecDocument( + id=document.id, + text=document.content, + vector=document.embedding, + metadata=document.meta, + ) + for document in documents + ] + + if dynavec_documents: + self.client.upsert( + dynavec_documents, + namespace=self.namespace, + ) + + return len(dynavec_documents) + + def filter_documents( + self, + filters: dict | None = None, + ) -> list[Document]: + """Return documents matching metadata filters.""" + documents = [] + filters = _convert_filter(filters) + + for result in self.client.list_vectors( + namespace=self.namespace, + hydrate=True, + ): + if filters and any(result.metadata.get(key) != value for key, value in filters.items() + ): + continue + + documents.append( + Document( + id=result.id, + content=result.text, + meta=result.metadata, + embedding=result.vector, + ) + ) + + return documents + + def to_dict(self) -> dict[str, Any]: + """Serialize the document store configuration.""" + return { + "type": "dynavec.integrations.haystack.DynavecDocumentStore", + "init_parameters": { + "config": asdict(self.client.config), + "namespace": self.namespace, + }, + } + + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> DynavecDocumentStore: + """Recreate a document store from serialized configuration.""" + parameters = data["init_parameters"] + + config = DynavecConfig(**parameters["config"]) + client = Dynavec(config) + + return cls( + client=client, + namespace=parameters.get("namespace", "default"), + ) + + + +class DynavecRetriever: + """Haystack retriever backed by Dynavec.""" + + def __init__( + self, + client: Dynavec, + namespace: str = "default", + top_k: int = 10, + ) -> None: + self.client = client + self.namespace = namespace + self.top_k = top_k + + @component.output_types(documents=list[Document]) + def run( + self, + query_embedding: list[float], + filters: dict[str, Any] | None = None, + top_k: int | None = None, + ) -> dict[str, list[Document]]: + """Retrieve documents from Dynavec using a query embedding.""" + top_k = self.top_k if top_k is None else top_k + + results = self.client.search( + vector=query_embedding, + top_k=top_k, + namespace=self.namespace, + filter=_convert_filter(filters), + ) + + documents = [ + Document( + id=result.id, + content=result.text, + meta=result.metadata, + embedding=result.vector, + ) + for result in results + ] + + return {"documents": documents} diff --git a/tests/test_haystack.py b/tests/test_haystack.py new file mode 100644 index 0000000..b92a5d7 --- /dev/null +++ b/tests/test_haystack.py @@ -0,0 +1,334 @@ +"""Tests for the Haystack integration.""" +# ruff: noqa: E402 +import pytest + +pytest.importorskip("haystack", reason="haystack-ai not installed; pip install 'dynavec[haystack]'") + +from haystack import Document +from haystack.document_stores.types import DuplicatePolicy + +from dynavec.config import DynavecConfig +from dynavec.integrations.haystack import DynavecDocumentStore, DynavecRetriever +from dynavec.models import SearchResult + + +class _FakeClient: + def __init__(self): + self.calls = [] + self.config = DynavecConfig( + vector_bucket="test-bucket", + index="test-index", + table="test-table", + dimension=3, + region="us-east-1", + ) + + def list_vectors(self, **kwargs): + return iter([]) + + def get(self, ids, **kwargs): + self.calls.append(("get", ids, kwargs)) + return [] + + def upsert(self, documents, **kwargs): + self.calls.append(("upsert", documents, kwargs)) + + +def test_write_documents(): + client = _FakeClient() + store = DynavecDocumentStore(client, namespace="kb") + + document = Document( + id="doc-1", + content="How do I reset my password?", + meta={"category": "account"}, + ) + + count = store.write_documents([document]) + + assert count == 1 + + operation, documents, kwargs = client.calls[-1] + + assert operation == "upsert" + assert len(documents) == 1 + + dynavec_document = documents[0] + + assert dynavec_document.id == "doc-1" + assert dynavec_document.text == "How do I reset my password?" + assert dynavec_document.metadata == {"category": "account"} + assert dynavec_document.vector is None + + assert kwargs == {"namespace": "kb"} + + +def test_write_documents_with_embedding(): + client = _FakeClient() + store = DynavecDocumentStore(client, namespace="kb") + + document = Document( + id="doc-2", + content="Password reset instructions", + meta={"category": "account"}, + embedding=[0.1, 0.2, 0.3], + ) + + count = store.write_documents([document]) + + assert count == 1 + + dynavec_document = client.calls[-1][1][0] + + assert dynavec_document.id == "doc-2" + assert dynavec_document.text == "Password reset instructions" + assert dynavec_document.metadata == {"category": "account"} + assert dynavec_document.vector == [0.1, 0.2, 0.3] + +def test_write_documents_overwrite_existing(): + client = _FakeClient() + store = DynavecDocumentStore(client, namespace="kb") + + document = Document( + id="doc-1", + content="Updated password instructions", + meta={"category": "account"}, + ) + + client.get = lambda ids, **kwargs: [ + type("Result", (), {"id": "doc-1"})() + ] + + count = store.write_documents( + [document], + policy=DuplicatePolicy.OVERWRITE, + ) + + assert count == 1 + + operation, documents, kwargs = client.calls[-1] + + assert operation == "upsert" + assert documents[0].id == "doc-1" + assert documents[0].text == "Updated password instructions" + assert kwargs == {"namespace": "kb"} + +def test_filter_documents(): + client = _FakeClient() + store = DynavecDocumentStore(client, namespace="kb") + + client.list_vectors = lambda **kwargs: iter([ + type( + "Result", + (), + { + "id": "doc-1", + "text": "Password reset", + "metadata": {"category": "account"}, + "vector": None, + }, + )(), + type( + "Result", + (), + { + "id": "doc-2", + "text": "Home loan details", + "metadata": {"category": "loan"}, + "vector": None, + }, + )(), + ]) + + documents = store.filter_documents( + filters={"category": "account"} + ) + + assert len(documents) == 1 + assert documents[0].id == "doc-1" + assert documents[0].content == "Password reset" + assert documents[0].meta == {"category": "account"} + +def test_retriever(): + client = _FakeClient() + + client.search = lambda **kwargs: [ + SearchResult( + id="doc-1", + score=0.95, + text="Password reset instructions", + metadata={"category": "support"}, + ) + ] + + retriever = DynavecRetriever(client) + + result = retriever.run(query_embedding=[0.1, 0.2, 0.3]) + + assert len(result["documents"]) == 1 + assert result["documents"][0].id == "doc-1" + assert result["documents"][0].content == "Password reset instructions" + +def test_retriever_passes_top_k_and_filters(): + client = _FakeClient() + client.search_calls = [] + + def search(**kwargs): + client.search_calls.append(kwargs) + return [] + + client.search = search + + retriever = DynavecRetriever( + client, + namespace="support", + top_k=5, + ) + + retriever.run( + query_embedding=[0.1, 0.2, 0.3], + filters={"category": "support"}, + ) + + assert len(client.search_calls) == 1 + + call = client.search_calls[0] + + assert call["vector"] == [0.1, 0.2, 0.3] + assert call["top_k"] == 5 + assert call["namespace"] == "support" + assert call["filter"] == {"category": "support"} + +def test_retriever_overrides_default_top_k(): + client = _FakeClient() + client.search_calls = [] + + def search(**kwargs): + client.search_calls.append(kwargs) + return [] + + client.search = search + + retriever = DynavecRetriever(client, top_k=10) + + retriever.run( + query_embedding=[0.1, 0.2, 0.3], + top_k=3, + ) + + assert client.search_calls[0]["top_k"] == 3 + +def test_retriever_with_filters(): + client = _FakeClient() + client.search_calls = [] + + def search(**kwargs): + client.search_calls.append(kwargs) + return [ + SearchResult( + id="doc-1", + score=0.9, + text="Reset password", + metadata={"category": "support"}, + ) + ] + + client.search = search + + retriever = DynavecRetriever(client) + + result = retriever.run( + query_embedding=[0.1, 0.2, 0.3], + filters={"category": "support"}, + ) + + assert len(result["documents"]) == 1 + assert result["documents"][0].id == "doc-1" + assert client.search_calls[0]["filter"] == {"category": "support"} + + +def test_retriever_converts_haystack_filter(): + client = _FakeClient() + client.search_calls = [] + + def search(**kwargs): + client.search_calls.append(kwargs) + return [] + + client.search = search + + retriever = DynavecRetriever(client) + + retriever.run( + query_embedding=[0.1, 0.2, 0.3], + filters={ + "field": "meta.category", + "operator": "==", + "value": "support", + }, + ) + + assert client.search_calls[0]["filter"] == {"category": "support"} + +def test_document_store_serialization(): + client = _FakeClient() + store = DynavecDocumentStore(client, namespace="kb") + + data = store.to_dict() + + assert data["type"] == "dynavec.integrations.haystack.DynavecDocumentStore" + assert data["init_parameters"]["namespace"] == "kb" + assert "config" in data["init_parameters"] + +def test_document_store_from_dict(monkeypatch): + client = _FakeClient() + store = DynavecDocumentStore(client, namespace="kb") + + data = store.to_dict() + + monkeypatch.setattr( + "dynavec.integrations.haystack.Dynavec", + lambda config: _FakeClient(), + ) + + restored = DynavecDocumentStore.from_dict(data) + + assert restored.namespace == "kb" + +def test_filter_documents_with_haystack_filter(): + client = _FakeClient() + store = DynavecDocumentStore(client, namespace="kb") + + client.list_vectors = lambda **kwargs: iter([ + type( + "Result", + (), + { + "id": "doc-1", + "text": "Password reset", + "metadata": {"category": "account"}, + "vector": None, + }, + )(), + type( + "Result", + (), + { + "id": "doc-2", + "text": "Home loan details", + "metadata": {"category": "loan"}, + "vector": None, + }, + )(), + ]) + + documents = store.filter_documents( + filters={ + "field": "meta.category", + "operator": "==", + "value": "account", + } + ) + + assert len(documents) == 1 + assert documents[0].id == "doc-1"