Vector Stores API

This module provides integrations for vector databases.

Base Vector Index

maticlib.vectorstores.base_index.BaseVectorIndex

BaseVectorIndex(embeddings)

Bases: ABC

Abstract base class for all vector store index backends.

Initializes the BaseVectorIndex.

Parameters:

Name	Type	Description	Default
embeddings	BaseEmbeddings	An embeddings provider matching BaseEmbeddings.	required

Source code in maticlib/vectorstores/base_index.py

def __init__(self, embeddings: BaseEmbeddings):
    """
    Initializes the BaseVectorIndex.

    Args:
        embeddings: An embeddings provider matching BaseEmbeddings.
    """
    self.embeddings = embeddings

add_segments abstractmethod

add_segments(segments)

Add a list of text segments to the index.

Parameters:

Name	Type	Description	Default
segments	List[TextSegment]	TextSegments to embed and store.	required

Source code in maticlib/vectorstores/base_index.py

@abstractmethod
def add_segments(self, segments: List[TextSegment]) -> None:
    """
    Add a list of text segments to the index.

    Args:
        segments: TextSegments to embed and store.
    """
    pass

delete abstractmethod

delete(segment_ids)

Delete segments from the index by their IDs.

Parameters:

Name	Type	Description	Default
segment_ids	List[str]	List of segment ID strings to remove.	required

Source code in maticlib/vectorstores/base_index.py

@abstractmethod
def delete(self, segment_ids: List[str]) -> None:
    """
    Delete segments from the index by their IDs.

    Args:
        segment_ids: List of segment ID strings to remove.
    """
    pass

similarity_search abstractmethod

similarity_search(query, k=4, filter_dict=None)

Search the index for the top k most similar segments to the query.

Parameters:

Name	Type	Description	Default
query	str	The natural language search query string.	required
k	int	Number of results to return.	4
filter_dict	Optional[Dict[str, Any]]	Optional metadata key/value filters.	None

Returns:

Type	Description
List[TextSegment]	A list of matching TextSegments.

Source code in maticlib/vectorstores/base_index.py

@abstractmethod
def similarity_search(
    self, query: str, k: int = 4, filter_dict: Optional[Dict[str, Any]] = None
) -> List[TextSegment]:
    """
    Search the index for the top k most similar segments to the query.

    Args:
        query: The natural language search query string.
        k: Number of results to return.
        filter_dict: Optional metadata key/value filters.

    Returns:
        A list of matching TextSegments.
    """
    pass

In-Memory Vector Index

maticlib.vectorstores.in_memory.InMemoryVectorIndex

InMemoryVectorIndex(embeddings)

Bases: BaseVectorIndex

Pure in-memory vector index using Numpy cosine similarity.

Suitable for rapid prototyping and small-scale workloads. Requires numpy.

Initializes the InMemoryVectorIndex.

Parameters:

Name	Type	Description	Default
embeddings	BaseEmbeddings	An embeddings provider matching BaseEmbeddings.	required

Raises:

Type	Description
MissingDependencyError	If numpy is not installed.

Source code in maticlib/vectorstores/in_memory.py

def __init__(self, embeddings: BaseEmbeddings):
    """
    Initializes the InMemoryVectorIndex.

    Args:
        embeddings: An embeddings provider matching BaseEmbeddings.

    Raises:
        MissingDependencyError: If numpy is not installed.
    """
    super().__init__(embeddings)
    self.segments: List[TextSegment] = []
    self.vectors: List[List[float]] = []

    try:
        import numpy as np
    except ImportError as e:
        raise MissingDependencyError(
            "numpy is required for InMemoryVectorIndex. Install it with: pip install maticlib[rag]"
        ) from e

Chroma Vector Index

maticlib.vectorstores.chroma.ChromaVectorIndex

ChromaVectorIndex(
    embeddings,
    collection_name="maticlib_collection",
    persist_directory=None,
    config=None,
)

Bases: BaseVectorIndex

Vector index backed by ChromaDB (ephemeral or persistent).

Initializes the ChromaVectorIndex.

Parameters:

Name	Type	Description	Default
embeddings	BaseEmbeddings	An embeddings provider matching BaseEmbeddings.	required
collection_name	str	Name of the Chroma collection. Default is maticlib_collection.	'maticlib_collection'
persist_directory	Optional[str]	If set, uses a PersistentClient saving to this directory.	None
config	Optional[VectorIndexConfig]	Optional VectorIndexConfig for distance metric settings.	None

Source code in maticlib/vectorstores/chroma.py

def __init__(
    self,
    embeddings: BaseEmbeddings,
    collection_name: str = "maticlib_collection",
    persist_directory: Optional[str] = None,
    config: Optional[VectorIndexConfig] = None,
):
    """
    Initializes the ChromaVectorIndex.

    Args:
        embeddings: An embeddings provider matching BaseEmbeddings.
        collection_name: Name of the Chroma collection. Default is ``maticlib_collection``.
        persist_directory: If set, uses a PersistentClient saving to this directory.
        config: Optional VectorIndexConfig for distance metric settings.
    """
    super().__init__(embeddings)
    self.collection_name = collection_name
    self.persist_directory = persist_directory
    self.config = config or VectorIndexConfig()

    try:
        import chromadb
        from chromadb.config import Settings
    except ImportError as e:
        raise MissingDependencyError(
            "chromadb is required for ChromaVectorIndex. Install it with: pip install maticlib[chroma]"
        ) from e

    if self.persist_directory:
        self.client = chromadb.PersistentClient(path=self.persist_directory)
    else:
        self.client = chromadb.EphemeralClient()

    # Chroma doesn't natively support HNSW config via standard create_collection easily without specific hnsw:space metadata
    metadata = {"hnsw:space": self.config.distance_metric}
    self.collection = self.client.get_or_create_collection(
        name=self.collection_name, metadata=metadata
    )

Milvus Vector Index

maticlib.vectorstores.milvus.MilvusVectorIndex

MilvusVectorIndex(
    embeddings,
    collection_name="maticlib_collection",
    uri="./milvus_demo.db",
    dim=None,
    config=None,
)

Bases: BaseVectorIndex

Vector index backed by Milvus (Lite file mode or standalone server).

Initializes the MilvusVectorIndex.

Parameters:

Name	Type	Description	Default
embeddings	BaseEmbeddings	An embeddings provider matching BaseEmbeddings.	required
collection_name	str	Name of the Milvus collection. Default is maticlib_collection.	'maticlib_collection'
uri	str	Milvus URI. Use a .db file path for Lite mode, or http://localhost:19530 for a standalone server.	'./milvus_demo.db'
dim	Optional[int]	Embedding dimension. Auto-detected if not provided.	None
config	Optional[VectorIndexConfig]	Optional VectorIndexConfig for distance metric and strategy.	None

Source code in maticlib/vectorstores/milvus.py

def __init__(
    self,
    embeddings: BaseEmbeddings,
    collection_name: str = "maticlib_collection",
    uri: str = "./milvus_demo.db",
    dim: Optional[int] = None,
    config: Optional[VectorIndexConfig] = None,
):
    """
    Initializes the MilvusVectorIndex.

    Args:
        embeddings: An embeddings provider matching BaseEmbeddings.
        collection_name: Name of the Milvus collection. Default is ``maticlib_collection``.
        uri: Milvus URI. Use a ``.db`` file path for Lite mode, or
            ``http://localhost:19530`` for a standalone server.
        dim: Embedding dimension. Auto-detected if not provided.
        config: Optional VectorIndexConfig for distance metric and strategy.
    """
    super().__init__(embeddings)
    self.collection_name = collection_name
    self.config = config or VectorIndexConfig()

    try:
        from pymilvus import MilvusClient
    except ImportError as e:
        raise MissingDependencyError(
            "pymilvus is required for MilvusVectorIndex. Install it with: pip install maticlib[milvus]"
        ) from e

    self.client = MilvusClient(uri=uri)

    # If dimension is not provided, we embed a dummy text to figure it out
    if not dim:
        dummy_res = self.embeddings.embed_query("test")
        dim = len(dummy_res.vector)

    if not self.client.has_collection(collection_name=self.collection_name):
        self.client.create_collection(
            collection_name=self.collection_name,
            dimension=dim,
            metric_type=(
                self.config.distance_metric.upper()
                if self.config.distance_metric in ["cosine", "l2", "ip"]
                else "L2"
            ),
        )

Pinecone Vector Index

maticlib.vectorstores.pinecone.PineconeVectorIndex

PineconeVectorIndex(
    embeddings, index_name, api_key=None, config=None
)

Bases: BaseVectorIndex

Vector index backed by Pinecone Cloud.

Initializes the PineconeVectorIndex.

Parameters:

Name	Type	Description	Default
embeddings	BaseEmbeddings	An embeddings provider matching BaseEmbeddings.	required
index_name	str	Name of an existing Pinecone index.	required
api_key	Optional[str]	Pinecone API key. Falls back to PINECONE_API_KEY env var.	None
config	Optional[VectorIndexConfig]	Optional VectorIndexConfig for distance metric and strategy.	None

Source code in maticlib/vectorstores/pinecone.py

def __init__(
    self,
    embeddings: BaseEmbeddings,
    index_name: str,
    api_key: Optional[str] = None,
    config: Optional[VectorIndexConfig] = None,
):
    """
    Initializes the PineconeVectorIndex.

    Args:
        embeddings: An embeddings provider matching BaseEmbeddings.
        index_name: Name of an existing Pinecone index.
        api_key: Pinecone API key. Falls back to ``PINECONE_API_KEY`` env var.
        config: Optional VectorIndexConfig for distance metric and strategy.
    """
    super().__init__(embeddings)
    self.index_name = index_name
    self.config = config or VectorIndexConfig()

    api_key = api_key or os.environ.get("PINECONE_API_KEY")
    if not api_key:
        raise ValueError(
            "PINECONE_API_KEY must be provided or set in environment variables."
        )

    try:
        from pinecone import Pinecone
    except ImportError as e:
        raise MissingDependencyError(
            "pinecone is required for PineconeVectorIndex. Install it with: pip install maticlib[pinecone]"
        ) from e

    self.pc = Pinecone(api_key=api_key)
    self.index = self.pc.Index(self.index_name)

Qdrant Vector Index

maticlib.vectorstores.qdrant.QdrantVectorIndex

QdrantVectorIndex(
    embeddings,
    collection_name="maticlib_collection",
    location=":memory:",
    url=None,
    api_key=None,
    dim=None,
    config=None,
)

Bases: BaseVectorIndex

Vector index backed by Qdrant (in-memory, local, or cloud).

Initializes the QdrantVectorIndex.

Parameters:

Name	Type	Description	Default
embeddings	BaseEmbeddings	An embeddings provider matching BaseEmbeddings.	required
collection_name	str	Name of the Qdrant collection. Default is maticlib_collection.	'maticlib_collection'
location	str	Qdrant storage location. Use ':memory:' for in-process, or a file path for local persistence.	':memory:'
url	Optional[str]	Optional URL to a remote Qdrant instance (e.g. http://localhost:6333).	None
api_key	Optional[str]	Optional API key for Qdrant Cloud.	None
dim	Optional[int]	Embedding dimension. Auto-detected if not provided.	None
config	Optional[VectorIndexConfig]	Optional VectorIndexConfig for distance metric and strategy.	None

Source code in maticlib/vectorstores/qdrant.py

def __init__(
    self,
    embeddings: BaseEmbeddings,
    collection_name: str = "maticlib_collection",
    location: str = ":memory:",
    url: Optional[str] = None,
    api_key: Optional[str] = None,
    dim: Optional[int] = None,
    config: Optional[VectorIndexConfig] = None,
):
    """
    Initializes the QdrantVectorIndex.

    Args:
        embeddings: An embeddings provider matching BaseEmbeddings.
        collection_name: Name of the Qdrant collection. Default is ``maticlib_collection``.
        location: Qdrant storage location. Use ``':memory:'`` for in-process,
            or a file path for local persistence.
        url: Optional URL to a remote Qdrant instance (e.g. ``http://localhost:6333``).
        api_key: Optional API key for Qdrant Cloud.
        dim: Embedding dimension. Auto-detected if not provided.
        config: Optional VectorIndexConfig for distance metric and strategy.
    """
    super().__init__(embeddings)
    self.collection_name = collection_name
    self.config = config or VectorIndexConfig()

    try:
        from qdrant_client import QdrantClient
        from qdrant_client.models import VectorParams, Distance
    except ImportError as e:
        raise MissingDependencyError(
            "qdrant-client is required for QdrantVectorIndex. Install it with: pip install maticlib[qdrant]"
        ) from e

    if url:
        self.client = QdrantClient(url=url, api_key=api_key)
    else:
        self.client = QdrantClient(location=location)

    # Map distance metric
    metric_map = {
        "cosine": Distance.COSINE,
        "l2": Distance.EUCLID,
        "ip": Distance.DOT,
    }
    distance = metric_map.get(self.config.distance_metric, Distance.COSINE)

    if not dim:
        dummy_res = self.embeddings.embed_query("test")
        dim = len(dummy_res.vector)

    if not self.client.collection_exists(collection_name=self.collection_name):
        self.client.create_collection(
            collection_name=self.collection_name,
            vectors_config=VectorParams(size=dim, distance=distance),
        )

Schema Vector Index

maticlib.vectorstores.schema_index.SchemaVectorIndex

SchemaVectorIndex(vector_index)

A specialized wrapper around a VectorIndex for storing and retrieving Database/Table schemas for Text2SQL workflows.

Source code in maticlib/vectorstores/schema_index.py

def __init__(self, vector_index: BaseVectorIndex):
    self.vector_index = vector_index

retrieve_relevant_tables

retrieve_relevant_tables(question, k=3)

Search the index for the most relevant tables given a natural language question. Returns a list of DDL strings.

Source code in maticlib/vectorstores/schema_index.py

def retrieve_relevant_tables(self, question: str, k: int = 3) -> List[str]:
    """
    Search the index for the most relevant tables given a natural language question.
    Returns a list of DDL strings.
    """
    results = self.vector_index.similarity_search(
        query=question, k=k, filter_dict={"type": "schema"}
    )
    return [res.content for res in results]