​
BaseIndex Objects

class BaseIndex(BaseModel)

Base class for indices using Pydantic’s BaseModel. This class outlines the expected interface for index classes. Actual method implementations should be provided in subclasses.

​
add

def add(embeddings: List[List[float]],
        routes: List[str],
        utterances: List[Any],
        function_schemas: Optional[List[Dict[str, Any]]] = None,
        metadata_list: List[Dict[str, Any]] = [],
        **kwargs)

Add embeddings to the index.

This method should be implemented by subclasses.

Arguments:

  • embeddings (List[List[float]]): List of embeddings to add to the index.
  • routes (List[str]): List of routes to add to the index.
  • utterances (List[str]): List of utterances to add to the index.
  • function_schemas (Optional[List[Dict[str, Any]]]): List of function schemas to add to the index.
  • metadata_list (List[Dict[str, Any]]): List of metadata to add to the index.

​
aadd

async def aadd(embeddings: List[List[float]],
               routes: List[str],
               utterances: List[str],
               function_schemas: Optional[Optional[List[Dict[str,
                                                             Any]]]] = None,
               metadata_list: List[Dict[str, Any]] = [],
               **kwargs)

Add vectors to the index asynchronously.

This method should be implemented by subclasses.

Arguments:

  • embeddings (List[List[float]]): List of embeddings to add to the index.
  • routes (List[str]): List of routes to add to the index.
  • utterances (List[str]): List of utterances to add to the index.
  • function_schemas (Optional[List[Dict[str, Any]]]): List of function schemas to add to the index.
  • metadata_list (List[Dict[str, Any]]): List of metadata to add to the index.

​
get_utterances

def get_utterances(include_metadata: bool = False) -> List[Utterance]

Gets a list of route and utterance objects currently stored in the

index, including additional metadata.

Arguments:

  • include_metadata (bool): Whether to include function schemas and metadata in the returned Utterance objects.

Returns:

List[Utterance]: A list of Utterance objects.

​
aget_utterances

async def aget_utterances(include_metadata: bool = False) -> List[Utterance]

Gets a list of route and utterance objects currently stored in the

index, including additional metadata.

Arguments:

  • include_metadata (bool): Whether to include function schemas and metadata in the returned Utterance objects.

Returns:

List[Utterance]: A list of Utterance objects.

​
get_routes

def get_routes() -> List[Route]

Gets a list of route objects currently stored in the index.

Returns:

List[Route]: A list of Route objects.

​
delete

def delete(route_name: str)

Deletes route by route name.

This method should be implemented by subclasses.

Arguments:

  • route_name (str): Name of the route to delete.

​
adelete

async def adelete(route_name: str) -> list[str]

Asynchronously delete specified route from index if it exists. Returns the IDs

of the vectors deleted. This method should be implemented by subclasses.

Arguments:

  • route_name (str): Name of the route to delete.

Returns:

list[str]: List of IDs of the vectors deleted.

​
describe

def describe() -> IndexConfig

Returns an IndexConfig object with index details such as type, dimensions,

and total vector count. This method should be implemented by subclasses.

Returns:

IndexConfig: An IndexConfig object.

​
is_ready

def is_ready() -> bool

Checks if the index is ready to be used.

This method should be implemented by subclasses.

Returns:

bool: True if the index is ready, False otherwise.

​
ais_ready

async def ais_ready() -> bool

Checks if the index is ready to be used asynchronously.

This method should be implemented by subclasses.

Returns:

bool: True if the index is ready, False otherwise.

​
query

def query(
    vector: np.ndarray,
    top_k: int = 5,
    route_filter: Optional[List[str]] = None,
    sparse_vector: dict[int, float] | SparseEmbedding | None = None
) -> Tuple[np.ndarray, List[str]]

Search the index for the query_vector and return top_k results.

This method should be implemented by subclasses.

Arguments:

  • vector (np.ndarray): The vector to search for.
  • top_k (int): The number of results to return.
  • route_filter (Optional[List[str]]): The routes to filter the search by.
  • sparse_vector (dict[int, float] | SparseEmbedding | None): The sparse vector to search for.

Returns:

Tuple[np.ndarray, List[str]]: A tuple containing the query vector and a list of route names.

​
aquery

async def aquery(
    vector: np.ndarray,
    top_k: int = 5,
    route_filter: Optional[List[str]] = None,
    sparse_vector: dict[int, float] | SparseEmbedding | None = None
) -> Tuple[np.ndarray, List[str]]

Search the index for the query_vector and return top_k results.

This method should be implemented by subclasses.

Arguments:

  • vector (np.ndarray): The vector to search for.
  • top_k (int): The number of results to return.
  • route_filter (Optional[List[str]]): The routes to filter the search by.
  • sparse_vector (dict[int, float] | SparseEmbedding | None): The sparse vector to search for.

Returns:

Tuple[np.ndarray, List[str]]: A tuple containing the query vector and a list of route names.

​
aget_routes

def aget_routes()

Asynchronously get a list of route and utterance objects currently stored in the index.

This method should be implemented by subclasses.

Raises:

  • NotImplementedError: If the method is not implemented by the subclass.

Returns:

list[tuple]: A list of tuples, each containing a route name and an associated utterance.

​
delete_all

def delete_all()

Deletes all records from the index.

This method should be implemented by subclasses.

Raises:

  • NotImplementedError: If the method is not implemented by the subclass.

​
delete_index

def delete_index()

Deletes or resets the index.

This method should be implemented by subclasses.

Raises:

  • NotImplementedError: If the method is not implemented by the subclass.

​
adelete_index

async def adelete_index()

Deletes or resets the index asynchronously. This method should be implemented by subclasses.

​
lock

def lock(value: bool,
         wait: int = 0,
         scope: str | None = None) -> ConfigParameter

Lock/unlock the index for a given scope (if applicable). If index

already locked/unlocked, raises ValueError.

Arguments:

  • scope (str | None): The scope to lock.
  • wait (int): The number of seconds to wait for the index to be unlocked, if set to 0, will raise an error if index is already locked/unlocked.

Returns:

ConfigParameter: The config parameter that was locked.

​
alock

async def alock(value: bool,
                wait: int = 0,
                scope: str | None = None) -> ConfigParameter

Lock/unlock the index for a given scope (if applicable). If index already locked/unlocked, raises ValueError.

​
__len__

def __len__()

Returns the total number of vectors in the index. If the index is not initialized

returns 0.

Returns:

int: The total number of vectors.

​
alen

async def alen()

Async version of len. Returns the total number of vectors in the index.

Default implementation just calls the sync version.

Returns:

int: The total number of vectors.

​
parse_route_info

def parse_route_info(metadata: List[Dict[str, Any]]) -> List[Tuple]

Parses metadata from index to extract route, utterance, function

schema and additional metadata.

Arguments:

  • metadata (List[Dict[str, Any]]): List of metadata dictionaries.

Returns:

List[Tuple]: A list of tuples, each containing route, utterance, function schema and additional metadata.