Source code for langchain.vectorstores.sqlitevss
from __future__ import annotations
import json
import logging
import warnings
from typing import (
    TYPE_CHECKING,
    Any,
    Iterable,
    List,
    Optional,
    Tuple,
    Type,
)
from langchain.docstore.document import Document
from langchain.schema.embeddings import Embeddings
from langchain.schema.vectorstore import VectorStore
if TYPE_CHECKING:
    import sqlite3
logger = logging.getLogger(__name__)
[docs]class SQLiteVSS(VectorStore):
    """Wrapper around SQLite with vss extension as a vector database.
    To use, you should have the ``sqlite-vss`` python package installed.
    Example:
        .. code-block:: python
            from langchain.vectorstores import SQLiteVSS
            from langchain.embeddings.openai import OpenAIEmbeddings
            ...
    """
[docs]    def __init__(
        self,
        table: str,
        connection: Optional[sqlite3.Connection],
        embedding: Embeddings,
        db_file: str = "vss.db",
    ):
        """Initialize with sqlite client with vss extension."""
        try:
            import sqlite_vss  # noqa  # pylint: disable=unused-import
        except ImportError:
            raise ImportError(
                "Could not import sqlite-vss python package. "
                "Please install it with `pip install sqlite-vss`."
            )
        if not connection:
            connection = self.create_connection(db_file)
        if not isinstance(embedding, Embeddings):
            warnings.warn("embeddings input must be Embeddings object.")
        self._connection = connection
        self._table = table
        self._embedding = embedding
        self.create_table_if_not_exists()