diff --git a/semantic_router/index/pinecone.py b/semantic_router/index/pinecone.py
index 6086de1c5a3de8d18023c3488064399220a9f8a5..7b4dba53dba3c36946ff1d01c4441a2dc5f9c310 100644
--- a/semantic_router/index/pinecone.py
+++ b/semantic_router/index/pinecone.py
@@ -387,7 +387,9 @@ class PineconeIndex(BaseIndex):
if sparse_vector is not None:
if isinstance(sparse_vector, dict):
sparse_vector = SparseEmbedding.from_dict(sparse_vector)
- sparse_vector = sparse_vector.to_pinecone()
+ if isinstance(sparse_vector, SparseEmbedding):
+ # unnecessary if-statement but mypy didn't like this otherwise
+ sparse_vector = sparse_vector.to_pinecone()
results = self.index.query(
vector=[query_vector_list],
sparse_vector=sparse_vector,
diff --git a/semantic_router/routers/base.py b/semantic_router/routers/base.py
index 0172e5d803c6a3e9f73aa4c44a41381f1b5de4d2..5d59ae580c74f1164a71bcf38f762d26f987499c 100644
--- a/semantic_router/routers/base.py
+++ b/semantic_router/routers/base.py
@@ -413,7 +413,7 @@ class BaseRouter(BaseModel):
if vector is None:
if text is None:
raise ValueError("Either text or vector must be provided")
- vector = self._encode(text=text)
+ vector = self._encode(text=[text])
route, top_class_scores = self._retrieve_top_route(vector, route_filter)
passed = self._check_threshold(top_class_scores, route)
if passed and route is not None and not simulate_static:
@@ -455,7 +455,7 @@ class BaseRouter(BaseModel):
if vector is None:
if text is None:
raise ValueError("Either text or vector must be provided")
- vector = await self._async_encode(text=text)
+ vector = await self._async_encode(text=[text])
route, top_class_scores = await self._async_retrieve_top_route(
vector, route_filter
@@ -494,7 +494,7 @@ class BaseRouter(BaseModel):
if vector is None:
if text is None:
raise ValueError("Either text or vector must be provided")
- vector_arr = self._encode(text=text)
+ vector_arr = self._encode(text=[text])
else:
vector_arr = np.array(vector)
print(f"{text=}")
@@ -967,26 +967,26 @@ class BaseRouter(BaseModel):
return route_names, utterances, function_schemas, metadata
return route_names, utterances, function_schemas
- def _encode(self, text: str) -> Any:
+ def _encode(self, text: list[str]) -> Any:
"""Generates embeddings for a given text.
Must be implemented by a subclass.
:param text: The text to encode.
- :type text: str
+ :type text: list[str]
:return: The embeddings of the text.
:rtype: Any
"""
# TODO: should encode "content" rather than text
raise NotImplementedError("This method should be implemented by subclasses.")
- async def _async_encode(self, text: str) -> Any:
+ async def _async_encode(self, text: list[str]) -> Any:
"""Asynchronously generates embeddings for a given text.
Must be implemented by a subclass.
:param text: The text to encode.
- :type text: str
+ :type text: list[str]
:return: The embeddings of the text.
:rtype: Any
"""
@@ -1029,6 +1029,9 @@ class BaseRouter(BaseModel):
def _semantic_classify(self, query_results: List[Dict]) -> Tuple[str, List[float]]:
scores_by_class = self.group_scores_by_class(query_results)
+ if self.aggregation_method is None:
+ raise ValueError("self.aggregation_method is not set.")
+
# Calculate total score for each class
total_scores = {
route: self.aggregation_method(scores)
@@ -1048,6 +1051,9 @@ class BaseRouter(BaseModel):
) -> Tuple[str, List[float]]:
scores_by_class = await self.async_group_scores_by_class(query_results)
+ if self.aggregation_method is None:
+ raise ValueError("self.aggregation_method is not set.")
+
# Calculate total score for each class
total_scores = {
route: self.aggregation_method(scores)
diff --git a/semantic_router/routers/hybrid.py b/semantic_router/routers/hybrid.py
index be3880abbbe9afd7e8f368695eaa6d34e7219ca8..91ecf2ef7b8df89eca5332b63f8d3277f5ab95d2 100644
--- a/semantic_router/routers/hybrid.py
+++ b/semantic_router/routers/hybrid.py
@@ -1,4 +1,4 @@
-from typing import Any, List, Optional
+from typing import List, Optional
import asyncio
from pydantic.v1 import Field
@@ -6,6 +6,7 @@ import numpy as np
from semantic_router.encoders import (
DenseEncoder,
+ SparseEncoder,
BM25Encoder,
TfidfEncoder,
)
@@ -21,13 +22,13 @@ class HybridRouter(BaseRouter):
"""A hybrid layer that uses both dense and sparse embeddings to classify routes."""
# there are a few additional attributes for hybrid
- sparse_encoder: Optional[DenseEncoder] = Field(default=None)
+ sparse_encoder: Optional[SparseEncoder] = Field(default=None)
alpha: float = 0.3
def __init__(
self,
encoder: DenseEncoder,
- sparse_encoder: Optional[DenseEncoder] = None,
+ sparse_encoder: Optional[SparseEncoder] = None,
llm: Optional[BaseLLM] = None,
routes: List[Route] = [],
index: Optional[HybridLocalIndex] = None,
@@ -70,47 +71,49 @@ class HybridRouter(BaseRouter):
index = index
return index
- def _set_sparse_encoder(self, sparse_encoder: Optional[DenseEncoder]):
+ def _set_sparse_encoder(self, sparse_encoder: Optional[SparseEncoder]):
if sparse_encoder is None:
logger.warning("No sparse_encoder provided. Using default BM25Encoder.")
self.sparse_encoder = BM25Encoder()
else:
self.sparse_encoder = sparse_encoder
- def _encode(self, text: list[str]) -> tuple[np.ndarray, list[dict[int, float]]]:
+ def _encode(self, text: list[str]) -> tuple[np.ndarray, list[SparseEmbedding]]:
"""Given some text, generates dense and sparse embeddings, then scales them
using the chosen alpha value.
"""
+ if self.sparse_encoder is None:
+ raise ValueError("self.sparse_encoder is not set.")
# TODO: should encode "content" rather than text
# TODO: add alpha as a parameter
# create dense query vector
xq_d = np.array(self.encoder(text))
# xq_d = np.squeeze(xq_d) # Reduce to 1d array.
# create sparse query vector dict
- xq_s_dict = self.sparse_encoder(text)
+ xq_s = self.sparse_encoder(text)
# xq_s = np.squeeze(xq_s)
# convex scaling
- xq_d, xq_s_dict = self._convex_scaling(xq_d, xq_s_dict)
- return xq_d, xq_s_dict
+ xq_d, xq_s = self._convex_scaling(dense=xq_d, sparse=xq_s)
+ return xq_d, xq_s
- async def _async_encode(self, text: List[str]) -> Any:
+ async def _async_encode(
+ self, text: List[str]
+ ) -> tuple[np.ndarray, list[SparseEmbedding]]:
"""Given some text, generates dense and sparse embeddings, then scales them
using the chosen alpha value.
"""
+ if self.sparse_encoder is None:
+ raise ValueError("self.sparse_encoder is not set.")
# TODO: should encode "content" rather than text
# TODO: add alpha as a parameter
# async encode both dense and sparse
dense_coro = self.encoder.acall(text)
sparse_coro = self.sparse_encoder.acall(text)
- dense_vec, sparse_vec = await asyncio.gather(dense_coro, sparse_coro)
+ dense_vec, xq_s = await asyncio.gather(dense_coro, sparse_coro)
# create dense query vector
xq_d = np.array(dense_vec)
- # xq_d = np.squeeze(xq_d) # reduce to 1d array
- # create sparse query vector
- xq_s = np.array(sparse_vec)
- # xq_s = np.squeeze(xq_s)
# convex scaling
- xq_d, xq_s = self._convex_scaling(xq_d, xq_s)
+ xq_d, xq_s = self._convex_scaling(dense=xq_d, sparse=xq_s)
return xq_d, xq_s
def __call__(
@@ -146,12 +149,18 @@ class HybridRouter(BaseRouter):
else:
return RouteChoice()
- def _convex_scaling(self, dense: np.ndarray, sparse: list[dict[int, float]]):
+ def _convex_scaling(
+ self, dense: np.ndarray, sparse: list[SparseEmbedding]
+ ) -> tuple[np.ndarray, list[SparseEmbedding]]:
+ # TODO: better way to do this?
+ sparse_dicts = [sparse_vec.to_dict() for sparse_vec in sparse]
# scale sparse and dense vecs
scaled_dense = np.array(dense) * self.alpha
scaled_sparse = []
- for sparse_dict in sparse:
+ for sparse_dict in sparse_dicts:
scaled_sparse.append(
- {k: v * (1 - self.alpha) for k, v in sparse_dict.items()}
+ SparseEmbedding.from_dict(
+ {k: v * (1 - self.alpha) for k, v in sparse_dict.items()}
+ )
)
return scaled_dense, scaled_sparse
diff --git a/semantic_router/routers/semantic.py b/semantic_router/routers/semantic.py
index 59cd8de18319e3049a458da997022dc5cd407aee..ff7f73f2a853563b574641aef98a94eedc94ccaa 100644
--- a/semantic_router/routers/semantic.py
+++ b/semantic_router/routers/semantic.py
@@ -35,16 +35,16 @@ class SemanticRouter(BaseRouter):
if self.auto_sync:
self._init_index_state()
- def _encode(self, text: str) -> Any:
+ def _encode(self, text: list[str]) -> Any:
"""Given some text, encode it."""
# create query vector
- xq = np.array(self.encoder([text]))
+ xq = np.array(self.encoder(text))
xq = np.squeeze(xq) # Reduce to 1d array.
return xq
- async def _async_encode(self, text: str) -> Any:
+ async def _async_encode(self, text: list[str]) -> Any:
"""Given some text, encode it."""
# create query vector
- xq = np.array(await self.encoder.acall(docs=[text]))
+ xq = np.array(await self.encoder.acall(docs=text))
xq = np.squeeze(xq) # Reduce to 1d array.
return xq