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={}, create_index=True)
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 {}.
- create_index (bool , optional) – Whether RedisVL creates and validates
the index. When False the constructor issues no index command at
all and writes nothing: the index must already exist, already
hold the reference vectors for
routes, and already have its stored config, since none of that is written or verified.routesmust match what is indexed, because each route’s distance threshold is applied from this local list – and add_route rewrites the stored config from that same list, so attaching with a partial set and then adding a route truncates the config every other client reads. SeeSemanticCachefor a worked example of the flag, and Install RedisVL for the ACL details. Defaults to True.
- Raises: ValueError – If both create_index is False and overwrite is True.
add_route(route)
Add a new route to the SemanticRouter.
Note that this replaces the router’s stored config with this instance’s route list, so a router constructed with a subset of the indexed routes will drop the rest from the config that from_existing reads.
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 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)
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()
Flush all routes from the semantic router index.
- Return type: None
delete()
Delete the semantic router index and its persisted route config.
- Return type: None
delete_route_references(route_name='', reference_ids=[], keys=[])
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)
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)
Return SemanticRouter instance from existing index.
Reads the stored route config with JSON.GET, so unlike
SearchIndex.from_existing() this needs no FT.INFO. Pass
create_index=False to keep it that way through construction, which
makes this the way to attach to a router with a credential that cannot
run index-metadata commands.
- Parameters:
- name (str)
- redis_client (Redis | RedisCluster | None)
- redis_url (str)
- Return type: SemanticRouter
classmethod from_yaml(file_path, **kwargs)
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)
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=[])
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 initialise 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)
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)
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()
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)
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)
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)
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)
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)
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)
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)
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.