semantic_router.index.local

LocalIndex Objects

class LocalIndex(BaseIndex)

add

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

Upsert records into the index, one per (route, utterance) pair. A pair

that is already stored has its embedding and metadata replaced.

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.
• metadata_list (Optional[List[Dict[str, Any]]]): List of metadata to add to the index.
• sparse_embeddings (Optional[List[SparseEmbedding]]): Not supported, use HybridLocalIndex.

aadd

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

Upsert records into the index. Note that this just points to the sync add method as async makes no difference for the local computations of the LocalIndex.

get_utterances

def get_utterances() -> List[Utterance]

Gets every record stored in the index as an Utterance.

Returns:

List[Utterance]: A list of Utterance objects.

aget_utterances

async def aget_utterances() -> List[Utterance]

Gets every record stored in the index as an Utterance. Note that this just

points to the sync get_utterances method.

Returns:

List[Utterance]: A list of Utterance objects.

describe

def describe() -> IndexConfig

Describe the index.

Returns:

IndexConfig: An IndexConfig object.

is_ready

def is_ready() -> bool

Checks if the index is ready to be used.

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.

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 and return top_k results.

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 and return top_k results. Note that this

just points to the sync query method, the computation is local and CPU-bound.

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.

delete

def delete(route_name: str)

Delete all records of a specific route from the index.

Arguments:

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

adelete

async def adelete(route_name: str)

Delete all records of a specific route from the index. Note that this just points

to the sync delete method as async makes no difference for the local computations of the LocalIndex.

Arguments:

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

delete_index

def delete_index()

Deletes the index, effectively clearing it and setting it to None.

Returns:

None: None

delete_all

def delete_all()

Deletes all records from the index.

Returns:

None: None

adelete_index

async def adelete_index()

Deletes the index, effectively clearing it and setting it to None. Note that this just points

to the sync delete_index method as async makes no difference for the local computations of the LocalIndex.

Returns:

None: None