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.

semantic_router.routers.base

On this page