semantic_router.index.base

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.

The index is the single source of truth for routes. Each record is one (route, utterance) pair with a deterministic id, so adding the same pair again replaces the record instead of duplicating it.

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.

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 route names, one per embedding.
• utterances (List[str]): List of utterances, one per embedding.
• metadata_list (Optional[List[Dict[str, Any]]]): List of metadata dicts, one per embedding. Holds the route's own metadata and the reserved sr_description and sr_confidence_threshold keys.
• List[List[float]]2 (List[List[float]]3): List of sparse embeddings, one per embedding. Only hybrid indexes store them.

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 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 route names, one per embedding.
• utterances (List[str]): List of utterances, one per embedding.
• metadata_list (Optional[List[Dict[str, Any]]]): List of metadata dicts, one per embedding.
• sparse_embeddings (Optional[List[SparseEmbedding]]): List of sparse embeddings, one per embedding.

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, asynchronously.

Returns:

List[Utterance]: A list of Utterance objects.

get_routes

def get_routes() -> List[Route]

Rebuilds the routes stored in the index from its records.

Returns:

List[Route]: A list of Route objects.

aget_routes

async def aget_routes() -> List[Route]

Rebuilds the routes stored in the index from its records, asynchronously.

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.

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.

__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.

check_records

def check_records(
    embeddings: List[List[float]],
    routes: List[str],
    utterances: List[Any],
    metadata_list: Optional[List[Dict[str,
                                      Any]]] = None) -> List[Dict[str, Any]]

Checks the lists passed to add line up, and returns one metadata dict per

record (a copy, so an index never mutates the caller's dicts).

Arguments:

• embeddings (List[List[float]]): List of embeddings.
• routes (List[str]): List of route names.
• utterances (List[Any]): List of utterances.
• metadata_list (Optional[List[Dict[str, Any]]]): List of metadata dicts, or None for empty metadata.

Returns:

``0: One metadata dict per record.

to_utterance

def to_utterance(record: Dict[str, Any]) -> Utterance

Builds an Utterance from a record's metadata, which holds the sr_route and

sr_utterance keys next to the rest of the record's metadata.

Arguments:

• record (Dict[str, Any]): The record's metadata.

Returns:

Utterance: An Utterance object.

to_routes

def to_routes(utterances: List[Utterance]) -> List[Route]

Rebuilds routes from index records, in the order each route is first seen.

The route-level fields come from the reserved sr_ keys of the route's first record, and every sr_ key is stripped from the route's metadata.

Arguments:

• utterances (List[Utterance]): The records, as returned by get_utterances.

Returns:

List[Route]: A list of Route objects.