Haystack integration#67

.github/workflows/ci.yml

@@ -19,6 +19,7 @@ jobs:
         with:
           python-version: ${{ matrix.python-version }}
       - name: Install (with dev extras)
+        run: uv pip install -e ".[dev,ingest,langchain,dspy,llamaindex,haystack]"
         run: uv pip install -e ".[dev,ingest,langchain,dspy,llamaindex]"
       - name: Lint
         run: uv run --no-sync ruff check src benchmarks

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)

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

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}