Documentation Index
Fetch the complete documentation index at: https://docs.aurelio.ai/llms.txt
Use this file to discover all available pages before exploring further.
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
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
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
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
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__
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 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.