​
MetricPgVecOperatorMap Objects

class MetricPgVecOperatorMap(Enum)

Enum to map the metric to PostgreSQL vector operators.

​
dotproduct

inner product

​
euclidean

L2 distance

​
manhattan

L1 distance

​
parse_vector

def parse_vector(vector_str: Union[str, Any]) -> List[float]

Parses a vector from a string or other representation.

Arguments:

  • vector_str (Union[str, Any]): The string or object representation of a vector.

Returns:

List[float]: A list of floats representing the vector.

​
clean_route_name

def clean_route_name(route_name: str) -> str

Cleans and formats the route name by stripping spaces and replacing them with hyphens.

Arguments:

  • route_name (str): The original route name.

Returns:

str: The cleaned and formatted route name.

​
PostgresIndexRecord Objects

class PostgresIndexRecord(BaseModel)

Model to represent a record in the Postgres index.

​
__init__

def __init__(**data) -> None

Initializes a new Postgres index record with given data.

Arguments:

  • data (dict): Field values for the record.

​
to_dict

def to_dict() -> Dict

Converts the record to a dictionary.

Returns:

Dict: A dictionary representation of the record.

​
PostgresIndex Objects

class PostgresIndex(BaseIndex)

Postgres implementation of Index.

​
__init__

def __init__(connection_string: Optional[str] = None,
             index_prefix: str = "semantic_router_",
             index_name: str = "index",
             metric: Metric = Metric.COSINE,
             namespace: Optional[str] = "",
             dimensions: int | None = None)

Initializes the Postgres index with the specified parameters.

Arguments:

  • connection_string (Optional[str]): The connection string for the PostgreSQL database.
  • index_prefix (str): The prefix for the index table name.
  • index_name (str): The name of the index table.
  • dimensions (int): The number of dimensions for the vectors.
  • metric (Metric): The metric used for vector comparisons.
  • Optional[str]0 (Optional[str]): An optional namespace for the index.

​
setup_index

def setup_index() -> None

Sets up the index by creating the table and vector extension if they do not exist.

Raises:

  • ValueError: If the existing table’s vector dimensions do not match the expected dimensions.
  • TypeError: If the database connection is not established.

​
add

def add(embeddings: List[List[float]],
        routes: List[str],
        utterances: List[str],
        function_schemas: Optional[List[Dict[str, Any]]] = None,
        metadata_list: List[Dict[str, Any]] = [],
        **kwargs) -> None

Adds records to the index.

Arguments:

  • embeddings (List[List[float]]): A list of vector embeddings to add.
  • routes (List[str]): A list of route names corresponding to the embeddings.
  • utterances (List[Any]): A list of utterances corresponding to the embeddings.
  • function_schemas (Optional[List[Dict[str, Any]]]): A list of function schemas corresponding to the embeddings.
  • metadata_list (List[Dict[str, Any]]): A list of metadata corresponding to the embeddings.

Raises:

  • List[List[float]]0: If the vector embeddings being added do not match the expected dimensions.
  • List[List[float]]1: If the database connection is not established.

​
delete

def delete(route_name: str) -> None

Deletes records with the specified route name.

Arguments:

  • route_name (str): The name of the route to delete records for.

Raises:

  • TypeError: If the database connection is not established.

​
describe

def describe() -> IndexConfig

Describes the index by returning its type, dimensions, and total vector count.

Returns:

IndexConfig: An IndexConfig object containing the index’s type, dimensions, and total vector count.

​
is_ready

def is_ready() -> bool

Checks if the index is ready to be used.

Returns:

bool: True if the index is ready, False otherwise.

​
query

def query(
    vector: np.ndarray,
    top_k: int = 5,
    route_filter: Optional[List[str]] = None,
    sparse_vector: dict[int, float] | SparseEmbedding | None = None
) -> Tuple[np.ndarray, List[str]]

Searches the index for the query vector and returns the top_k results.

Arguments:

  • vector (np.ndarray): The query vector.
  • top_k (int): The number of top results to return.
  • route_filter (Optional[List[str]]): Optional list of routes to filter the results by.
  • sparse_vector (dict[int, float] | SparseEmbedding | None): Optional sparse vector to filter the results by.

Raises:

  • TypeError: If the database connection is not established.

Returns:

Tuple[np.ndarray, List[str]]: A tuple containing the scores and routes of the top_k results.

​
delete_all

def delete_all()

Deletes all records from the Postgres index.

Raises:

  • TypeError: If the database connection is not established.

​
delete_index

def delete_index() -> None

Deletes the entire table for the index.

Raises:

  • TypeError: If the database connection is not established.

​
aget_routes

def aget_routes()

Asynchronously get all routes from the index.

Not yet implemented for PostgresIndex.

Returns:

List[str]: A list of routes.

​
__len__

def __len__()

Returns the total number of vectors in the index.

Raises:

  • TypeError: If the database connection is not established.

Returns:

int: The total number of vectors.