a h*@sddlmZddlmZmZddlZddlZddlm Z ddlm Z m Z dddd d Z dddd d Z dddd dZdddddZeddddddZeddddddZddddddZddddddZdddd d!d"Zdd#d$d%d&ZdS)') annotations)AnyoverloadN) coo_matrix)Tensordevicezlist | np.ndarray | Tensorr)areturncCsht|trz%_convert_to_tensor..cSsg|]}|jtjdqS)Zdtype)coalescetotorchfloat32r rrr rz&_convert_to_tensor..r) r listallrstacktensorrr rrrrrr_convert_to_tensor s    rcCs|dkr|d}|S)z If the tensor `a` is 1-dimensional, it is unsqueezed to add a batch dimension. Args: a (Tensor): The input tensor. Returns: Tensor: The tensor with a batch dimension. r)dim unsqueezerrrr_convert_to_batch$s  r#cCs"t|}|dkr|d}|S)a Converts the input data to a tensor with a batch dimension. Handles lists of sparse tensors by stacking them. Args: a (Union[list, np.ndarray, Tensor]): The input data to be converted. Returns: Tensor: The converted tensor with a batch dimension. r r)rr!r"rrrr_convert_to_batch_tensor3s   r$) embeddingsr cCs|jstjjj|dddS|}||}}tj| d|j d}| d|d|dt | d|d}|dk}|}||||<t||| S)z Normalizes the embeddings matrix, so that each sentence embedding has unit length. Args: embeddings (Tensor): The input embeddings matrix. Returns: Tensor: The normalized embeddings matrix. r )pr!rr)r rnnZ functional normalizerindicesvaluesZzerossizerZ index_add_sqrtZ index_selectcloneZsparse_coo_tensor)r%r+r,Z row_normsmaskZnormalized_valuesrrrnormalize_embeddingsDs r1z np.ndarrayz int | None)r% truncate_dimr cCsdSr rr%r2rrrtruncate_embeddingsasr4z torch.TensorcCsdSr rr3rrrr4esznp.ndarray | torch.TensorcCs|dd|fS)a Truncates the embeddings matrix. Args: embeddings (Union[np.ndarray, torch.Tensor]): Embeddings to truncate. truncate_dim (Optional[int]): The dimension to truncate sentence embeddings to. `None` does no truncation. Example: >>> from sentence_transformers import SentenceTransformer >>> from sentence_transformers.util import truncate_embeddings >>> model = SentenceTransformer("tomaarsen/mpnet-base-nli-matryoshka") >>> embeddings = model.encode(["It's so nice outside!", "Today is a beautiful day.", "He drove to work earlier"]) >>> embeddings.shape (3, 768) >>> model.similarity(embeddings, embeddings) tensor([[1.0000, 0.8100, 0.1426], [0.8100, 1.0000, 0.2121], [0.1426, 0.2121, 1.0000]]) >>> truncated_embeddings = truncate_embeddings(embeddings, 128) >>> truncated_embeddings.shape >>> model.similarity(truncated_embeddings, truncated_embeddings) tensor([[1.0000, 0.8092, 0.1987], [0.8092, 1.0000, 0.2716], [0.1987, 0.2716, 1.0000]]) Returns: Union[np.ndarray, torch.Tensor]: Truncated embeddings. .Nrr3rrrr4is)r%max_active_dimsr c Cs|dur |St|tjr"t|}|j\}}|j}tjt|t ||dd\}}tj |tj d}tj ||d ddt ||}d|||f<d||<|S) a Keeps only the top-k values (in absolute terms) for each embedding and creates a sparse tensor. Args: embeddings (Union[np.ndarray, torch.Tensor]): Embeddings to sparsify by keeping only top_k values. max_active_dims (int): Number of values to keep as non-zeros per embedding. Returns: torch.Tensor: A sparse tensor containing only the top-k values per embedding. Nr )kr!rr(Tr)r npZndarrayrrshaperZtopkabsminZ zeros_likeboolZaranger"expandflatten) r%r5Z batch_sizer!r_Z top_indicesr0Z batch_indicesrrrselect_max_active_dimss     " r@zdict[str, Any]r)batch target_devicer cCs.|D]$}t||tr|||||<q|S)au Send a PyTorch batch (i.e., a dictionary of string keys to Tensors) to a device (e.g. "cpu", "cuda", "mps"). Args: batch (Dict[str, Tensor]): The batch to send to the device. target_device (torch.device): The target device (e.g. "cpu", "cuda", "mps"). Returns: Dict[str, Tensor]: The batch with tensors sent to the target device. )r rr)rArBkeyrrrbatch_to_devices rDr)rr cCsF|}|}|}t||d|dff|jdS)Nrr )r9)rr+cpunumpyr,rr9)rr+r,rrr to_scipy_coosrG) __future__rtypingrrrFr8rZ scipy.sparserrrrr#r$r1r4r@rDrGrrrrs"   "