lint

semantic_router/index/base.py

@@ -13,8 +13,8 @@ class BaseIndex(BaseModel):
     # You can define common attributes here if there are any.
     # For example, a placeholder for the index attribute:
     index: Optional[Any] = None
-    routes: Optional[List[str]] = None
-    utterances: Optional[List[str]] = None
+    routes: Optional[np.ndarray] = None
+    utterances: Optional[np.ndarray] = None
     dimensions: Union[int, None] = None
     type: str = "base"
 

semantic_router/index/local.py

@@ -8,12 +8,10 @@ class LocalIndex(BaseIndex):
     def __init__(
         self,
         index: Optional[np.ndarray] = None,
-        routes: Optional[List[str]] = None,
-        utterances: Optional[List[str]] = None,
+        routes: Optional[np.ndarray] = None,
+        utterances: Optional[np.ndarray] = None,
     ):
-        super().__init__(
-            index=index, routes=routes, utterances=utterances
-        )
+        super().__init__(index=index, routes=routes, utterances=utterances)
         self.type = "local"
 
     class Config:  # Stop pydantic from complaining about  Optional[np.ndarray] type hints.
@@ -41,9 +39,9 @@ def get_routes(self) -> List[Tuple]:
         Returns:
             List[Tuple]: A list of (route_name, utterance) objects.
         """
-        if self.route_names is None or self.utterances is None:
+        if self.routes is None or self.utterances is None:
             raise ValueError("No routes have been added to the index.")
-        return list(zip(self.route_names, self.utterances))
+        return list(zip(self.routes, self.utterances))
 
     def describe(self) -> dict:
         return {
@@ -64,7 +62,7 @@ def query(self, vector: np.ndarray, top_k: int = 5) -> Tuple[np.ndarray, List[st
         # get routes from index values
         route_names = self.routes[idx].copy()
         return scores, route_names
-    
+
     def delete(self, route_name: str):
         """
         Delete all records of a specific route from the index.
@@ -95,7 +93,7 @@ def _get_indices_for_route(self, route_name: str):
             raise ValueError("Routes are not populated.")
         idx = [i for i, route in enumerate(self.routes) if route == route_name]
         return idx
-    
+
     def __len__(self):
         if self.index is not None:
             return self.index.shape[0]

semantic_router/index/pinecone.py

@@ -3,7 +3,7 @@
 import time
 import hashlib
 import os
-from typing import Any, List, Tuple, Optional, Union
+from typing import Any, Dict, List, Tuple, Optional, Union
 from semantic_router.index.base import BaseIndex
 from semantic_router.utils.logger import logger
 import numpy as np
@@ -128,11 +128,13 @@ def _get_route_ids(self, route_name: str):
         clean_route = clean_route_name(route_name)
         ids, _ = self._get_all(prefix=f"{clean_route}#")
         return ids
-    
+
     def _get_all(self, prefix: Optional[str] = None, include_metadata: bool = False):
         """
         Retrieves all vector IDs from the Pinecone index using pagination.
         """
+        if self.index is None:
+            raise ValueError("Index is None, could not retrieve vector IDs.")
         all_vector_ids = []
         next_page_token = None
 
@@ -143,8 +145,8 @@ def _get_all(self, prefix: Optional[str] = None, include_metadata: bool = False)
 
         # Construct the request URL for listing vectors. Adjust parameters as needed.
         list_url = f"https://{self.host}/vectors/list{prefix_str}"
-        params = {}
-        headers = {"Api-Key": os.getenv("PINECONE_API_KEY")}
+        params: Dict = {}
+        headers = {"Api-Key": os.environ["PINECONE_API_KEY"]}
         metadata = []
 
         while True:
@@ -164,17 +166,15 @@ def _get_all(self, prefix: Optional[str] = None, include_metadata: bool = False)
             if include_metadata:
                 res_meta = self.index.fetch(ids=vector_ids)
                 # extract metadata only
-                metadata.extend(
-                    [x["metadata"] for x in res_meta["vectors"].values()]
-                )
+                metadata.extend([x["metadata"] for x in res_meta["vectors"].values()])
 
             # Check if there's a next page token; if not, break the loop
             next_page_token = response_data.get("pagination", {}).get("next")
             if not next_page_token:
                 break
 
         return all_vector_ids, metadata
-    
+
     def get_routes(self) -> List[Tuple]:
         """
         Gets a list of route and utterance objects currently stored in the index.

semantic_router/layer.py

@@ -303,8 +303,7 @@ def delete(self, route_name: str):
             self.index.delete(route_name=route_name)
 
     def _refresh_routes(self):
-        """Pulls out the latest routes from the index.
-        """
+        """Pulls out the latest routes from the index."""
         raise NotImplementedError("This method has not yet been implemented.")
         route_mapping = {route.name: route for route in self.routes}
         index_routes = self.index.get_routes()
@@ -320,7 +319,6 @@ def _refresh_routes(self):
             route = route_mapping[route_name]
             self.routes.append(route)
 
-
     def _add_routes(self, routes: List[Route]):
         # create embeddings for all routes
         all_utterances = [
@@ -458,7 +456,7 @@ def _vec_evaluate(self, Xq: Union[List[float], Any], y: List[str]) -> float:
                 correct += 1
         accuracy = correct / len(Xq)
         return accuracy
-    
+
     def _get_route_names(self) -> List[str]:
         return [route.name for route in self.routes]
 

tests/unit/test_layer.py

@@ -302,7 +302,9 @@ def test_config(self, openai_encoder, routes):
         # now load from config and confirm it's the same
         route_layer_from_config = RouteLayer.from_config(layer_config)
         assert (route_layer_from_config.index.index == route_layer.index.index).all()
-        assert route_layer_from_config._get_route_names() == route_layer._get_route_names()
+        assert (
+            route_layer_from_config._get_route_names() == route_layer._get_route_names()
+        )
         assert route_layer_from_config.score_threshold == route_layer.score_threshold
 
     def test_get_thresholds(self, openai_encoder, routes):