Source code for redisvl.extensions.router.schema

from enum import Enum
from typing import Dict, List, Optional

from pydantic.v1 import BaseModel, Field, validator

from redisvl.schema import IndexInfo, IndexSchema


[docs] class Route(BaseModel): """Model representing a routing path with associated metadata and thresholds.""" name: str """The name of the route.""" references: List[str] """List of reference phrases for the route.""" metadata: Dict[str, str] = Field(default={}) """Metadata associated with the route.""" distance_threshold: Optional[float] = Field(default=None) """Distance threshold for matching the route.""" @validator("name") def name_must_not_be_empty(cls, v): if not v or not v.strip(): raise ValueError("Route name must not be empty") return v @validator("references") def references_must_not_be_empty(cls, v): if not v: raise ValueError("References must not be empty") if any(not ref.strip() for ref in v): raise ValueError("All references must be non-empty strings") return v @validator("distance_threshold") def distance_threshold_must_be_positive(cls, v): if v is not None and v <= 0: raise ValueError("Route distance threshold must be greater than zero") return v
[docs] class RouteMatch(BaseModel): """Model representing a matched route with distance information.""" name: Optional[str] = None """The matched route name.""" distance: Optional[float] = Field(default=None) """The vector distance between the statement and the matched route."""
[docs] class DistanceAggregationMethod(Enum): """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."""
[docs] class RoutingConfig(BaseModel): """Configuration for routing behavior.""" distance_threshold: float = Field(default=0.5) """The threshold for semantic distance.""" max_k: int = Field(default=1) """The maximum number of top matches to return.""" aggregation_method: DistanceAggregationMethod = Field( default=DistanceAggregationMethod.avg ) """Aggregation method to use to classify queries.""" @validator("max_k") def max_k_must_be_positive(cls, v): if v <= 0: raise ValueError("max_k must be a positive integer") return v @validator("distance_threshold") def distance_threshold_must_be_valid(cls, v): if v <= 0 or v > 1: raise ValueError("distance_threshold must be between 0 and 1") return v
class SemanticRouterIndexSchema(IndexSchema): """Customized index schema for SemanticRouter.""" @classmethod def from_params(cls, name: str, vector_dims: int) -> "SemanticRouterIndexSchema": """Create an index schema based on router name and vector dimensions. Args: name (str): The name of the index. vector_dims (int): The dimensions of the vectors. Returns: SemanticRouterIndexSchema: The constructed index schema. """ return cls( index=IndexInfo(name=name, prefix=name), fields=[ # type: ignore {"name": "route_name", "type": "tag"}, {"name": "reference", "type": "text"}, { "name": "vector", "type": "vector", "attrs": { "algorithm": "flat", "dims": vector_dims, "distance_metric": "cosine", "datatype": "float32", }, }, ], )