vector_store

zocrates.rag.vector_store

Hybrid vector store: Chroma (semantic) + SQLite FTS5 (BM25 keyword).

Architecture

• ChromaDB holds embeddings and chunk metadata for semantic (dense) search.
• An auxiliary SQLite database with an FTS5 virtual table holds the raw chunk text for fast full-text (BM25) keyword search.
• Uses FTS5 trigram tokenizer to preserve math expressions (LaTeX) and code.
• Results from both indices are combined with Reciprocal Rank Fusion (RRF) to produce a single ranked list of chunks.

Reranking hook

search() accepts an optional reranker callable so a cross-encoder or other reranker can be plugged in later without changing the call-site.

RETRIEVAL_FINAL_K module-attribute

RETRIEVAL_FINAL_K = int(os.getenv('RETRIEVAL_FINAL_K', '8'))

RETRIEVAL_TOP_K module-attribute

RETRIEVAL_TOP_K = int(os.getenv('RETRIEVAL_TOP_K', '20'))

_FTS_INIT_LOCK module-attribute

_FTS_INIT_LOCK = threading.Lock()

_FTS_SCHEMA module-attribute

_FTS_SCHEMA = '\nCREATE VIRTUAL TABLE IF NOT EXISTS chunks_fts USING fts5(\n    chunk_id,        -- "{item_key}::{chunk_index}"\n    item_key,\n    chunk_text,\n    tokenize = \'trigram\'\n);\n'

_INITIALIZED_FTS_DBS module-attribute

_INITIALIZED_FTS_DBS = set()

_META_SCHEMA module-attribute

_META_SCHEMA = "\nCREATE TABLE IF NOT EXISTS chunk_meta (\n    chunk_id        TEXT PRIMARY KEY,\n    item_key        TEXT NOT NULL,\n    item_id         INTEGER NOT NULL,\n    chunk_index     INTEGER NOT NULL,\n    chunk_type      TEXT NOT NULL DEFAULT 'text',\n    heading         TEXT,\n    parent_item_key TEXT,\n    image_path      TEXT,\n    caption_source  TEXT,\n    vlm_model       TEXT,\n    line_start      INTEGER,\n    line_end        INTEGER\n);\n"

logger module-attribute

logger = logging.getLogger(__name__)

ChunkResult dataclass

ChunkResult(text, heading_path, chunk_index, document_title='', chunk_type='text', image_path=None, caption_source=None, vlm_model=None, line_start=1, line_end=1)

A single text chunk produced from a Markdown document.

heading_str property

heading_str

Return the heading path as a ' > '-delimited breadcrumb string.

EmbedFn

Bases: Protocol

Callable protocol accepted by VectorStore.

SearchResult dataclass

SearchResult(chunk_text, item_key, item_id, heading_path, chunk_index, score, chunk_type='text', image_path=None, caption_source=None, vlm_model=None, raw_semantic_score=None, raw_keyword_rank=None, retrieval_method='hybrid', title='', creators=None, year=None, chunk_id='', line_start=None, line_end=None)

A single retrieved chunk with provenance information.

VectorStore

VectorStore(chroma_dir, fts_db_path, embedder)

Hybrid vector + keyword search index for Zotero document chunks. Uses versioned Chroma collections per embedding model/dimension.

Initialize the VectorStore.

Parameters:

Name	Type	Description	Default
chroma_dir	Path	Directory for ChromaDB persistent storage.	required
fts_db_path	Path	Filesystem path for the SQLite FTS5 database.	required
embedder	EmbedFn	Callable that maps a list of strings to a list of embedding vectors and exposes _model_name and dimension attributes.	required

Source code in zocrates/rag/vector_store.py

def __init__(
        self,
        chroma_dir: Path,
        fts_db_path: Path,
        embedder: EmbedFn,
) -> None:
    """Initialize the VectorStore.

    Args:
        chroma_dir: Directory for ChromaDB persistent storage.
        fts_db_path: Filesystem path for the SQLite FTS5 database.
        embedder: Callable that maps a list of strings to a list of
            embedding vectors and exposes ``_model_name`` and
            ``dimension`` attributes.
    """
    self._rw_lock = threading.RLock()
    self._embedder = embedder
    self._fts_path = fts_db_path
    self._client = chromadb.PersistentClient(path=str(chroma_dir))

    # Determine model name and target embedding dimension
    model_name = getattr(embedder, "_model_name", "default")
    target_dim = getattr(embedder, "dimension", 768)

    # Generate versioned collection name
    self.active_collection_name = get_collection_name(model_name, target_dim)
    logger.info(
        "Initializing VectorStore with active collection: %s (model=%s, dim=%s)",
        self.active_collection_name,
        model_name,
        target_dim,
    )

    self._collection: Collection = self._client.get_or_create_collection(
        name=self.active_collection_name,
        metadata={
            "hnsw:space": "cosine",
            "model_name": str(model_name),
            "dimension": str(target_dim),
        },
    )

fts_path property

fts_path

Return the path to the shared SQLite FTS database.

acquire_write_lock

acquire_write_lock()

Return the store's RLock for external write operations (e.g., model swap).

Returns:

Type	Description
Any	The threading.RLock guarding Chroma and FTS access.

Source code in zocrates/rag/vector_store.py

def acquire_write_lock(self) -> Any:
    """Return the store's RLock for external write operations (e.g., model swap).

    Returns:
        The ``threading.RLock`` guarding Chroma and FTS access.
    """
    return self._rw_lock

close

close()

Close the Chroma client and release underlying storage handles.

Source code in zocrates/rag/vector_store.py

def close(self) -> None:
    """Close the Chroma client and release underlying storage handles."""
    try:
        if hasattr(self._client, "close"):
            self._client.close()
    except Exception as exc:
        logger.warning("Error closing Chroma client: %s", exc)

delete_document

delete_document(item_key)

Remove all chunks for the given item_key from both Chroma and FTS5.

Parameters:

Name	Type	Description	Default
item_key	str	Zotero item key identifying the document to delete.	required

Source code in zocrates/rag/vector_store.py

def delete_document(self, item_key: str) -> None:
    """Remove all chunks for the given item_key from both Chroma and FTS5.

    Args:
        item_key: Zotero item key identifying the document to delete.
    """
    self._delete_item(item_key)

get_all_chunk_ids_for_item

get_all_chunk_ids_for_item(item_key)

Return all chunk IDs for an item key in chunk_index order.

Parameters:

Name	Type	Description	Default
item_key	str	Zotero item key.	required

Returns:

Type	Description
list[str]	List of chunk IDs.

Source code in zocrates/rag/vector_store.py

def get_all_chunk_ids_for_item(self, item_key: str) -> list[str]:
    """Return all chunk IDs for an item key in chunk_index order.

    Args:
        item_key: Zotero item key.

    Returns:
        List of chunk IDs.
    """
    try:
        with self._rw_lock:
            with _open_fts_db(self._fts_path) as conn:
                cursor = conn.execute(
                    "SELECT chunk_id FROM chunk_meta WHERE item_key = ? ORDER BY chunk_index ASC",
                    (item_key,),
                )
                return [row["chunk_id"] for row in cursor.fetchall()]
    except Exception as exc:
        logger.error("Failed to query all chunk IDs for %s: %s", item_key, exc)
        return []

get_all_ingested_item_keys

get_all_ingested_item_keys()

Return the set of all unique item_key strings currently ingested in the active collection.

Returns:

Type	Description
set[str]	Set of unique item_key strings.

Source code in zocrates/rag/vector_store.py

def get_all_ingested_item_keys(self) -> set[str]:
    """Return the set of all unique item_key strings currently ingested in the active collection.

    Returns:
        Set of unique item_key strings.
    """
    with self._rw_lock:
        result = self._collection.get(include=["metadatas"])
        metadatas = result.get("metadatas") or []
        return {m["item_key"] for m in metadatas if m and "item_key" in m}

get_all_item_keys

get_all_item_keys()

Return the set of all unique item keys indexed in the chunk_meta SQLite table.

Returns:

Type	Description
set[str]	Set of unique item_key strings.

Source code in zocrates/rag/vector_store.py

def get_all_item_keys(self) -> set[str]:
    """Return the set of all unique item keys indexed in the chunk_meta SQLite table.

    Returns:
        Set of unique item_key strings.
    """
    try:
        with _open_fts_db(self._fts_path) as conn:
            rows = conn.execute("SELECT DISTINCT item_key FROM chunk_meta").fetchall()
            return {row["item_key"] for row in rows if row["item_key"]}
    except Exception as exc:
        logger.error("Failed to query item keys from chunk_meta: %s", exc)
        return set()

get_intersecting_chunk_ids

get_intersecting_chunk_ids(item_key, start_line, end_line)

Return chunk IDs for an item key that intersect the given line range [start_line, end_line].

Parameters:

Name	Type	Description	Default
item_key	str	Zotero item key.	required
start_line	int	1-indexed start line of the range.	required
end_line	int	1-indexed end line of the range.	required

Returns:

Type	Description
list[str]	List of chunk IDs in chunk_index order.

Source code in zocrates/rag/vector_store.py

def get_intersecting_chunk_ids(
    self, item_key: str, start_line: int, end_line: int
) -> list[str]:
    """Return chunk IDs for an item key that intersect the given line range [start_line, end_line].

    Args:
        item_key: Zotero item key.
        start_line: 1-indexed start line of the range.
        end_line: 1-indexed end line of the range.

    Returns:
        List of chunk IDs in chunk_index order.
    """
    try:
        with self._rw_lock:
            with _open_fts_db(self._fts_path) as conn:
                cursor = conn.execute(
                    """
                    SELECT chunk_id FROM chunk_meta
                    WHERE item_key = ?
                      AND line_start IS NOT NULL
                      AND line_end IS NOT NULL
                      AND line_start <= ?
                      AND line_end >= ?
                    ORDER BY chunk_index ASC
                    """,
                    (item_key, end_line, start_line),
                )
                return [row["chunk_id"] for row in cursor.fetchall()]
    except Exception as exc:
        logger.error(
            "Failed to query intersecting chunks for %s [%d-%d]: %s",
            item_key,
            start_line,
            end_line,
            exc,
        )
        return []

ingest_document

ingest_document(item_key, item_id, chunks, extra_metadata=None)

Index all chunks from a single document into Chroma and FTS5.

Deletes any existing chunks for item_key before inserting, making the operation idempotent.

Parameters:

Name	Type	Description	Default
item_key	str	Unique Zotero item key used as the primary document ID.	required
item_id	int	Numeric Zotero item ID stored as chunk metadata.	required
chunks	list[ChunkResult]	Ordered list of ChunkResult objects to index.	required
extra_metadata	dict[str, Any] | None	Additional key/value pairs merged into every chunk's Chroma metadata dict.	None

Source code in zocrates/rag/vector_store.py

def ingest_document(
        self,
        item_key: str,
        item_id: int,
        chunks: list[ChunkResult],
        extra_metadata: dict[str, Any] | None = None,
) -> None:
    """
    Index all chunks from a single document into Chroma and FTS5.

    Deletes any existing chunks for ``item_key`` before inserting, making
    the operation idempotent.

    Args:
        item_key: Unique Zotero item key used as the primary document ID.
        item_id: Numeric Zotero item ID stored as chunk metadata.
        chunks: Ordered list of ``ChunkResult`` objects to index.
        extra_metadata: Additional key/value pairs merged into every
            chunk's Chroma metadata dict.
    """
    if not chunks:
        return

    with self._rw_lock:
        self._delete_item(item_key)

        texts = [c.text for c in chunks]
        embeddings = self._embedder(texts, task="retrieval", prompt_name="document")
        if embeddings and len(embeddings[0]) > 0:
            self._ensure_collection_dimension(len(embeddings[0]))

        parent_item_key = (extra_metadata or {}).get("parent_item_key")
        ids = [f"{item_key}::{c.chunk_index}" for c in chunks]
        metadatas = []
        for c in chunks:
            m = {
                "item_key": item_key,
                "parent_item_key": parent_item_key or "",
                "item_id": item_id,
                "chunk_index": c.chunk_index,
                "chunk_type": c.chunk_type,
                "heading": c.heading_str or "",
                "document_title": c.document_title or "",
                "chunk_id": f"{item_key}::{c.chunk_index}",
                "line_start": getattr(c, "line_start", 1),
                "line_end": getattr(c, "line_end", 1),
            }
            if c.image_path:
                m["image_path"] = c.image_path
            if c.caption_source:
                m["caption_source"] = c.caption_source
            if c.vlm_model:
                m["vlm_model"] = c.vlm_model
            if extra_metadata:
                for k, v in extra_metadata.items():
                    if v is not None:
                        m[k] = v
            metadatas.append(m)

        try:
            # Chroma batch upsert
            self._collection.upsert(
                ids=ids,
                embeddings=embeddings,
                documents=texts,
                metadatas=metadatas,
            )

            # FTS5 idempotent delete-then-insert
            with _open_fts_db(self._fts_path) as conn:
                conn.execute("DELETE FROM chunks_fts WHERE item_key = ?", (item_key,))
                conn.execute("DELETE FROM chunk_meta WHERE item_key = ?", (item_key,))
                conn.executemany(
                    "INSERT INTO chunks_fts (chunk_id, item_key, chunk_text) VALUES (?, ?, ?)",
                    [(cid, item_key, text) for cid, text in zip(ids, texts, strict=False)],
                )
                conn.executemany(
                    """
                    INSERT INTO chunk_meta (
                        chunk_id, item_key, parent_item_key, item_id,
                        chunk_index, chunk_type, heading, image_path,
                        caption_source, vlm_model, line_start, line_end
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    [
                        (
                            cid,
                            item_key,
                            parent_item_key,
                            item_id,
                            c.chunk_index,
                            c.chunk_type,
                            c.heading_str,
                            c.image_path,
                            c.caption_source,
                            c.vlm_model,
                            getattr(c, "line_start", 1),
                            getattr(c, "line_end", 1),
                        )
                        for cid, c in zip(ids, chunks, strict=False)
                    ],
                )
                conn.commit()

            logger.info("Ingested %d chunks for item_key=%s", len(chunks), item_key)
        finally:
            del texts, embeddings, ids, metadatas

is_ingested

is_ingested(item_key)

Check whether any chunks exist for item_key in the active collection.

Parameters:

Name	Type	Description	Default
item_key	str	Zotero item key to look up.	required

Returns:

Type	Description
bool	True if at least one chunk is indexed, False otherwise.

Source code in zocrates/rag/vector_store.py

def is_ingested(self, item_key: str) -> bool:
    """Check whether any chunks exist for item_key in the active collection.

    Args:
        item_key: Zotero item key to look up.

    Returns:
        ``True`` if at least one chunk is indexed, ``False`` otherwise.
    """
    with self._rw_lock:
        result = self._collection.get(where={"item_key": item_key}, limit=1)
        return bool(result["ids"])

list_collections

list_collections()

List all stored versioned Chroma collections with their metadata and chunk counts.

Returns:

Type	Description
list[dict[str, Any]]	List of dicts, each with keys name, model_name, dimension,
list[dict[str, Any]]	count, and is_active.

Source code in zocrates/rag/vector_store.py

def list_collections(self) -> list[dict[str, Any]]:
    """List all stored versioned Chroma collections with their metadata and chunk counts.

    Returns:
        List of dicts, each with keys ``name``, ``model_name``, ``dimension``,
        ``count``, and ``is_active``.
    """
    result: list[dict[str, Any]] = []
    try:
        raw_cols = self._client.list_collections()
        for col_entry in raw_cols:
            col_name = (
                col_entry
                if isinstance(col_entry, str)
                else getattr(col_entry, "name", str(col_entry))
            )
            try:
                if hasattr(col_entry, "count") and hasattr(col_entry, "metadata"):
                    col = col_entry
                else:
                    col = self._client.get_collection(name=col_name)
                count = col.count()
                meta = col.metadata or {}
                dim = meta.get("dimension")
                if not dim:
                    try:
                        sample = col.get(limit=1, include=["embeddings"])
                        emb = sample.get("embeddings") if sample else None
                        if emb is not None and len(emb) > 0:
                            dim = str(len(emb[0]))
                        else:
                            dim = "unknown"
                    except Exception:
                        dim = "unknown"
                result.append({
                    "name": col_name,
                    "model_name": meta.get("model_name", "unknown"),
                    "dimension": str(dim),
                    "count": count,
                    "is_active": (col_name == self.active_collection_name),
                })
            except Exception as inner_exc:
                logger.warning("Error inspecting collection %s: %s", col_name, inner_exc)
    except Exception as exc:
        logger.error("Failed to list collections: %s", exc)
    return result

reset_collection

reset_collection(target='active')

Reset/clear vector store collection(s).

Parameters:

Name	Type	Description	Default
target	str	Which collection(s) to reset: - "active": delete and recreate the active versioned collection. - "all": delete all Chroma collections and the FTS SQLite database. - <collection_name>: delete a specific named collection.	'active'

Returns:

Type	Description
dict[str, Any]	Dict with keys status, target, deleted_collections,
dict[str, Any]	fts_reset, and active_collection.

Source code in zocrates/rag/vector_store.py

def reset_collection(self, target: str = "active") -> dict[str, Any]:
    """
    Reset/clear vector store collection(s).

    Args:
        target: Which collection(s) to reset:
            - ``"active"``: delete and recreate the active versioned collection.
            - ``"all"``: delete all Chroma collections and the FTS SQLite database.
            - ``<collection_name>``: delete a specific named collection.

    Returns:
        Dict with keys ``status``, ``target``, ``deleted_collections``,
        ``fts_reset``, and ``active_collection``.
    """
    with self._rw_lock:
        deleted_collections: list[str] = []
        fts_reset = False

        if target == "all":
            try:
                raw_cols = self._client.list_collections()
                for c in raw_cols:
                    c_name = c if isinstance(c, str) else getattr(c, "name", str(c))
                    self._client.delete_collection(name=c_name)
                    deleted_collections.append(c_name)
            except Exception as exc:
                logger.error("Failed to delete all Chroma collections: %s", exc)

            if self._fts_path.exists():
                try:
                    self._fts_path.unlink()
                    fts_reset = True
                    logger.info("Unlinked shared SQLite FTS database: %s", self._fts_path)
                except Exception as exc:
                    logger.error("Failed to unlink FTS database: %s", exc)

            model_name = getattr(self._embedder, "_model_name", "default")
            target_dim = getattr(self._embedder, "dimension", 768)
            collection_metadata = {
                "hnsw:space": "cosine",
                "model_name": str(model_name),
                "dimension": str(target_dim),
            }

            self._collection = self._client.get_or_create_collection(
                name=self.active_collection_name,
                metadata=collection_metadata,
            )
        elif target == "active":
            try:
                self._client.delete_collection(name=self.active_collection_name)
                deleted_collections.append(self.active_collection_name)
            except Exception as exc:
                logger.warning("Failed to delete active collection %s: %s", self.active_collection_name, exc)

            model_name = getattr(self._embedder, "_model_name", "default")
            target_dim = getattr(self._embedder, "dimension", 768)
            collection_metadata = {
                "hnsw:space": "cosine",
                "model_name": str(model_name),
                "dimension": str(target_dim),
            }

            self._collection = self._client.get_or_create_collection(
                name=self.active_collection_name,
                metadata=collection_metadata,
            )
        else:
            try:
                self._client.delete_collection(name=target)
                deleted_collections.append(target)
                if target == self.active_collection_name:
                    model_name = getattr(self._embedder, "_model_name", "default")
                    target_dim = getattr(self._embedder, "dimension", 768)
                    collection_metadata = {
                        "hnsw:space": "cosine",
                        "model_name": str(model_name),
                        "dimension": str(target_dim),
                    }
                    self._collection = self._client.get_or_create_collection(
                        name=self.active_collection_name,
                        metadata=collection_metadata,
                    )
            except Exception as exc:
                logger.error("Failed to delete specific collection %s: %s", target, exc)

        return {
            "status": "success",
            "target": target,
            "deleted_collections": deleted_collections,
            "fts_reset": fts_reset,
            "active_collection": self.active_collection_name,
        }

search

search(query, top_k=RETRIEVAL_TOP_K, final_k=RETRIEVAL_FINAL_K, item_key_filter=None, chunk_type_filter=None, exclude_chunk_ids=None, reranker=None)

Hybrid (semantic + BM25) search with RRF fusion and raw score metrics.

Parameters:

Name	Type	Description	Default
query	str	Natural language query string.	required
top_k	int	Number of candidates retrieved from each index before fusion.	RETRIEVAL_TOP_K
final_k	int	Maximum number of results to return after fusion.	RETRIEVAL_FINAL_K
item_key_filter	list[str] | None	If provided, restrict search to these Zotero item keys.	None
chunk_type_filter	str | list[str] | None	Optional chunk type or list of chunk types to filter by.	None
exclude_chunk_ids	set[str] | list[str] | None	Optional set or list of chunk IDs to exclude from results.	None
reranker	Callable[[str, list[str]], list[float]] | None	Optional callable (query, texts) -> scores used to re-rank the fused results after retrieval.	None

Returns:

Type	Description
list[SearchResult]	List of SearchResult objects sorted by descending relevance score,
list[SearchResult]	capped at final_k entries.

Source code in zocrates/rag/vector_store.py

def search(
        self,
        query: str,
        top_k: int = RETRIEVAL_TOP_K,
        final_k: int = RETRIEVAL_FINAL_K,
        item_key_filter: list[str] | None = None,
        chunk_type_filter: str | list[str] | None = None,
        exclude_chunk_ids: set[str] | list[str] | None = None,
        reranker: Callable[[str, list[str]], list[float]] | None = None,
) -> list[SearchResult]:
    """
    Hybrid (semantic + BM25) search with RRF fusion and raw score metrics.

    Args:
        query: Natural language query string.
        top_k: Number of candidates retrieved from each index before fusion.
        final_k: Maximum number of results to return after fusion.
        item_key_filter: If provided, restrict search to these Zotero item keys.
        chunk_type_filter: Optional chunk type or list of chunk types to filter by.
        exclude_chunk_ids: Optional set or list of chunk IDs to exclude from results.
        reranker: Optional callable ``(query, texts) -> scores`` used to
            re-rank the fused results after retrieval.

    Returns:
        List of ``SearchResult`` objects sorted by descending relevance score,
        capped at ``final_k`` entries.
    """
    if not query or not query.strip():
        return []

    with self._rw_lock:
        semantic_results = self._semantic_search_with_scores(
            query, top_k, item_key_filter, chunk_type_filter, exclude_chunk_ids
        )
        keyword_results = self._keyword_search_with_ranks(
            query, top_k, item_key_filter, chunk_type_filter, exclude_chunk_ids
        )

    semantic_ids = [cid for cid, _ in semantic_results]
    keyword_ids = [cid for cid, _ in keyword_results]

    # Reciprocal Rank Fusion collapses duplicates natively
    fused = _rrf_fuse(semantic_ids, keyword_ids)[:final_k]

    if not fused:
        return []

    # Map details for explanation metrics
    semantic_scores = dict(semantic_results)
    keyword_ranks = dict(keyword_results)

    fused_ids = [cid for cid, _ in fused]
    chunk_data = self._fetch_chunks(fused_ids)

    results: list[SearchResult] = []
    for cid, score in fused:
        if cid not in chunk_data:
            continue
        data = chunk_data[cid]

        sem_score = semantic_scores.get(cid)
        kw_rank = keyword_ranks.get(cid)

        # Determine retrieval method
        if sem_score is not None and kw_rank is not None:
            method = "hybrid"
        elif sem_score is not None:
            method = "semantic"
        else:
            method = "keyword"

        results.append(
            SearchResult(
                chunk_text=data["text"],
                item_key=data["item_key"],
                item_id=data["item_id"],
                heading_path=data["heading"].split(" > ") if data["heading"] else [],
                chunk_index=data["chunk_index"],
                score=score,
                chunk_type=data.get("chunk_type", "text"),
                image_path=data.get("image_path"),
                caption_source=data.get("caption_source"),
                vlm_model=data.get("vlm_model"),
                raw_semantic_score=sem_score,
                raw_keyword_rank=kw_rank,
                retrieval_method=method,
                chunk_id=cid,
                line_start=data.get("line_start"),
                line_end=data.get("line_end"),
            )
        )

    # Optional reranking hook
    if reranker is not None:
        texts = [r.chunk_text for r in results]
        rerank_scores = reranker(query, texts)
        for r, s in zip(results, rerank_scores, strict=False):
            r.score = s
        results.sort(key=lambda r: r.score, reverse=True)

    return results

update_chunk_text_preserve_vector

update_chunk_text_preserve_vector(chunk_id, new_text)

Update stored document text in Chroma while explicitly preserving existing vector.

Passing existing embeddings explicitly bypasses Chroma's internal embedding function, guaranteeing zero ML overhead, zero network roundtrips, and strict dimension alignment.

Parameters:

Name	Type	Description	Default
chunk_id	str	Canonical chunk identifier (e.g. "{item_key}::0").	required
new_text	str	New document text for the chunk.	required

Returns:

Type	Description
bool	True if chunk was found, had embeddings, and was updated; False otherwise.

Source code in zocrates/rag/vector_store.py

def update_chunk_text_preserve_vector(self, chunk_id: str, new_text: str) -> bool:
    """Update stored document text in Chroma while explicitly preserving existing vector.

    Passing existing embeddings explicitly bypasses Chroma's internal embedding function,
    guaranteeing zero ML overhead, zero network roundtrips, and strict dimension alignment.

    Args:
        chunk_id: Canonical chunk identifier (e.g. ``"{item_key}::0"``).
        new_text: New document text for the chunk.

    Returns:
        True if chunk was found, had embeddings, and was updated; False otherwise.
    """
    if not chunk_id or new_text is None:
        return False

    with self._rw_lock:
        res = self._collection.get(ids=[chunk_id], include=["embeddings"])
        embs = res.get("embeddings")
        if embs is not None and len(embs) > 0 and len(embs[0]) > 0:
            self._collection.update(
                ids=[chunk_id],
                documents=[new_text],
                embeddings=[embs[0]],
            )
            logger.info("Updated chunk text preserving vector for chunk_id=%s", chunk_id)
            return True
        return False

update_metadata

update_metadata(item_key, metadata)

Update metadata for all chunks belonging to item_key without re-embedding.

Parameters:

Name	Type	Description	Default
item_key	str	Zotero item key.	required
metadata	dict[str, Any]	Key/value pairs to merge into existing chunk metadatas.	required

Returns:

Type	Description
bool	True if chunks were found and updated, False if item was not found.

Source code in zocrates/rag/vector_store.py

def update_metadata(self, item_key: str, metadata: dict[str, Any]) -> bool:
    """Update metadata for all chunks belonging to item_key without re-embedding.

    Args:
        item_key: Zotero item key.
        metadata: Key/value pairs to merge into existing chunk metadatas.

    Returns:
        True if chunks were found and updated, False if item was not found.
    """
    if not metadata:
        return False

    with self._rw_lock:
        res = self._collection.get(where={"item_key": item_key})
        ids = res.get("ids", [])
        metadatas = res.get("metadatas", [])

        if not ids:
            return False

        clean_metadata = {k: v for k, v in metadata.items() if v is not None}
        if not clean_metadata:
            return False

        updated_metadatas = []
        for existing_meta in metadatas:
            m = dict(existing_meta) if existing_meta else {}
            m.update(clean_metadata)
            clean_m = {k: v for k, v in m.items() if v is not None}
            updated_metadatas.append(clean_m)

        self._collection.update(ids=ids, metadatas=updated_metadatas)

        # Update chunk_meta if heading is supplied in metadata
        if "heading" in metadata:
            with _open_fts_db(self._fts_path) as conn:
                conn.execute(
                    "UPDATE chunk_meta SET heading = ? WHERE item_key = ?",
                    (metadata["heading"], item_key),
                )
                conn.commit()

        logger.info("Updated metadata for %d chunks of item_key=%s", len(ids), item_key)
        return True

_open_fts_db

_open_fts_db(path)

Open (or create) the FTS5 and chunk metadata SQLite databases.

Creates the chunks_fts FTS5 virtual table (trigram tokenizer) and the chunk_meta table if they do not already exist. Migrates missing columns on chunk_meta if opened on a legacy database.

Parameters:

Name	Type	Description	Default
path	Path	Filesystem path to the SQLite database file.	required

Returns:

Type	Description
Connection	An open sqlite3.Connection with row_factory set to
Connection	sqlite3.Row.

Source code in zocrates/rag/vector_store.py

def _open_fts_db(path: Path) -> sqlite3.Connection:
    """Open (or create) the FTS5 and chunk metadata SQLite databases.

    Creates the ``chunks_fts`` FTS5 virtual table (trigram tokenizer) and
    the ``chunk_meta`` table if they do not already exist. Migrates missing
    columns on ``chunk_meta`` if opened on a legacy database.

    Args:
        path: Filesystem path to the SQLite database file.

    Returns:
        An open ``sqlite3.Connection`` with ``row_factory`` set to
        ``sqlite3.Row``.
    """
    conn = sqlite3.connect(str(path), timeout=5.0)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL;")
    conn.execute("PRAGMA busy_timeout = 5000;")

    if path not in _INITIALIZED_FTS_DBS:
        with _FTS_INIT_LOCK:
            if path not in _INITIALIZED_FTS_DBS:
                conn.execute(_FTS_SCHEMA)
                conn.execute(_META_SCHEMA)

                # Check and migrate columns on chunk_meta
                existing_cols = {
                    row["name"] for row in conn.execute("PRAGMA table_info(chunk_meta)").fetchall()
                }
                for col_name in ("image_path", "caption_source", "vlm_model", "line_start", "line_end"):
                    if col_name not in existing_cols:
                        col_type = "INTEGER" if col_name in ("line_start", "line_end") else "TEXT"
                        conn.execute(f"ALTER TABLE chunk_meta ADD COLUMN {col_name} {col_type}")

                conn.commit()
                _INITIALIZED_FTS_DBS.add(path)

    return conn

_rrf_fuse

_rrf_fuse(semantic_ids, keyword_ids, k=60)

Combine two ranked lists of chunk IDs using Reciprocal Rank Fusion (RRF).

Parameters:

Name	Type	Description	Default
semantic_ids	list[str]	Chunk IDs ordered by descending semantic similarity.	required
keyword_ids	list[str]	Chunk IDs ordered by ascending BM25 rank (rank 1 = best).	required
k	int	RRF smoothing constant (higher reduces the impact of top-ranked items).	60

Returns:

Type	Description
list[tuple[str, float]]	List of (chunk_id, rrf_score) tuples sorted by descending score.

Source code in zocrates/rag/vector_store.py

def _rrf_fuse(
        semantic_ids: list[str],
        keyword_ids: list[str],
        k: int = 60,
) -> list[tuple[str, float]]:
    """
    Combine two ranked lists of chunk IDs using Reciprocal Rank Fusion (RRF).

    Args:
        semantic_ids: Chunk IDs ordered by descending semantic similarity.
        keyword_ids: Chunk IDs ordered by ascending BM25 rank (rank 1 = best).
        k: RRF smoothing constant (higher reduces the impact of top-ranked items).

    Returns:
        List of ``(chunk_id, rrf_score)`` tuples sorted by descending score.
    """
    scores: dict[str, float] = {}

    for rank, cid in enumerate(semantic_ids, start=1):
        scores[cid] = scores.get(cid, 0.0) + 1.0 / (k + rank)

    for rank, cid in enumerate(keyword_ids, start=1):
        scores[cid] = scores.get(cid, 0.0) + 1.0 / (k + rank)

    return sorted(scores.items(), key=lambda x: x[1], reverse=True)

get_collection_name

get_collection_name(model_name, dimension)

Generate a sanitized, versioned ChromaDB collection name from model and dimension.

Replaces any non-alphanumeric characters with underscores, collapses consecutive underscores, and appends the dimension suffix.

Example::

get_collection_name("jinaai/jina-embeddings-v5-text-nano", 768)
# -> 'zotero_library_jinaai_jina_embeddings_v5_text_nano_768d'

Parameters:

Name	Type	Description	Default
model_name	str	Hugging Face or Gemini model identifier.	required
dimension	int	Embedding vector dimension.	required

Returns:

Type	Description
str	A lowercase, underscore-delimited collection name string.

Source code in zocrates/config.py

def get_collection_name(model_name: str, dimension: int) -> str:
    """
    Generate a sanitized, versioned ChromaDB collection name from model and dimension.

    Replaces any non-alphanumeric characters with underscores, collapses
    consecutive underscores, and appends the dimension suffix.

    Example::

        get_collection_name("jinaai/jina-embeddings-v5-text-nano", 768)
        # -> 'zotero_library_jinaai_jina_embeddings_v5_text_nano_768d'

    Args:
        model_name: Hugging Face or Gemini model identifier.
        dimension: Embedding vector dimension.

    Returns:
        A lowercase, underscore-delimited collection name string.
    """
    sanitized_model = re.sub(r"[^a-zA-Z0-9_]", "_", model_name.strip())
    sanitized_model = re.sub(r"_+", "_", sanitized_model).strip("_")
    return f"zotero_library_{sanitized_model}_{dimension}d"

get_semantic_similarity_threshold

get_semantic_similarity_threshold(zotero_data_dir=None, cfg=None)

Return the minimum cosine similarity threshold for semantic vector retrieval.

Reads strictly from semantic_similarity_threshold in zocrates_config.json (clamped between 0.0 and 1.0, defaulting to 0.50).

Parameters:

Name	Type	Description	Default
zotero_data_dir	Path | None	Path to the Zotero data directory.	None
cfg	dict[str, Any] | None	Optional in-memory config dict. If provided, bypasses disk read.	None

Returns:

Type	Description
float	Float similarity threshold between 0.0 and 1.0.

Source code in zocrates/config.py

def get_semantic_similarity_threshold(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> float:
    """Return the minimum cosine similarity threshold for semantic vector retrieval.

    Reads strictly from ``semantic_similarity_threshold`` in ``zocrates_config.json``
    (clamped between 0.0 and 1.0, defaulting to ``0.50``).

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        Float similarity threshold between 0.0 and 1.0.
    """
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    try:
        val = float(cfg.get("semantic_similarity_threshold", DEFAULT_SEMANTIC_SIMILARITY_THRESHOLD))
        return max(0.0, min(1.0, val))
    except (ValueError, TypeError):
        return DEFAULT_SEMANTIC_SIMILARITY_THRESHOLD