semantic_router.index.redis

RedisIndex Objects

class RedisIndex(BaseIndex)

Redis implementation of Index, backed by RediSearch vector search.

Requires a Redis server with the RediSearch module (e.g. Redis Stack, or Redis 8+, which bundles it by default). Plain open-source Redis without the module cannot be used, since it has no vector index type.

add

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

Add records to the index.

Arguments:

• embeddings (List[List[float]]): The embeddings to add.
• routes (List[str]): The routes to add.
• utterances (List[str]): The utterances to add.
• function_schemas (Optional[List[Dict[str, Any]]]): The function schemas to add.
• metadata_list (List[Dict[str, Any]]): The metadata to add.
• List[List[float]]0 (List[List[float]]1): The batch size to use for the upload.

aadd

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

Asynchronously add records to the index.

Arguments:

• embeddings (List[List[float]]): The embeddings to add.
• routes (List[str]): The routes to add.
• utterances (List[str]): The utterances to add.
• function_schemas (Optional[List[Dict[str, Any]]]): The function schemas to add.
• metadata_list (List[Dict[str, Any]]): The metadata to add.
• List[List[float]]0 (List[List[float]]1): The batch size to use for the upload.

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]]

Query the index.

Arguments:

• vector (np.ndarray): The vector to query.
• top_k (int): The number of results to return.
• route_filter (Optional[List[str]]): The route filter to apply.
• sparse_vector (dict[int, float] | SparseEmbedding | None): Unused; RedisIndex does not support hybrid search.

Returns:

Tuple[np.ndarray, List[str]]: A tuple of the scores and 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]]

Asynchronously query the index.

Arguments:

• vector (np.ndarray): The vector to query.
• top_k (int): The number of results to return.
• route_filter (Optional[List[str]]): The route filter to apply.
• sparse_vector (dict[int, float] | SparseEmbedding | None): Unused; RedisIndex does not support hybrid search.

Returns:

Tuple[np.ndarray, List[str]]: A tuple of the scores and route names.

aget_routes

async def aget_routes() -> list[tuple]

Asynchronously get a list of route and utterance objects currently

stored in the index.

Returns:

list[tuple]: A list of (route_name, utterance, function_schemas, metadata) tuples.

delete

def delete(route_name: str)

Delete records from the index by route name.

Arguments:

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

adelete

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

Asynchronously delete records from the index by route name.

Arguments:

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

Returns:

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

delete_all

def delete_all()

Deletes all records from the index (but keeps the index/schema).

delete_index

def delete_index()

Delete the RediSearch index, all of its documents and its config (every namespace's sync hash and lock).

adelete_index

async def adelete_index()

Asynchronously delete the RediSearch index, all of its documents and its config (every namespace's sync hash and lock).

describe

def describe() -> IndexConfig

Describe the index.

Returns:

IndexConfig: The index configuration.

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.

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

Returns:

int: The total number of vectors.

close

def close()

Closes the Redis client connections if they exist.