From d9bbce5c6c62cb8284b025e5da65063c0b27562e Mon Sep 17 00:00:00 2001 From: Keoseong Park Date: Tue, 21 Jul 2026 18:06:10 +0900 Subject: [PATCH] feat: support Haystack 3.x alongside 2.x Haystack 3.0 removed AsyncPipeline (Pipeline gained run_async) and moved the sentence-transformers embedders into their own package. The library code was already compatible: update tests, examples and README for both lines, and run CI against both. Review fixes bundled in: pack embeddings during write validation so the write path is typed end to end, chunk bulk deletes under SQLite's bind-parameter cap, mypy override for stub-less sqlite_vec, drop stale 2.x wording. --- .github/workflows/test.yml | 7 +++ CHANGELOG.md | 4 +- README.md | 16 ++++--- examples/async_pipeline.py | 15 ++++--- examples/embedding_retrieval.py | 23 +++++++--- pyproject.toml | 5 +++ .../retrievers/sqlite_vec/retriever.py | 2 +- .../sqlite_vec/document_store.py | 44 ++++++++++++------- tests/test_document_store.py | 10 +++++ tests/test_retriever_async.py | 7 ++- 10 files changed, 97 insertions(+), 36 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index eb38a63..eda0a25 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -26,6 +26,12 @@ jobs: fail-fast: false matrix: python-version: ["3.10", "3.11", "3.12", "3.13"] + haystack-version: ["3.*"] + include: + # the Haystack 2.x line is also supported — keep one job covering it + - python-version: "3.12" + haystack-version: "2.*" + name: Python ${{ matrix.python-version }} / haystack-ai ${{ matrix.haystack-version }} steps: - uses: actions/checkout@v4 - uses: actions/setup-python@v5 @@ -35,6 +41,7 @@ jobs: run: | python -m pip install --upgrade pip pip install -e . + pip install "haystack-ai==${{ matrix.haystack-version }}" pip install pytest pytest-asyncio - name: Test run: pytest -q diff --git a/CHANGELOG.md b/CHANGELOG.md index c10ea03..bc12697 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ the project follows [Semantic Versioning](https://semver.org/). First public release. ### Added +- Support for both the Haystack 2.x (≥2.27) and 3.x release lines; CI runs + the test suite and examples against both. - `SQLiteVecDocumentStore` implementing the Haystack `DocumentStore` protocol (`count_documents`, `filter_documents`, `write_documents`, `delete_documents`) on top of `sqlite-vec`'s `vec0` virtual table. @@ -35,6 +37,6 @@ First public release. - Examples in `examples/`: `quickstart.py`, `async_pipeline.py`, `embedding_retrieval.py`. - Benchmark in `scripts/bench_lock_overhead.py`. -- Test suite (149 tests) covering init validation, schema compatibility, +- Test suite (150 tests) covering init validation, schema compatibility, CRUD, retrieval, filtering, filter policies, async, concurrency, and serialization. diff --git a/README.md b/README.md index c320b29..9605f27 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ [![test](https://github.com/keosung/sqlite-vec-haystack/actions/workflows/test.yml/badge.svg)](https://github.com/keosung/sqlite-vec-haystack/actions/workflows/test.yml) [![Apache-2.0](https://img.shields.io/badge/license-Apache--2.0-blue.svg)](LICENSE) -A [Haystack](https://haystack.deepset.ai/) 2.x **DocumentStore + Retriever** backed by [`sqlite-vec`](https://github.com/asg017/sqlite-vec) — embedded vector search in a single SQLite file, cross-platform (Linux / macOS / Windows / Android / iOS / WASM). +A [Haystack](https://haystack.deepset.ai/) **DocumentStore + Retriever** backed by [`sqlite-vec`](https://github.com/asg017/sqlite-vec) — embedded vector search in a single SQLite file, cross-platform (Linux / macOS / Windows / Android / iOS / WASM). [Haystack](https://github.com/deepset-ai/haystack) is deepset's open-source framework for building LLM applications and RAG pipelines out of composable components. [sqlite-vec](https://github.com/asg017/sqlite-vec) is a pure-C SQLite extension that adds vector search to any SQLite database. This package connects the two. @@ -17,13 +17,13 @@ Most Haystack document stores are backed by a server process. This one is a sing pip install sqlite-vec-haystack ``` -This pulls in `haystack-ai>=2.27.0` and `sqlite-vec>=0.1.6` automatically. +This pulls in `haystack-ai>=2.27.0` and `sqlite-vec>=0.1.6` automatically. Both the Haystack 2.x (≥2.27) and 3.x release lines are supported; CI tests against both. ## At a glance ```python from haystack import Document, Pipeline -from haystack.components.embedders import ( +from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersDocumentEmbedder, SentenceTransformersTextEmbedder, ) @@ -58,6 +58,8 @@ result = pipeline.run({"text_embedder": {"text": "What is the capital of Germany print(result["retriever"]["documents"][0].content) ``` +The snippet uses the sentence-transformers embedders, which on Haystack 3.x live in the separate [`sentence-transformers-haystack`](https://pypi.org/project/sentence-transformers-haystack/) package; on Haystack 2.x they are bundled — import them from `haystack.components.embedders` instead. Any Haystack embedder works. + Retrieved documents carry the raw distance reported by sqlite-vec in `Document.score`, ordered ascending: **lower score means closer**. For `cosine`, `0.0` is an identical direction. The score is `None` when the metric is undefined for a pair (for example cosine against a zero vector). ## Metadata filtering @@ -93,14 +95,14 @@ The retriever also supports `filter_policy` (`"replace"` by default, or `"merge" ## Async API -Every store method has an `*_async` counterpart, and the retriever exposes `run_async` so it drops into `AsyncPipeline` without further wiring: +Every store method has an `*_async` counterpart, and the retriever exposes `run_async`, so async pipeline runs dispatch to it without further wiring: ```python import asyncio -from haystack import AsyncPipeline +from haystack import Pipeline retriever = SQLiteVecEmbeddingRetriever(document_store=store, top_k=5) -pipeline = AsyncPipeline() +pipeline = Pipeline() pipeline.add_component("retriever", retriever) async def main() -> None: @@ -112,6 +114,8 @@ async def main() -> None: asyncio.run(main()) ``` +On Haystack 2.x, plain `Pipeline` has no `run_async` — use `AsyncPipeline` there; the rest of the snippet is unchanged. + Under the hood the store dispatches each `*_async` call through `asyncio.to_thread` and serialises DB access with a `threading.RLock`. Concurrent `asyncio.gather` over the same store is safe; benchmarks (`scripts/bench_lock_overhead.py`) show the lock adds no measurable throughput cost. ## Embedded / on-device RAG diff --git a/examples/async_pipeline.py b/examples/async_pipeline.py index 66c31e2..0a77c57 100644 --- a/examples/async_pipeline.py +++ b/examples/async_pipeline.py @@ -2,12 +2,12 @@ # # SPDX-License-Identifier: Apache-2.0 -"""Run a SQLiteVec retriever inside a Haystack AsyncPipeline. +"""Run a SQLiteVec retriever inside an async Haystack pipeline. This demonstrates the async surface of the integration: - ``write_documents_async`` / ``filter_documents_async`` on the store -- ``SQLiteVecEmbeddingRetriever.run_async`` dispatched by ``AsyncPipeline`` +- ``SQLiteVecEmbeddingRetriever.run_async`` dispatched by the pipeline - thread-safe access from concurrent ``asyncio.gather`` calls python examples/async_pipeline.py @@ -19,7 +19,12 @@ from pathlib import Path from tempfile import TemporaryDirectory -from haystack import AsyncPipeline, Document +from haystack import Document + +try: # Haystack 2.x ships a dedicated AsyncPipeline + from haystack import AsyncPipeline +except ImportError: # Haystack 3.x folds run_async into Pipeline + from haystack import Pipeline as AsyncPipeline from haystack_integrations.components.retrievers.sqlite_vec import SQLiteVecEmbeddingRetriever from haystack_integrations.document_stores.sqlite_vec import SQLiteVecDocumentStore @@ -39,9 +44,9 @@ async def main() -> None: print(f"Concurrently wrote {sum(results)} docs across 3 batches.") print(f"Final count via async: {await store.count_documents_async()}") - # Wire the retriever into an AsyncPipeline. The `@component` decorator + # Wire the retriever into a pipeline. The `@component` decorator # detects `run_async` and sets `__haystack_supports_async__ = True`, - # so AsyncPipeline dispatches through that path automatically. + # so `pipeline.run_async` dispatches through that path automatically. retriever = SQLiteVecEmbeddingRetriever(document_store=store, top_k=3) pipeline = AsyncPipeline() pipeline.add_component("retriever", retriever) diff --git a/examples/embedding_retrieval.py b/examples/embedding_retrieval.py index 9c006c2..a8441d7 100644 --- a/examples/embedding_retrieval.py +++ b/examples/embedding_retrieval.py @@ -8,7 +8,12 @@ examples follow: a full indexing pipeline (embedder → writer) and a separate query pipeline (text embedder → retriever). -Install the optional dependency first: +Install the embedder dependency first. On Haystack 3.x the +sentence-transformers components live in their own package: + + pip install sentence-transformers-haystack + +On Haystack 2.x they are bundled, so this is enough: pip install "sentence-transformers>=2.2.0" @@ -26,10 +31,18 @@ from pathlib import Path from haystack import Document, Pipeline -from haystack.components.embedders import ( - SentenceTransformersDocumentEmbedder, - SentenceTransformersTextEmbedder, -) + +try: # Haystack 3.x: shipped separately as sentence-transformers-haystack + from haystack_integrations.components.embedders.sentence_transformers import ( + SentenceTransformersDocumentEmbedder, + SentenceTransformersTextEmbedder, + ) +except ImportError: # Haystack 2.x: bundled with haystack-ai + from haystack.components.embedders import ( + SentenceTransformersDocumentEmbedder, + SentenceTransformersTextEmbedder, + ) + from haystack.components.writers import DocumentWriter from haystack.document_stores.types import DuplicatePolicy diff --git a/pyproject.toml b/pyproject.toml index dee2646..9602868 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -69,6 +69,11 @@ non_interactive = true check_untyped_defs = true disallow_incomplete_defs = true +[[tool.mypy.overrides]] +# sqlite-vec is a thin C-extension wrapper without stubs or py.typed +module = "sqlite_vec" +ignore_missing_imports = true + [tool.ruff] line-length = 120 diff --git a/src/haystack_integrations/components/retrievers/sqlite_vec/retriever.py b/src/haystack_integrations/components/retrievers/sqlite_vec/retriever.py index ed86833..936466a 100644 --- a/src/haystack_integrations/components/retrievers/sqlite_vec/retriever.py +++ b/src/haystack_integrations/components/retrievers/sqlite_vec/retriever.py @@ -2,7 +2,7 @@ # # SPDX-License-Identifier: Apache-2.0 -"""SQLiteVecEmbeddingRetriever — Haystack 2.x retriever component.""" +"""SQLiteVecEmbeddingRetriever — Haystack retriever component.""" from __future__ import annotations diff --git a/src/haystack_integrations/document_stores/sqlite_vec/document_store.py b/src/haystack_integrations/document_stores/sqlite_vec/document_store.py index 3790d18..79494f9 100644 --- a/src/haystack_integrations/document_stores/sqlite_vec/document_store.py +++ b/src/haystack_integrations/document_stores/sqlite_vec/document_store.py @@ -3,7 +3,7 @@ # SPDX-License-Identifier: Apache-2.0 """ -SQLiteVecDocumentStore — Haystack 2.x DocumentStore backed by sqlite-vec. +SQLiteVecDocumentStore — Haystack DocumentStore backed by sqlite-vec. Schema ------ @@ -46,6 +46,11 @@ VALID_DISTANCE_METRICS = ("cosine", "l2", "dot") +# SQLite caps the number of bound parameters per statement +# (SQLITE_MAX_VARIABLE_NUMBER — 999 in older builds). Id lists are chunked +# to stay under the cap so bulk deletes work regardless of the build. +_MAX_PARAMS_PER_STATEMENT = 900 + _VEC_DIM_PATTERN = re.compile(r"FLOAT\[(\d+)\]", re.IGNORECASE) _VEC_METRIC_PATTERN = re.compile(r"distance_metric\s*=\s*(\w+)", re.IGNORECASE) @@ -234,6 +239,7 @@ def write_documents( if not all(isinstance(d, Document) for d in documents): msg = "documents must be a list of Document instances" raise ValueError(msg) + packed_embeddings: list[bytes] = [] for doc in documents: if doc.embedding is None: msg = f"Document {doc.id!r} has no embedding — embed before writing to SQLiteVecDocumentStore" @@ -241,19 +247,21 @@ def write_documents( if len(doc.embedding) != self.embedding_dim: msg = f"Document {doc.id!r} embedding dim {len(doc.embedding)} != store dim {self.embedding_dim}" raise ValueError(msg) + packed_embeddings.append(_pack(doc.embedding)) with self._lock: - return self._write_documents_locked(documents, policy) + return self._write_documents_locked(documents, packed_embeddings, policy) def _write_documents_locked( self, documents: list[Document], + packed_embeddings: list[bytes], policy: DuplicatePolicy, ) -> int: conn = self._connect() written = 0 with conn: - for doc in documents: + for doc, packed in zip(documents, packed_embeddings, strict=True): meta_json = json.dumps(doc.meta) if doc.meta else None if policy == DuplicatePolicy.OVERWRITE: @@ -290,7 +298,7 @@ def _write_documents_locked( conn.execute("DELETE FROM vec_items WHERE rowid = ?", (rowid,)) conn.execute( "INSERT INTO vec_items(rowid, embedding) VALUES (?, ?)", - (rowid, _pack(doc.embedding)), + (rowid, packed), ) written += 1 return written @@ -310,18 +318,20 @@ def delete_documents(self, document_ids: list[str]) -> None: def _delete_documents_locked(self, document_ids: list[str]) -> None: conn = self._connect() - with conn: - # Only ``?`` placeholders are interpolated; all values are bound. - placeholders = ",".join("?" * len(document_ids)) - rows = conn.execute( - f"SELECT id FROM documents WHERE doc_id IN ({placeholders})", # noqa: S608 - tuple(document_ids), - ).fetchall() - int_ids = [r[0] for r in rows] - if int_ids: - qs = ",".join("?" * len(int_ids)) - conn.execute(f"DELETE FROM vec_items WHERE rowid IN ({qs})", int_ids) # noqa: S608 - conn.execute(f"DELETE FROM documents WHERE id IN ({qs})", int_ids) # noqa: S608 + with conn: # one transaction across all chunks + for start in range(0, len(document_ids), _MAX_PARAMS_PER_STATEMENT): + chunk = document_ids[start : start + _MAX_PARAMS_PER_STATEMENT] + # Only ``?`` placeholders are interpolated; all values are bound. + placeholders = ",".join("?" * len(chunk)) + rows = conn.execute( + f"SELECT id FROM documents WHERE doc_id IN ({placeholders})", # noqa: S608 + tuple(chunk), + ).fetchall() + int_ids = [r[0] for r in rows] + if int_ids: + qs = ",".join("?" * len(int_ids)) + conn.execute(f"DELETE FROM vec_items WHERE rowid IN ({qs})", int_ids) # noqa: S608 + conn.execute(f"DELETE FROM documents WHERE id IN ({qs})", int_ids) # noqa: S608 # -- KNN query (used by SQLiteVecEmbeddingRetriever) -------------------- @@ -381,7 +391,7 @@ def _query_by_embedding( # -- Async API ---------------------------------------------------------- # SQLite I/O is blocking either way, so instead of adding an aiosqlite # dependency each ``*_async`` method runs its sync counterpart on the - # default thread pool, keeping ``AsyncPipeline`` callers off the event + # default thread pool, keeping async pipeline callers off the event # loop's critical path. async def count_documents_async(self) -> int: diff --git a/tests/test_document_store.py b/tests/test_document_store.py index 225884b..9b36bc9 100644 --- a/tests/test_document_store.py +++ b/tests/test_document_store.py @@ -12,6 +12,7 @@ from haystack.document_stores.types import DuplicatePolicy from haystack_integrations.document_stores.sqlite_vec import SQLiteVecDocumentStore +from haystack_integrations.document_stores.sqlite_vec import document_store as ds_module def _doc(id_: str, *, embedding: list[float] | None = None, meta: dict | None = None) -> Document: @@ -205,6 +206,15 @@ def test_delete_empty_list_is_noop(self, document_store: SQLiteVecDocumentStore) document_store.delete_documents([]) assert document_store.count_documents() == 1 + def test_delete_chunks_id_lists_beyond_param_limit( + self, document_store: SQLiteVecDocumentStore, monkeypatch: pytest.MonkeyPatch + ): + """Id lists longer than the per-statement bind limit are chunked.""" + monkeypatch.setattr(ds_module, "_MAX_PARAMS_PER_STATEMENT", 2) + document_store.write_documents([_doc(f"d{i}") for i in range(5)]) + document_store.delete_documents([f"d{i}" for i in range(5)]) + assert document_store.count_documents() == 0 + class TestFilterDocuments: def test_no_filters_returns_all(self, document_store: SQLiteVecDocumentStore): diff --git a/tests/test_retriever_async.py b/tests/test_retriever_async.py index 217dfb4..1adc5c2 100644 --- a/tests/test_retriever_async.py +++ b/tests/test_retriever_async.py @@ -3,7 +3,12 @@ # SPDX-License-Identifier: Apache-2.0 import pytest -from haystack import AsyncPipeline, Document +from haystack import Document + +try: # Haystack 2.x ships a dedicated AsyncPipeline + from haystack import AsyncPipeline +except ImportError: # Haystack 3.x folds run_async into Pipeline + from haystack import Pipeline as AsyncPipeline from haystack_integrations.components.retrievers.sqlite_vec import SQLiteVecEmbeddingRetriever from haystack_integrations.document_stores.sqlite_vec import SQLiteVecDocumentStore