semantic_router.routers.base

is_valid

def is_valid(layer_config: str) -> bool

Make sure the given string is json format and contains the 3 keys: [“encoder_name”, “encoder_type”, “routes”]

RouterConfig Objects

class RouterConfig()

Generates a RouterConfig object that can be used for initializing routers.

init

def __init__(routes: List[Route] = [],
             encoder_type: str = "openai",
             encoder_name: Optional[str] = None)

Initialize a RouterConfig object. Arguments:

routes (List[Route]): A list of routes.
encoder_type (str): The type of encoder to use.
encoder_name (Optional[str]): The name of the encoder to use.

from_file

@classmethod
def from_file(cls, path: str) -> "RouterConfig"

Initialize a RouterConfig from a file. Expects a JSON or YAML file with file extension .json, .yaml, or .yml. Arguments:

path (str): The path to the file to load the RouterConfig from.

from_tuples

@classmethod
def from_tuples(cls,
                route_tuples: List[Tuple[str, str, Optional[List[Dict[str,
                                                                      Any]]],
                                         Dict[str, Any]]],
                encoder_type: str = "openai",
                encoder_name: Optional[str] = None)

Initialize a RouterConfig from a list of tuples of routes and utterances. Arguments:

route_tuples (List[Tuple[str, str]]): A list of tuples, each containing a route name and an associated utterance.
encoder_type (str, optional): The type of encoder to use, defaults to “openai”.
encoder_name (Optional[str], optional): The name of the encoder to use, defaults to None.

from_index

@classmethod
def from_index(cls,
               index: BaseIndex,
               encoder_type: str = "openai",
               encoder_name: Optional[str] = None)

Initialize a RouterConfig from a BaseIndex object. Arguments:

index (BaseIndex): The index to initialize the RouterConfig from.
encoder_type (str, optional): The type of encoder to use, defaults to “openai”.
encoder_name (Optional[str], optional): The name of the encoder to use, defaults to None.

to_dict

def to_dict() -> Dict[str, Any]

Convert the RouterConfig to a dictionary. Returns: Dict[str, Any]: A dictionary representation of the RouterConfig.

to_file

def to_file(path: str)

Save the routes to a file in JSON or YAML format. Arguments:

path (str): The path to save the RouterConfig to.

to_utterances

def to_utterances() -> List[Utterance]

Convert the routes to a list of Utterance objects. Returns: List[Utterance]: A list of Utterance objects.

add

def add(route: Route)

Add a route to the RouterConfig. Arguments:

route (Route): The route to add.

get

def get(name: str) -> Optional[Route]

Get a route from the RouterConfig by name. Arguments:

name (str): The name of the route to get.

Returns: Optional[Route]: The route if found, otherwise None.

remove

def remove(name: str)

Remove a route from the RouterConfig by name. Arguments:

name (str): The name of the route to remove.

get_hash

def get_hash() -> ConfigParameter

Get the hash of the RouterConfig. Used for syncing. Returns: ConfigParameter: The hash of the RouterConfig.

xq_reshape

def xq_reshape(xq: List[float] | np.ndarray) -> np.ndarray

Reshape the query vector to be a 2D numpy array. Arguments:

xq (List[float] | np.ndarray): The query vector.

Returns: np.ndarray: The reshaped query vector.

BaseRouter Objects

class BaseRouter(BaseModel)

Base class for all routers.

init

def __init__(encoder: Optional[DenseEncoder] = None,
             sparse_encoder: Optional[SparseEncoder] = None,
             llm: Optional[BaseLLM] = None,
             routes: Optional[List[Route]] = None,
             index: Optional[BaseIndex] = None,
             top_k: int = 5,
             aggregation: str = "mean",
             auto_sync: Optional[str] = None,
             init_async_index: bool = False)

Initialize a BaseRouter object. Expected to be used as a base class only, not directly instantiated. Arguments:

encoder (Optional[DenseEncoder]): The encoder to use.
sparse_encoder (Optional[SparseEncoder]): The sparse encoder to use.
llm (Optional[BaseLLM]): The LLM to use.
routes (Optional[List[Route]]): The routes to use.
index (Optional[BaseIndex]): The index to use.
Optional[DenseEncoder]0 (Optional[DenseEncoder]1): The number of routes to return.
Optional[DenseEncoder]2 (Optional[DenseEncoder]3): The aggregation method to use.
Optional[DenseEncoder]4 (Optional[DenseEncoder]5): The auto sync mode to use.

check_for_matching_routes

def check_for_matching_routes(top_class: str) -> Optional[Route]

Check for a matching route in the routes list. Arguments:

top_class (str): The top class to check for.

Returns: Optional[Route]: The matching route if found, otherwise None.

call

def __call__(text: Optional[str] = None,
             vector: Optional[List[float] | np.ndarray] = None,
             simulate_static: bool = False,
             route_filter: Optional[List[str]] = None,
             limit: int | None = 1) -> RouteChoice | List[RouteChoice]

Call the router to get a route choice. Arguments:

text (Optional[str]): The text to route.
vector (Optional[List[float] | np.ndarray]): The vector to route.
simulate_static (bool): Whether to simulate a static route.
route_filter (Optional[List[str]]): The route filter to use.
limit (int | None): The number of routes to return, defaults to 1. If set to None, no limit is applied and all routes are returned.

Returns: Optional[str]0: The route choice.

acall

async def acall(
    text: Optional[str] = None,
    vector: Optional[List[float] | np.ndarray] = None,
    limit: int | None = 1,
    simulate_static: bool = False,
    route_filter: Optional[List[str]] = None
) -> RouteChoice | list[RouteChoice]

Asynchronously call the router to get a route choice. Arguments:

text (Optional[str]): The text to route.
vector (Optional[List[float] | np.ndarray]): The vector to route.
simulate_static (bool): Whether to simulate a static route (ie avoid dynamic route LLM calls during fit or evaluate).
route_filter (Optional[List[str]]): The route filter to use.

Returns: RouteChoice: The route choice.

sync

def sync(sync_mode: str, force: bool = False, wait: int = 0) -> List[str]

Runs a sync of the local routes with the remote index. Arguments:

sync_mode (str): The mode to sync the routes with the remote index.
force (bool, optional): Whether to force the sync even if the local and remote hashes already match. Defaults to False.
wait (int): The number of seconds to wait for the index to be unlocked before proceeding with the sync. If set to 0, will raise an error if index is already locked/unlocked.

Returns: List[str]: A list of diffs describing the addressed differences between the local and remote route layers.

async_sync

async def async_sync(sync_mode: str,
                     force: bool = False,
                     wait: int = 0) -> List[str]

Runs a sync of the local routes with the remote index. Arguments:

sync_mode (str): The mode to sync the routes with the remote index.
force (bool, optional): Whether to force the sync even if the local and remote hashes already match. Defaults to False.
wait (int): The number of seconds to wait for the index to be unlocked before proceeding with the sync. If set to 0, will raise an error if index is already locked/unlocked.

Returns: List[str]: A list of diffs describing the addressed differences between the local and remote route layers.

from_json

@classmethod
def from_json(cls, file_path: str)

Load a RouterConfig from a JSON file. Arguments:

file_path (str): The path to the JSON file.

Returns: RouterConfig: The RouterConfig object.

from_yaml

@classmethod
def from_yaml(cls, file_path: str)

Load a RouterConfig from a YAML file. Arguments:

file_path (str): The path to the YAML file.

Returns: RouterConfig: The RouterConfig object.

from_config

@classmethod
def from_config(cls, config: RouterConfig, index: Optional[BaseIndex] = None)

Create a Router from a RouterConfig object. Arguments:

config (RouterConfig): The RouterConfig object.
index (Optional[BaseIndex]): The index to use.

add

def add(routes: List[Route] | Route)

Add a route to the local SemanticRouter and index. Arguments:

route (Route): The route to add.

aadd

async def aadd(routes: List[Route] | Route)

Add a route to the local SemanticRouter and index asynchronously. Arguments:

route (Route): The route to add.

update

def update(name: str,
           threshold: Optional[float] = None,
           utterances: Optional[List[str]] = None)

Updates the route specified in name. Allows the update of threshold and/or utterances. If no values are provided via the threshold or utterances parameters, those fields are not updated. If neither field is provided raises a ValueError. The name must exist within the local SemanticRouter, if not a KeyError will be raised. Arguments:

name (str): The name of the route to update.
threshold (Optional[float]): The threshold to update.
utterances (Optional[List[str]]): The utterances to update.

delete

def delete(route_name: str)

Deletes a route given a specific route name. Arguments:

route_name: the name of the route to be deleted

adelete

async def adelete(route_name: str)

Deletes a route given a specific route name asynchronously. Arguments:

route_name: the name of the route to be deleted

is_synced

def is_synced() -> bool

Check if the local and remote route layer instances are synchronized. Returns: bool: True if the local and remote route layers are synchronized, False otherwise.

async_is_synced

async def async_is_synced() -> bool

Check if the local and remote route layer instances are synchronized asynchronously. Returns: bool: True if the local and remote route layers are synchronized, False otherwise.

get_utterance_diff

def get_utterance_diff(include_metadata: bool = False) -> List[str]

Get the difference between the local and remote utterances. Returns a list of strings showing what is different in the remote when compared to the local. For example: [” route1: utterance1”, ” route1: utterance2”, ”- route2: utterance3”, ”- route2: utterance4”] Tells us that the remote is missing “route2: utterance3” and “route2: utterance4”, which do exist locally. If we see: [” route1: utterance1”, ” route1: utterance2”, ”+ route2: utterance3”, ”+ route2: utterance4”] This diff tells us that the remote has “route2: utterance3” and “route2: utterance4”, which do not exist locally. Arguments:

include_metadata (bool): Whether to include metadata in the diff.

Returns: List[str]: A list of strings showing the difference between the local and remote utterances.

aget_utterance_diff

async def aget_utterance_diff(include_metadata: bool = False) -> List[str]

Get the difference between the local and remote utterances asynchronously. Returns a list of strings showing what is different in the remote when compared to the local. For example: [” route1: utterance1”, ” route1: utterance2”, ”- route2: utterance3”, ”- route2: utterance4”] Tells us that the remote is missing “route2: utterance3” and “route2: utterance4”, which do exist locally. If we see: [” route1: utterance1”, ” route1: utterance2”, ”+ route2: utterance3”, ”+ route2: utterance4”] This diff tells us that the remote has “route2: utterance3” and “route2: utterance4”, which do not exist locally. Arguments:

include_metadata (bool): Whether to include metadata in the diff.

Returns: List[str]: A list of strings showing the difference between the local and remote utterances.

get

def get(name: str) -> Optional[Route]

Get a route by name. Arguments:

name (str): The name of the route to get.

Returns: Optional[Route]: The route.

group_scores_by_class

def group_scores_by_class(query_results: List[Dict]) -> Dict[str, List[float]]

Group the scores by class. Arguments:

query_results (List[Dict]): The query results to group. Expected format is a list of dictionaries with “route” and “score” keys.

Returns: Dict[str, List[float]]: A dictionary of route names and their associated scores.

set_threshold

def set_threshold(threshold: float, route_name: str | None = None)

Set the score threshold for a specific route or all routes. A threshold of 0.0 will mean that the route will be returned no matter how low it scores whereas a threshold of 1.0 will mean that a route must contain an exact utterance match to be returned. Arguments:

threshold (float): The threshold to set.
route_name (str | None): The name of the route to set the threshold for. If None, the threshold will be set for all routes.

to_config

def to_config() -> RouterConfig

Convert the router to a RouterConfig object. Returns: RouterConfig: The RouterConfig object.

to_json

def to_json(file_path: str)

Convert the router to a JSON file. Arguments:

file_path (str): The path to the JSON file.

to_yaml

def to_yaml(file_path: str)

Convert the router to a YAML file. Arguments:

file_path (str): The path to the YAML file.

get_thresholds

def get_thresholds() -> Dict[str, float]

Get the score thresholds for each route. Returns: Dict[str, float]: A dictionary of route names and their associated thresholds.

fit

def fit(X: List[str],
        y: List[str],
        batch_size: int = 500,
        max_iter: int = 500,
        local_execution: bool = False)

Fit the router to the data. Works best with a large number of examples for each route and with many None utterances. Arguments:

X (List[str]): The input data.
y (List[str]): The output data.
batch_size (int): The batch size to use for fitting.
max_iter (int): The maximum number of iterations to use for fitting.
local_execution (X0): Whether to execute the fitting locally.

evaluate

def evaluate(X: List[str], y: List[str], batch_size: int = 500) -> float

Evaluate the accuracy of the route selection. Arguments:

X (List[str]): The input data.
y (List[str]): The output data.
batch_size (int): The batch size to use for evaluation.

Returns: float: The accuracy of the route selection.

threshold_random_search

def threshold_random_search(
        route_layer: BaseRouter,
        search_range: Union[int, float]) -> Dict[str, float]

Performs a random search iteration given a route layer and a search range. Arguments:

route_layer (BaseRouter): The route layer to search.
search_range (Union[int, float]): The search range to use.

Returns: Dict[str, float]: A dictionary of route names and their associated thresholds.

​is_valid

​RouterConfig Objects

​__init__

​from_file

​from_tuples

​from_index

​to_dict

​to_file

​to_utterances

​add

​get

​remove

​get_hash

​xq_reshape

​BaseRouter Objects

​__init__

​check_for_matching_routes

​__call__

​acall

​sync

​async_sync

​from_json

​from_yaml

​from_config

​add

​aadd

​update

​delete

​adelete

​is_synced

​async_is_synced

​get_utterance_diff

​aget_utterance_diff

​get

​group_scores_by_class

​set_threshold

​to_config

​to_json

​to_yaml

​get_thresholds

​fit

​evaluate

​threshold_random_search

is_valid

RouterConfig Objects

init

from_file

from_tuples

from_index

to_dict

to_file

to_utterances

add

get

remove

get_hash

xq_reshape

BaseRouter Objects

init

check_for_matching_routes

call

acall

sync

async_sync

from_json

from_yaml

from_config

add

aadd

update

delete

adelete

is_synced

async_is_synced

get_utterance_diff

aget_utterance_diff

get

group_scores_by_class

set_threshold

to_config

to_json

to_yaml

get_thresholds

fit

evaluate

threshold_random_search