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