Semantic Router

Semantic Router#

Semantic Router#

class SemanticRouter(name, routes, vectorizer=None, routing_config=None, redis_client=None, redis_url='redis://localhost:6379', overwrite=False, connection_kwargs={})[source]#

Semantic Router for managing and querying route vectors.

Initialize the SemanticRouter.

Parameters:

  • name (str) – The name of the semantic router.

  • routes (List[Route]) – List of Route objects.

  • vectorizer (BaseVectorizer, optional) – The vectorizer used to embed route references. Defaults to default HFTextVectorizer.

  • routing_config (RoutingConfig, optional) – Configuration for routing behavior. Defaults to the default RoutingConfig.

  • redis_client (Optional[SyncRedisClient], optional) – Redis client for connection. Defaults to None.

  • redis_url (str, optional) – The redis url. Defaults to redis://localhost:6379.

  • overwrite (bool, optional) – Whether to overwrite existing index. Defaults to False.

  • connection_kwargs (Dict[str, Any]) – The connection arguments for the redis client. Defaults to empty {}.

add_route(route)[source]#

Add a new route to the SemanticRouter.

Embeds the route’s references, writes them to the Redis index, appends the route to self.routes, and persists the updated router config so the route survives SemanticRouter.from_existing().

Parameters:

route (Route) – A fully-formed Route (name, references, distance_threshold, optional metadata).

Returns:

The added route’s name.

Return type:

str

Raises:

ValueError – If a route with this name already exists on the router. Use add_route_references() to extend an existing route.

add_route_references(route_name, references)[source]#

Add a reference(s) to an existing route.

Parameters:

  • router_name (str) – The name of the router.

  • references (Union[str, List[str]]) – The reference or list of references to add.

  • route_name (str)

Returns:

The list of added references keys.

Return type:

List[str]

clear()[source]#

Flush all routes from the semantic router index.

Return type:

None

delete()[source]#

Delete the semantic router index.

Return type:

None

delete_route_references(route_name='', reference_ids=[], keys=[])[source]#

Get references for an existing semantic router route.

Parameters:

  • Optional (keys) – The name of the router.

  • Optional – The reference or list of references to delete.

  • Optional – List of fully qualified keys (prefix:router:reference_id) to delete.

  • route_name (str)

  • reference_ids (list[str])

  • keys (list[str])

Returns:

Number of objects deleted

Return type:

int

classmethod from_dict(data, **kwargs)[source]#

Create a SemanticRouter from a dictionary.

Parameters:

data (Dict[str, Any]) – The dictionary containing the semantic router data.

Returns:

The semantic router instance.

Return type:

SemanticRouter

Raises:

ValueError – If required data is missing or invalid.

from redisvl.extensions.router import SemanticRouter
router_data = {
    "name": "example_router",
    "routes": [{"name": "route1", "references": ["ref1"], "distance_threshold": 0.5}],
    "vectorizer": {"type": "openai", "model": "text-embedding-ada-002"},
}
router = SemanticRouter.from_dict(router_data)
classmethod from_existing(name, redis_client=None, redis_url='redis://localhost:6379', **kwargs)[source]#

Return SemanticRouter instance from existing index.

Parameters:

  • name (str)

  • redis_client (Redis | RedisCluster | None)

  • redis_url (str)

Return type:

SemanticRouter

classmethod from_yaml(file_path, **kwargs)[source]#

Create a SemanticRouter from a YAML file.

Parameters:

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

Returns:

The semantic router instance.

Return type:

SemanticRouter

Raises:

  • ValueError – If the file path is invalid.

  • FileNotFoundError – If the file does not exist.

from redisvl.extensions.router import SemanticRouter
router = SemanticRouter.from_yaml("router.yaml", redis_url="redis://localhost:6379")
get(route_name)[source]#

Get a route by its name.

Parameters:

route_name (str) – Name of the route.

Returns:

The selected Route object or None if not found.

Return type:

Optional[Route]

get_route_references(route_name='', reference_ids=[], keys=[])[source]#

Get references for an existing route route.

Parameters:

  • router_name (str) – The name of the router.

  • references (Union[str, List[str]]) – The reference or list of references to add.

  • route_name (str)

  • reference_ids (list[str])

  • keys (list[str])

Returns:

Reference objects stored

Return type:

List[Dict[str, Any]]]

model_post_init(context, /)#

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self (BaseModel) – The BaseModel instance.

  • context (Any) – The context.

Return type:

None

remove_route(route_name)[source]#

Remove a route and all references from the semantic router.

Parameters:

route_name (str) – Name of the route to remove.

Return type:

None

route_many(statement=None, vector=None, max_k=None, distance_threshold=None, aggregation_method=None)[source]#

Query the semantic router with a given statement or vector for multiple matches.

Parameters:

  • statement (Optional[str]) – The input statement to be queried.

  • vector (Optional[List[float]]) – The input vector to be queried.

  • max_k (Optional[int]) – The maximum number of top matches to return.

  • distance_threshold (Optional[float]) – The threshold for semantic distance.

  • aggregation_method (Optional[DistanceAggregationMethod]) – The aggregation method used for vector distances.

Returns:

The matching routes and their details.

Return type:

List[RouteMatch]

to_dict()[source]#

Convert the SemanticRouter instance to a dictionary.

Returns:

The dictionary representation of the SemanticRouter.

Return type:

Dict[str, Any]

from redisvl.extensions.router import SemanticRouter
router = SemanticRouter(name="example_router", routes=[], redis_url="redis://localhost:6379")
router_dict = router.to_dict()
to_yaml(file_path, overwrite=True)[source]#

Write the semantic router to a YAML file.

Parameters:

  • file_path (str) – The path to the YAML file.

  • overwrite (bool) – Whether to overwrite the file if it already exists.

Raises:

FileExistsError – If the file already exists and overwrite is False.

Return type:

None

from redisvl.extensions.router import SemanticRouter
router = SemanticRouter(
    name="example_router",
    routes=[],
    redis_url="redis://localhost:6379"
)
router.to_yaml("router.yaml")
update_route_thresholds(route_thresholds)[source]#

Update the distance thresholds for each route.

Parameters:

route_thresholds (Dict[str, float]) – Dictionary of route names and their distance thresholds.

update_routing_config(routing_config)[source]#

Update the routing configuration.

Parameters:

routing_config (RoutingConfig) – The new routing configuration.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str#

The name of the semantic router.

property route_names: list[str]#

Get the list of route names.

Returns:

List of route names.

Return type:

List[str]

property route_thresholds: dict[str, float | None]#

Get the distance thresholds for each route.

Returns:

Dictionary of route names and their distance thresholds.

Return type:

Dict[str, float]

routes: list[Route]#

List of Route objects.

routing_config: RoutingConfig#

Configuration for routing behavior.

vectorizer: BaseVectorizer#

The vectorizer used to embed route references.

Routing Config#

class RoutingConfig(*, max_k=1, aggregation_method=DistanceAggregationMethod.avg)[source]#

Configuration for routing behavior.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

  • max_k (Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Strict(strict=True), Gt(gt=0)])])

  • aggregation_method (DistanceAggregationMethod)

max_k: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Strict(strict=True), Gt(gt=0)])]#

Aggregation method to use to classify queries.

model_config: ClassVar[ConfigDict] = {'extra': 'ignore'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Route#

class Route(*, name, references, metadata={}, distance_threshold=0.5)[source]#

Model representing a routing path with associated metadata and thresholds.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

  • name (str)

  • references (list[str])

  • metadata (dict[str, Any])

  • distance_threshold (Annotated[float, FieldInfo(annotation=NoneType, required=True, metadata=[Strict(strict=True), Gt(gt=0), Le(le=2)])])

distance_threshold: Annotated[float, FieldInfo(annotation=NoneType, required=True, metadata=[Strict(strict=True), Gt(gt=0), Le(le=2)])]#

Distance threshold for matching the route.

metadata: dict[str, Any]#

Metadata associated with the route.

model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str#

The name of the route.

references: list[str]#

List of reference phrases for the route.

Route Match#

class RouteMatch(*, name=None, distance=None)[source]#

Model representing a matched route with distance information.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

  • name (str | None)

  • distance (float | None)

distance: float | None#

The vector distance between the statement and the matched route.

model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str | None#

The matched route name.

Distance Aggregation Method#

class DistanceAggregationMethod(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Enumeration for distance aggregation methods.

avg = 'avg'#

Compute the average of the vector distances.

min = 'min'#

Compute the minimum of the vector distances.

sum = 'sum'#

Compute the sum of the vector distances.