semantic_router.index.pinecone

​

build_records

def build_records(
    embeddings: List[List[float]],
    routes: List[str],
    utterances: List[str],
    function_schemas: Optional[Optional[List[Dict[str, Any]]]] = None,
    metadata_list: List[Dict[str, Any]] = [],
    sparse_embeddings: Optional[Optional[List[SparseEmbedding]]] = None
) -> List[Dict]

Build records for Pinecone upsert.

Arguments:

• embeddings (List[List[float]]): List of embeddings to upsert.
• routes (List[str]): List of routes to upsert.
• utterances (List[str]): List of utterances to upsert.
• function_schemas (Optional[List[Dict[str, Any]]]): List of function schemas to upsert.
• metadata_list (List[Dict[str, Any]]): List of metadata to upsert.
• List[List[float]]0 (List[List[float]]1): List of sparse embeddings to upsert.

Returns:

List[List[float]]2: List of records to upsert.

​

PineconeRecord Objects

class PineconeRecord(BaseModel)

​

metadata

Additional metadata dictionary

​

__init__

def __init__(**data)

Initialize PineconeRecord.

Arguments:

• **data (dict): Keyword arguments to pass to the BaseModel constructor.

​

to_dict

def to_dict()

Convert PineconeRecord to a dictionary.

Returns:

dict: Dictionary representation of the PineconeRecord.

​

PineconeIndex Objects

class PineconeIndex(BaseIndex)

​

__init__

def __init__(api_key: Optional[str] = None,
             index_name: str = "index",
             dimensions: Optional[int] = None,
             metric: str = "dotproduct",
             cloud: str = "aws",
             region: str = "us-east-1",
             host: str = "",
             namespace: Optional[str] = "",
             base_url: Optional[str] = "https://api.pinecone.io",
             init_async_index: bool = False)

Initialize PineconeIndex.

Arguments:

• api_key (Optional[str]): Pinecone API key.
• index_name (str): Name of the index.
• dimensions (Optional[int]): Dimensions of the index.
• metric (str): Metric of the index.
• cloud (str): Cloud provider of the index.
• Optional[str]0 (str): Region of the index.
• Optional[str]2 (str): Host of the index.
• Optional[str]4 (Optional[str]): Namespace of the index.
• Optional[str]6 (Optional[str]): Base URL of the Pinecone API.
• Optional[str]8 (Optional[str]9): Whether to initialize the index asynchronously.

​

add

def add(embeddings: List[List[float]],
        routes: List[str],
        utterances: List[str],
        function_schemas: Optional[Optional[List[Dict[str, Any]]]] = None,
        metadata_list: List[Dict[str, Any]] = [],
        batch_size: int = 100,
        sparse_embeddings: Optional[Optional[List[SparseEmbedding]]] = None,
        **kwargs)

Add vectors to Pinecone in batches.

Arguments:

• embeddings (List[List[float]]): List of embeddings to upsert.
• routes (List[str]): List of routes to upsert.
• utterances (List[str]): List of utterances to upsert.
• function_schemas (Optional[List[Dict[str, Any]]]): List of function schemas to upsert.
• metadata_list (List[Dict[str, Any]]): List of metadata to upsert.
• List[List[float]]0 (List[List[float]]1): Number of vectors to upsert in a single batch.
• List[List[float]]2 (List[List[float]]3): List of sparse embeddings to upsert.

​

aadd

async def aadd(
        embeddings: List[List[float]],
        routes: List[str],
        utterances: List[str],
        function_schemas: Optional[Optional[List[Dict[str, Any]]]] = None,
        metadata_list: List[Dict[str, Any]] = [],
        batch_size: int = 100,
        sparse_embeddings: Optional[Optional[List[SparseEmbedding]]] = None,
        **kwargs)

Add vectors to Pinecone in batches.

Arguments:

• embeddings (List[List[float]]): List of embeddings to upsert.
• routes (List[str]): List of routes to upsert.
• utterances (List[str]): List of utterances to upsert.
• function_schemas (Optional[List[Dict[str, Any]]]): List of function schemas to upsert.
• metadata_list (List[Dict[str, Any]]): List of metadata to upsert.
• List[List[float]]0 (List[List[float]]1): Number of vectors to upsert in a single batch.
• List[List[float]]2 (List[List[float]]3): List of sparse embeddings to upsert.

​

delete

def delete(route_name: str) -> list[str]

Delete specified route from index if it exists. Returns the IDs of the vectors

deleted.

Arguments:

• route_name (str): Name of the route to delete.

Returns:

list[str]: List of IDs of the vectors deleted.

​

adelete

async def adelete(route_name: str) -> list[str]

Asynchronously delete specified route from index if it exists. Returns the IDs

of the vectors deleted.

Arguments:

• route_name (str): Name of the route to delete.

Returns:

list[str]: List of IDs of the vectors deleted.

​

delete_all

def delete_all()

Delete all routes from index if it exists.

Returns:

None: None

​

describe

def describe() -> IndexConfig

Describe the index.

Returns:

IndexConfig: IndexConfig

​

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]]

Search the index for the query vector and return the top_k results.

Arguments:

• vector (np.ndarray): The query vector to search for.
• top_k (int, optional): The number of top results to return, defaults to 5.
• route_filter (Optional[List[str]], optional): A list of route names to filter the search results, defaults to None.
• sparse_vector (Optional[SparseEmbedding]): An optional sparse vector to include in the query.
• kwargs (Any): Additional keyword arguments for the query, including sparse_vector.

Raises:

• np.ndarray0: If the index is not populated.

Returns:

np.ndarray1: A tuple containing an array of scores and a list of route names.

​

aquery

async def aquery(
    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]]

Asynchronously search the index for the query vector and return the top_k results.

Arguments:

• vector (np.ndarray): The query vector to search for.
• top_k (int, optional): The number of top results to return, defaults to 5.
• route_filter (Optional[List[str]], optional): A list of route names to filter the search results, defaults to None.
• kwargs (Any): Additional keyword arguments for the query, including sparse_vector.
• sparse_vector (Optional[dict]): An optional sparse vector to include in the query.

Raises:

• np.ndarray0: If the index is not populated.

Returns:

np.ndarray1: A tuple containing an array of scores and a list of route names.

​

aget_routes

async def aget_routes() -> list[tuple]

Asynchronously get a list of route and utterance objects currently

stored in the index.

Returns:

List[Tuple]: A list of (route_name, utterance) objects.

​

delete_index

def delete_index()

Delete the index.

Returns:

None: None

​

adelete_index

async def adelete_index()

Asynchronously delete the index.

​

__len__

def __len__()

Returns the total number of vectors in the index. If the index is not initialized

returns 0.

Returns:

int: The total number of vectors.