codekingpro/portable-devtools
114k
1from __future__ import annotations2 3import base644import logging5import os6import uuid7from copy import deepcopy8from typing import (9 TYPE_CHECKING,10 Any,11 Callable,12 Dict,13 Iterable,14 List,15 Literal,16 Optional,17 Sized,18 Tuple,19 Type,20 Union,21 get_args,22)23 24import numpy as np25from langchain_core._api.deprecation import deprecated26from langchain_core.documents import Document27from langchain_core.embeddings import Embeddings28from langchain_core.vectorstores import VectorStore29 30from langchain_community.vectorstores.utils import maximal_marginal_relevance31 32if TYPE_CHECKING:33 import vdms34 35 36DISTANCE_METRICS = Literal[37 "L2", # Euclidean Distance38 "IP", # Inner Product39]40AVAILABLE_DISTANCE_METRICS: List[DISTANCE_METRICS] = list(get_args(DISTANCE_METRICS))41ENGINES = Literal[42 "TileDBDense", # TileDB Dense43 "TileDBSparse", # TileDB Sparse44 "FaissFlat", # FAISS IndexFlat45 "FaissIVFFlat", # FAISS IndexIVFFlat46 "Flinng", # FLINNG47]48AVAILABLE_ENGINES: List[ENGINES] = list(get_args(ENGINES))49DEFAULT_COLLECTION_NAME = "langchain"50DEFAULT_INSERT_BATCH_SIZE = 3251# Number of Documents to return.52DEFAULT_K = 353# Number of Documents to fetch to pass to knn when filters applied.54DEFAULT_FETCH_K = DEFAULT_K * 555DEFAULT_PROPERTIES = ["_distance", "id", "content"]56INVALID_DOC_METADATA_KEYS = ["_distance", "content", "blob"]57INVALID_METADATA_VALUE = ["Missing property", None, {}] # type: List58 59 60logger = logging.getLogger(__name__)61 62 63def _len_check_if_sized(x: Any, y: Any, x_name: str, y_name: str) -> None:64 """65 Check that sizes of two variables are the same66 67 Args:68 x: Variable to compare69 y: Variable to compare70 x_name: Name for variable x71 y_name: Name for variable y72 """73 if isinstance(x, Sized) and isinstance(y, Sized) and len(x) != len(y):74 raise ValueError(75 f"{x_name} and {y_name} expected to be equal length but "76 f"len({x_name})={len(x)} and len({y_name})={len(y)}"77 )78 return79 80 81def _results_to_docs(results: Any) -> List[Document]:82 return [doc for doc, _ in _results_to_docs_and_scores(results)]83 84 85def _results_to_docs_and_scores(results: Any) -> List[Tuple[Document, float]]:86 final_res: List[Any] = []87 try:88 responses, blobs = results[0]89 if (90 len(responses) > 091 and "FindDescriptor" in responses[0]92 and "entities" in responses[0]["FindDescriptor"]93 ):94 result_entities = responses[0]["FindDescriptor"]["entities"]95 # result_blobs = blobs96 for ent in result_entities:97 distance = round(ent["_distance"], 10)98 txt_contents = ent["content"]99 for p in INVALID_DOC_METADATA_KEYS:100 if p in ent:101 del ent[p]102 props = {103 mkey: mval104 for mkey, mval in ent.items()105 if mval not in INVALID_METADATA_VALUE106 }107 108 final_res.append(109 (110 Document(page_content=txt_contents, metadata=props),111 distance,112 )113 )114 except Exception as e:115 logger.warning(f"No results returned. Error while parsing results: {e}")116 return final_res117 118 119def VDMS_Client(host: str = "localhost", port: int = 55555) -> vdms.vdms:120 """VDMS client for the VDMS server.121 122 Args:123 host: IP or hostname of VDMS server124 port: Port to connect to VDMS server125 """126 try:127 import vdms128 except ImportError:129 raise ImportError(130 "Could not import vdms python package. "131 "Please install it with `pip install vdms."132 )133 134 client = vdms.vdms()135 client.connect(host, port)136 return client137 138 139@deprecated(since="0.3.18", removal="1.0.0", alternative_import="langchain_vdms.VDMS")140class VDMS(VectorStore):141 """Intel Lab's VDMS for vector-store workloads.142 143 To use, you should have both:144 - the ``vdms`` python package installed145 - a host (str) and port (int) associated with a deployed VDMS Server146 147 Visit https://github.com/IntelLabs/vdms/wiki more information.148 149 IT IS HIGHLY SUGGESTED TO NORMALIZE YOUR DATA.150 151 Args:152 client: VDMS Client used to connect to VDMS server153 collection_name: Name of data collection [Default: langchain]154 distance_strategy: Method used to calculate distances. VDMS supports155 "L2" (euclidean distance) or "IP" (inner product) [Default: L2]156 engine: Underlying implementation for indexing and computing distances.157 VDMS supports TileDBDense, TileDBSparse, FaissFlat, FaissIVFFlat,158 and Flinng [Default: FaissFlat]159 embedding: Any embedding function implementing160 `langchain_core.embeddings.Embeddings` interface.161 relevance_score_fn: Function for obtaining relevance score162 163 Example:164 .. code-block:: python165 166 from langchain_huggingface import HuggingFaceEmbeddings167 from langchain_community.vectorstores.vdms import VDMS, VDMS_Client168 169 model_name = "sentence-transformers/all-mpnet-base-v2"170 vectorstore = VDMS(171 client=VDMS_Client("localhost", 55555),172 embedding=HuggingFaceEmbeddings(model_name=model_name),173 collection_name="langchain-demo",174 distance_strategy="L2",175 engine="FaissFlat",176 )177 """178 179 def __init__(180 self,181 client: vdms.vdms,182 *,183 embedding: Optional[Embeddings] = None,184 collection_name: str = DEFAULT_COLLECTION_NAME, # DescriptorSet name185 distance_strategy: DISTANCE_METRICS = "L2",186 engine: ENGINES = "FaissFlat",187 relevance_score_fn: Optional[Callable[[float], float]] = None,188 embedding_dimensions: Optional[int] = None,189 ) -> None:190 # Check required parameters191 self._client = client192 self.similarity_search_engine = engine193 self.distance_strategy = distance_strategy194 self.embedding = embedding195 self._check_required_inputs(collection_name, embedding_dimensions)196 197 # Update other parameters198 self.override_relevance_score_fn = relevance_score_fn199 200 # Initialize collection201 self._collection_name = self.add_set(202 collection_name,203 engine=self.similarity_search_engine,204 metric=self.distance_strategy,205 )206 207 @property208 def embeddings(self) -> Optional[Embeddings]:209 return self.embedding210 211 def _embed_documents(self, texts: List[str]) -> List[List[float]]:212 if isinstance(self.embedding, Embeddings):213 return self.embedding.embed_documents(texts)214 else:215 p_str = "Must provide `embedding` which is expected"216 p_str += " to be an Embeddings object"217 raise ValueError(p_str)218 219 def _embed_video(self, paths: List[str], **kwargs: Any) -> List[List[float]]:220 if self.embedding is not None and hasattr(self.embedding, "embed_video"):221 return self.embedding.embed_video(paths=paths, **kwargs)222 else:223 raise ValueError(224 "Must provide `embedding` which has attribute `embed_video`"225 )226 227 def _embed_image(self, uris: List[str]) -> List[List[float]]:228 if self.embedding is not None and hasattr(self.embedding, "embed_image"):229 return self.embedding.embed_image(uris=uris)230 else:231 raise ValueError(232 "Must provide `embedding` which has attribute `embed_image`"233 )234 235 def _embed_query(self, text: str) -> List[float]:236 if isinstance(self.embedding, Embeddings):237 return self.embedding.embed_query(text)238 else:239 raise ValueError(240 "Must provide `embedding` which is expected to be an Embeddings object"241 )242 243 def _select_relevance_score_fn(self) -> Callable[[float], float]:244 """245 The 'correct' relevance function246 may differ depending on a few things, including:247 - the distance / similarity metric used by the VectorStore248 - the scale of your embeddings (OpenAI's are unit normed. Many others are not!)249 - embedding dimensionality250 - etc.251 """252 if self.override_relevance_score_fn is not None:253 return self.override_relevance_score_fn254 255 # Default strategy is to rely on distance strategy provided256 # in vectorstore constructor257 if self.distance_strategy.lower() in ["ip", "l2"]:258 return lambda x: x259 else:260 raise ValueError(261 "No supported normalization function"262 f" for distance_strategy of {self.distance_strategy}."263 "Consider providing relevance_score_fn to VDMS constructor."264 )265 266 def _similarity_search_with_relevance_scores(267 self,268 query: str,269 k: int = DEFAULT_K,270 fetch_k: int = DEFAULT_FETCH_K,271 filter: Optional[Dict[str, Any]] = None,272 **kwargs: Any,273 ) -> List[Tuple[Document, float]]:274 """Return docs and their similarity scores on a scale from 0 to 1."""275 if self.override_relevance_score_fn is None:276 kwargs["normalize_distance"] = True277 docs_and_scores = self.similarity_search_with_score(278 query=query,279 k=k,280 fetch_k=fetch_k,281 filter=filter,282 **kwargs,283 )284 285 docs_and_rel_scores: List[Any] = []286 for doc, score in docs_and_scores:287 if self.override_relevance_score_fn is None:288 docs_and_rel_scores.append((doc, score))289 else:290 docs_and_rel_scores.append(291 (292 doc,293 self.override_relevance_score_fn(score),294 )295 )296 return docs_and_rel_scores297 298 def add(299 self,300 collection_name: str,301 texts: List[str],302 embeddings: List[List[float]],303 metadatas: Optional[Union[List[None], List[Dict[str, Any]]]] = None,304 ids: Optional[List[str]] = None,305 ) -> List:306 _len_check_if_sized(texts, embeddings, "texts", "embeddings")307 308 metadatas = metadatas if metadatas is not None else [None for _ in texts]309 _len_check_if_sized(texts, metadatas, "texts", "metadatas")310 311 ids = ids if ids is not None else [str(uuid.uuid4()) for _ in texts]312 _len_check_if_sized(texts, ids, "texts", "ids")313 314 all_queries: List[Any] = []315 all_blobs: List[Any] = []316 inserted_ids: List[Any] = []317 for meta, emb, doc, id in zip(metadatas, embeddings, texts, ids):318 query, blob = self.__get_add_query(319 collection_name, metadata=meta, embedding=emb, document=doc, id=id320 )321 322 if blob is not None:323 all_queries.append(query)324 all_blobs.append(blob)325 inserted_ids.append(id)326 327 response, response_array = self.__run_vdms_query(all_queries, all_blobs)328 329 return inserted_ids330 331 def add_set(332 self,333 collection_name: str,334 engine: ENGINES = "FaissFlat",335 metric: DISTANCE_METRICS = "L2",336 ) -> str:337 query = _add_descriptorset(338 "AddDescriptorSet",339 collection_name,340 self.embedding_dimension,341 engine=getattr(engine, "value", engine),342 metric=getattr(metric, "value", metric),343 )344 345 response, _ = self.__run_vdms_query([query])346 347 if "FailedCommand" in response[0]:348 raise ValueError(f"Failed to add collection {collection_name}")349 350 return collection_name351 352 def __delete(353 self,354 collection_name: str,355 ids: Union[None, List[str]] = None,356 constraints: Union[None, Dict[str, Any]] = None,357 ) -> bool:358 """359 Deletes entire collection if id is not provided360 """361 all_queries: List[Any] = []362 all_blobs: List[Any] = []363 364 collection_properties = self.__get_properties(collection_name)365 results = {"list": collection_properties}366 367 if constraints is None:368 constraints = {"_deletion": ["==", 1]}369 else:370 constraints["_deletion"] = ["==", 1]371 372 if ids is not None:373 constraints["id"] = ["==", ids[0]] # if len(ids) > 1 else ids[0]]374 375 query = _add_descriptor(376 "FindDescriptor",377 collection_name,378 label=None,379 ref=None,380 props=None,381 link=None,382 k_neighbors=None,383 constraints=constraints,384 results=results,385 )386 387 all_queries.append(query)388 response, response_array = self.__run_vdms_query(all_queries, all_blobs)389 390 # Update/store indices after deletion391 query = _add_descriptorset(392 "FindDescriptorSet", collection_name, storeIndex=True393 )394 responseSet, _ = self.__run_vdms_query([query], all_blobs)395 return "FindDescriptor" in response[0]396 397 def __get_add_query(398 self,399 collection_name: str,400 metadata: Optional[Any] = None,401 embedding: Union[List[float], None] = None,402 document: Optional[Any] = None,403 id: Optional[str] = None,404 ) -> Tuple[Dict[str, Dict[str, Any]], Union[bytes, None]]:405 if id is None:406 props: Dict[str, Any] = {}407 else:408 props = {"id": id}409 id_exists, query = _check_descriptor_exists_by_id(410 self._client, collection_name, id411 )412 if id_exists:413 skipped_value = {414 prop_key: prop_val[-1]415 for prop_key, prop_val in query["FindDescriptor"][416 "constraints"417 ].items()418 }419 pstr = f"[!] Embedding with id ({id}) exists in DB;"420 pstr += "Therefore, skipped and not inserted"421 print(pstr) # noqa: T201422 print(f"\tSkipped values are: {skipped_value}") # noqa: T201423 return query, None424 425 if metadata:426 props.update(metadata)427 if document not in [None, ""]:428 props["content"] = document429 430 for k in props.keys():431 if k not in self.collection_properties:432 self.collection_properties.append(k)433 434 query = _add_descriptor(435 "AddDescriptor",436 collection_name,437 label=None,438 ref=None,439 props=props,440 link=None,441 k_neighbors=None,442 constraints=None,443 results=None,444 )445 446 blob = embedding2bytes(embedding)447 448 return (449 query,450 blob,451 )452 453 def __get_properties(454 self,455 collection_name: str,456 unique_entity: Optional[bool] = False,457 deletion: Optional[bool] = False,458 ) -> List[str]:459 find_query = _find_property_entity(460 collection_name, unique_entity=unique_entity, deletion=deletion461 )462 response, response_blob = self.__run_vdms_query([find_query])463 if len(response_blob) > 0:464 collection_properties = _bytes2str(response_blob[0]).split(",")465 else:466 collection_properties = deepcopy(DEFAULT_PROPERTIES)467 return collection_properties468 469 def __run_vdms_query(470 self,471 all_queries: List[Dict],472 all_blobs: Optional[List] = [],473 print_last_response: Optional[bool] = False,474 ) -> Tuple[Any, Any]:475 response, response_array = self._client.query(all_queries, all_blobs)476 477 _ = _check_valid_response(all_queries, response)478 if print_last_response:479 self._client.print_last_response()480 return response, response_array481 482 def __update(483 self,484 collection_name: str,485 ids: List[str],486 documents: List[str],487 embeddings: List[List[float]],488 metadatas: Optional[Union[List[None], List[Dict[str, Any]]]] = None,489 ) -> None:490 """491 Updates (find, delete, add) a collection based on id.492 If more than one collection returned with id, error occuers493 """494 _len_check_if_sized(ids, documents, "ids", "documents")495 496 _len_check_if_sized(ids, embeddings, "ids", "embeddings")497 498 metadatas = metadatas if metadatas is not None else [None for _ in ids]499 _len_check_if_sized(ids, metadatas, "ids", "metadatas")500 501 orig_props = self.__get_properties(collection_name)502 503 updated_ids: List[Any] = []504 for meta, emb, doc, id in zip(metadatas, embeddings, documents, ids):505 results = {"list": self.collection_properties}506 507 constraints = {"_deletion": ["==", 1]}508 509 if id is not None:510 constraints["id"] = ["==", id]511 512 query = _add_descriptor(513 "FindDescriptor",514 collection_name,515 label=None,516 ref=None,517 props=None,518 link=None,519 k_neighbors=None,520 constraints=constraints,521 results=results,522 )523 524 response, response_array = self.__run_vdms_query([query])525 526 query, blob = self.__get_add_query(527 collection_name,528 metadata=meta,529 embedding=emb,530 document=doc,531 id=id,532 )533 if blob is not None:534 response, response_array = self.__run_vdms_query([query], [blob])535 updated_ids.append(id)536 537 self.__update_properties(538 collection_name, orig_props, self.collection_properties539 )540 541 def __update_properties(542 self,543 collection_name: str,544 current_collection_properties: List,545 new_collection_properties: Optional[List],546 ) -> None:547 if new_collection_properties is not None:548 old_collection_properties = deepcopy(current_collection_properties)549 for prop in new_collection_properties:550 if prop not in current_collection_properties:551 current_collection_properties.append(prop)552 553 if current_collection_properties != old_collection_properties:554 all_queries, blob_arr = _build_property_query(555 collection_name,556 command_type="update",557 all_properties=current_collection_properties,558 )559 response, _ = self.__run_vdms_query(all_queries, [blob_arr])560 561 def add_images(562 self,563 uris: List[str],564 metadatas: Optional[List[dict]] = None,565 ids: Optional[List[str]] = None,566 batch_size: int = DEFAULT_INSERT_BATCH_SIZE,567 add_path: Optional[bool] = True,568 **kwargs: Any,569 ) -> List[str]:570 """Run more images through the embeddings and add to the vectorstore.571 572 Images are added as embeddings (AddDescriptor) instead of separate573 entity (AddImage) within VDMS to leverage similarity search capability574 575 Args:576 uris: List of paths to the images to add to the vectorstore.577 metadatas: Optional list of metadatas associated with the images.578 ids: Optional list of unique IDs.579 batch_size (int): Number of concurrent requests to send to the server.580 add_path: Bool to add image path as metadata581 582 Returns:583 List of ids from adding images into the vectorstore.584 """585 # Map from uris to blobs to base64586 b64_texts = [self.encode_image(image_path=uri) for uri in uris]587 588 if add_path and metadatas:589 for midx, uri in enumerate(uris):590 metadatas[midx]["image_path"] = uri591 elif add_path:592 metadatas = []593 for uri in uris:594 metadatas.append({"image_path": uri})595 596 # Populate IDs597 ids = ids if ids is not None else [str(uuid.uuid4()) for _ in uris]598 599 # Set embeddings600 embeddings = self._embed_image(uris=uris)601 602 if metadatas is None:603 metadatas = [{} for _ in uris]604 else:605 metadatas = [_validate_vdms_properties(m) for m in metadatas]606 607 self.add_from(608 texts=b64_texts,609 embeddings=embeddings,610 ids=ids,611 metadatas=metadatas,612 batch_size=batch_size,613 **kwargs,614 )615 return ids616 617 def add_videos(618 self,619 paths: List[str],620 texts: Optional[List[str]] = None,621 metadatas: Optional[List[dict]] = None,622 ids: Optional[List[str]] = None,623 batch_size: int = 1,624 add_path: Optional[bool] = True,625 **kwargs: Any,626 ) -> List[str]:627 """Run videos through the embeddings and add to the vectorstore.628 629 Videos are added as embeddings (AddDescriptor) instead of separate630 entity (AddVideo) within VDMS to leverage similarity search capability631 632 Args:633 paths: List of paths to the videos to add to the vectorstore.634 metadatas: Optional list of text associated with the videos.635 metadatas: Optional list of metadatas associated with the videos.636 ids: Optional list of unique IDs.637 batch_size (int): Number of concurrent requests to send to the server.638 add_path: Bool to add video path as metadata639 640 Returns:641 List of ids from adding videos into the vectorstore.642 """643 if texts is None:644 texts = ["" for _ in paths]645 646 if add_path and metadatas:647 for midx, path in enumerate(paths):648 metadatas[midx]["video_path"] = path649 elif add_path:650 metadatas = []651 for path in paths:652 metadatas.append({"video_path": path})653 654 # Populate IDs655 ids = ids if ids is not None else [str(uuid.uuid4()) for _ in paths]656 657 # Set embeddings658 embeddings = self._embed_video(paths=paths, **kwargs)659 660 if metadatas is None:661 metadatas = [{} for _ in paths]662 663 self.add_from(664 texts=texts,665 embeddings=embeddings,666 ids=ids,667 metadatas=metadatas,668 batch_size=batch_size,669 **kwargs,670 )671 return ids672 673 def add_texts(674 self,675 texts: Iterable[str],676 metadatas: Optional[List[dict]] = None,677 ids: Optional[List[str]] = None,678 batch_size: int = DEFAULT_INSERT_BATCH_SIZE,679 **kwargs: Any,680 ) -> List[str]:681 """Run more texts through the embeddings and add to the vectorstore.682 683 Args:684 texts: List of strings to add to the vectorstore.685 metadatas: Optional list of metadatas associated with the texts.686 ids: Optional list of unique IDs.687 batch_size (int): Number of concurrent requests to send to the server.688 689 Returns:690 List of ids from adding the texts into the vectorstore.691 """692 693 texts = list(texts)694 if ids is None:695 ids = [str(uuid.uuid4()) for _ in texts]696 697 embeddings = self._embed_documents(texts)698 699 if metadatas is None:700 metadatas = [{} for _ in texts]701 else:702 metadatas = [_validate_vdms_properties(m) for m in metadatas]703 704 inserted_ids = self.add_from(705 texts=texts,706 embeddings=embeddings,707 ids=ids,708 metadatas=metadatas,709 batch_size=batch_size,710 **kwargs,711 )712 return inserted_ids713 714 def add_from(715 self,716 texts: List[str],717 embeddings: List[List[float]],718 ids: List[str],719 metadatas: Optional[List[dict]] = None,720 batch_size: int = DEFAULT_INSERT_BATCH_SIZE,721 **kwargs: Any,722 ) -> List[str]:723 # Get initial properties724 orig_props = self.__get_properties(self._collection_name)725 inserted_ids: List[str] = []726 for start_idx in range(0, len(texts), batch_size):727 end_idx = min(start_idx + batch_size, len(texts))728 729 batch_texts = texts[start_idx:end_idx]730 batch_embedding_vectors = embeddings[start_idx:end_idx]731 batch_ids = ids[start_idx:end_idx]732 if metadatas:733 batch_metadatas = metadatas[start_idx:end_idx]734 735 result = self.add(736 self._collection_name,737 embeddings=batch_embedding_vectors,738 texts=batch_texts,739 metadatas=batch_metadatas,740 ids=batch_ids,741 )742 743 inserted_ids.extend(result)744 745 # Update Properties746 self.__update_properties(747 self._collection_name, orig_props, self.collection_properties748 )749 return inserted_ids750 751 def _check_required_inputs(752 self, collection_name: str, embedding_dimensions: Union[int, None]753 ) -> None:754 # Check connection to client755 if not self._client.is_connected():756 raise ValueError(757 "VDMS client must be connected to a VDMS server."758 + "Please use VDMS_Client to establish a connection"759 )760 761 # Check Distance Metric762 if self.distance_strategy not in AVAILABLE_DISTANCE_METRICS:763 raise ValueError("distance_strategy must be either 'L2' or 'IP'")764 765 # Check Engines766 if self.similarity_search_engine not in AVAILABLE_ENGINES:767 raise ValueError(768 "engine must be either 'TileDBDense', 'TileDBSparse', "769 + "'FaissFlat', 'FaissIVFFlat', or 'Flinng'"770 )771 772 # Check Embedding Func is provided and store dimension size773 if self.embedding is None:774 raise ValueError("Must provide embedding function")775 776 if embedding_dimensions is not None:777 self.embedding_dimension = embedding_dimensions778 elif self.embedding is not None and hasattr(self.embedding, "embed_query"):779 self.embedding_dimension = len(780 self._embed_query("This is a sample sentence.")781 )782 elif self.embedding is not None and (783 hasattr(self.embedding, "embed_image")784 or hasattr(self.embedding, "embed_video")785 ):786 if hasattr(self.embedding, "model"):787 try:788 self.embedding_dimension = (789 self.embedding.model.token_embedding.embedding_dim790 )791 except ValueError:792 raise ValueError(793 "Embedding dimension needed. Please define embedding_dimensions"794 )795 else:796 raise ValueError(797 "Embedding dimension needed. Please define embedding_dimensions"798 )799 800 # Check for properties801 current_props = self.__get_properties(collection_name)802 if hasattr(self, "collection_properties"):803 self.collection_properties.extend(current_props)804 else:805 self.collection_properties: List[str] = current_props806 807 def count(self, collection_name: str) -> int:808 all_queries: List[Any] = []809 all_blobs: List[Any] = []810 811 results = {"count": "", "list": ["id"]} # collection_properties}812 query = _add_descriptor(813 "FindDescriptor",814 collection_name,815 label=None,816 ref=None,817 props=None,818 link=None,819 k_neighbors=None,820 constraints=None,821 results=results,822 )823 824 all_queries.append(query)825 826 response, response_array = self.__run_vdms_query(all_queries, all_blobs)827 return response[0]["FindDescriptor"]["returned"]828 829 def decode_image(self, base64_image: str) -> bytes:830 return base64.b64decode(base64_image)831 832 def delete(833 self,834 ids: Optional[List[str]] = None,835 collection_name: Optional[str] = None,836 constraints: Optional[Dict] = None,837 **kwargs: Any,838 ) -> bool:839 """Delete by ID. These are the IDs in the vectorstore.840 841 Args:842 ids: List of ids to delete.843 844 Returns:845 Optional[bool]: True if deletion is successful,846 False otherwise, None if not implemented.847 """848 name = collection_name if collection_name is not None else self._collection_name849 return self.__delete(name, ids=ids, constraints=constraints)850 851 def get_k_candidates(852 self,853 setname: str,854 fetch_k: Optional[int],855 results: Optional[Dict[str, Any]] = None,856 all_blobs: Optional[List] = None,857 normalize: Optional[bool] = False,858 ) -> Tuple[List[Dict[str, Any]], List, float]:859 max_dist = 1860 command_str = "FindDescriptor"861 query = _add_descriptor(862 command_str,863 setname,864 k_neighbors=fetch_k,865 results=results,866 )867 response, response_array = self.__run_vdms_query([query], all_blobs)868 869 if normalize and command_str in response[0]:870 max_dist = response[0][command_str]["entities"][-1]["_distance"]871 872 return response, response_array, max_dist873 874 def get_descriptor_response(875 self,876 command_str: str,877 setname: str,878 k_neighbors: int = DEFAULT_K,879 fetch_k: int = DEFAULT_FETCH_K,880 constraints: Optional[dict] = None,881 results: Optional[Dict[str, Any]] = None,882 query_embedding: Optional[List[float]] = None,883 normalize_distance: bool = False,884 ) -> Tuple[List[Dict[str, Any]], List]:885 all_blobs: List[Any] = []886 blob = embedding2bytes(query_embedding)887 if blob is not None:888 all_blobs.append(blob)889 890 if constraints is None:891 # K results returned892 response, response_array, max_dist = self.get_k_candidates(893 setname, k_neighbors, results, all_blobs, normalize=normalize_distance894 )895 else:896 if results is None:897 results = {"list": ["id"]}898 elif "list" not in results:899 results["list"] = ["id"]900 elif "id" not in results["list"]:901 results["list"].append("id")902 903 # (1) Find docs satisfy constraints904 query = _add_descriptor(905 command_str,906 setname,907 constraints=constraints,908 results=results,909 )910 response, response_array = self.__run_vdms_query([query])911 if command_str in response[0] and response[0][command_str]["returned"] > 0:912 ids_of_interest = [913 ent["id"] for ent in response[0][command_str]["entities"]914 ]915 else:916 return [], []917 918 # (2) Find top fetch_k results919 response, response_array, max_dist = self.get_k_candidates(920 setname, fetch_k, results, all_blobs, normalize=normalize_distance921 )922 if command_str not in response[0] or (923 command_str in response[0] and response[0][command_str]["returned"] == 0924 ):925 return [], []926 927 # (3) Intersection of (1) & (2) using ids928 new_entities: List[Dict] = []929 for ent in response[0][command_str]["entities"]:930 if ent["id"] in ids_of_interest:931 new_entities.append(ent)932 if len(new_entities) == k_neighbors:933 break934 response[0][command_str]["entities"] = new_entities935 response[0][command_str]["returned"] = len(new_entities)936 if len(new_entities) < k_neighbors:937 p_str = "Returned items < k_neighbors; Try increasing fetch_k"938 print(p_str) # noqa: T201939 940 if normalize_distance:941 max_dist = 1.0 if max_dist in [0, np.inf] else max_dist942 for ent_idx, ent in enumerate(response[0][command_str]["entities"]):943 ent["_distance"] = ent["_distance"] / max_dist944 response[0][command_str]["entities"][ent_idx]["_distance"] = ent[945 "_distance"946 ]947 948 return response, response_array949 950 def encode_image(self, image_path: str) -> str:951 with open(image_path, "rb") as f:952 blob = f.read()953 return base64.b64encode(blob).decode("utf-8")954 955 @classmethod956 def from_documents(957 cls: Type[VDMS],958 documents: List[Document],959 embedding: Optional[Embeddings] = None,960 ids: Optional[List[str]] = None,961 batch_size: int = DEFAULT_INSERT_BATCH_SIZE,962 collection_name: str = DEFAULT_COLLECTION_NAME, # Add this line963 **kwargs: Any,964 ) -> VDMS:965 """Create a VDMS vectorstore from a list of documents.966 967 Args:968 collection_name (str): Name of the collection to create.969 documents (List[Document]): List of documents to add to vectorstore.970 embedding (Embeddings): Embedding function. Defaults to None.971 ids (Optional[List[str]]): List of document IDs. Defaults to None.972 batch_size (int): Number of concurrent requests to send to the server.973 974 Returns:975 VDMS: VDMS vectorstore.976 """977 client: vdms.vdms = kwargs["client"]978 979 return cls.from_texts(980 client=client,981 texts=[doc.page_content for doc in documents],982 metadatas=[doc.metadata for doc in documents],983 embedding=embedding,984 ids=ids,985 batch_size=batch_size,986 collection_name=collection_name,987 # **kwargs,988 )989 990 @classmethod991 def from_texts(992 cls: Type[VDMS],993 texts: List[str],994 embedding: Optional[Embeddings] = None,995 metadatas: Optional[List[dict]] = None,996 ids: Optional[List[str]] = None,997 batch_size: int = DEFAULT_INSERT_BATCH_SIZE,998 collection_name: str = DEFAULT_COLLECTION_NAME,999 **kwargs: Any,1000 ) -> VDMS:1001 """Create a VDMS vectorstore from a raw documents.1002 1003 Args:1004 texts (List[str]): List of texts to add to the collection.1005 embedding (Embeddings): Embedding function. Defaults to None.1006 metadatas (Optional[List[dict]]): List of metadatas. Defaults to None.1007 ids (Optional[List[str]]): List of document IDs. Defaults to None.1008 batch_size (int): Number of concurrent requests to send to the server.1009 collection_name (str): Name of the collection to create.1010 1011 Returns:1012 VDMS: VDMS vectorstore.1013 """1014 client: vdms.vdms = kwargs["client"]1015 vdms_collection = cls(1016 collection_name=collection_name,1017 embedding=embedding,1018 client=client,1019 # **kwargs,1020 )1021 if ids is None:1022 ids = [str(uuid.uuid4()) for _ in texts]1023 vdms_collection.add_texts(1024 texts=texts,1025 metadatas=metadatas,1026 ids=ids,1027 batch_size=batch_size, # **kwargs1028 )1029 return vdms_collection1030 1031 def get(1032 self,1033 collection_name: str,1034 constraints: Optional[Dict] = None,1035 limit: Optional[int] = None,1036 include: List[str] = ["metadata"],1037 ) -> Tuple[Any, Any]:1038 """Gets the collection.1039 Get embeddings and their associated data from the data store.1040 If no constraints provided returns all embeddings up to limit.1041 1042 Args:1043 constraints: A dict used to filter results by.1044 E.g. `{"color" : ["==", "red"], "price": [">", 4.00]}`. Optional.1045 limit: The number of documents to return. Optional.1046 include: A list of what to include in the results.1047 Can contain `"embeddings"`, `"metadatas"`, `"documents"`.1048 Ids are always included.1049 Defaults to `["metadatas", "documents"]`. Optional.1050 """1051 all_queries: List[Any] = []1052 all_blobs: List[Any] = []1053 1054 results: Dict[str, Any] = {"count": ""}1055 1056 if limit is not None:1057 results["limit"] = limit1058 1059 # Include metadata1060 if "metadata" in include:1061 collection_properties = self.__get_properties(collection_name)1062 results["list"] = collection_properties1063 1064 # Include embedding1065 if "embeddings" in include:1066 results["blob"] = True1067 1068 query = _add_descriptor(1069 "FindDescriptor",1070 collection_name,1071 k_neighbors=None,1072 constraints=constraints,1073 results=results,1074 )1075 1076 all_queries.append(query)1077 1078 response, response_array = self.__run_vdms_query(all_queries, all_blobs)1079 return response, response_array1080 1081 def max_marginal_relevance_search(1082 self,1083 query: str,1084 k: int = DEFAULT_K,1085 fetch_k: int = DEFAULT_FETCH_K,1086 lambda_mult: float = 0.5,1087 filter: Optional[Dict[str, List]] = None,1088 **kwargs: Any,1089 ) -> List[Document]:1090 """Return docs selected using the maximal marginal relevance.1091 Maximal marginal relevance optimizes for similarity to query AND diversity1092 among selected documents.1093 1094 Args:1095 query (str): Query to look up. Text or path for image or video.1096 k: Number of Documents to return. Defaults to 4.1097 fetch_k: Number of Documents to fetch to pass to MMR algorithm.1098 lambda_mult: Number between 0 and 1 that determines the degree1099 of diversity among the results with 0 corresponding1100 to maximum diversity and 1 to minimum diversity.1101 Defaults to 0.5.1102 filter (Optional[Dict[str, str]]): Filter by metadata. Defaults to None.1103 1104 Returns:1105 List of Documents selected by maximal marginal relevance.1106 """1107 if self.embedding is None:1108 raise ValueError(1109 "For MMR search, you must specify an embedding function oncreation."1110 )1111 1112 # embedding_vector: List[float] = self._embed_query(query)1113 embedding_vector: List[float]1114 if not os.path.isfile(query) and hasattr(self.embedding, "embed_query"):1115 embedding_vector = self._embed_query(query)1116 elif os.path.isfile(query) and hasattr(self.embedding, "embed_image"):1117 embedding_vector = self._embed_image(uris=[query])[0]1118 elif os.path.isfile(query) and hasattr(self.embedding, "embed_video"):1119 embedding_vector = self._embed_video(paths=[query])[0]1120 else:1121 error_msg = f"Could not generate embedding for query '{query}'."1122 error_msg += "If using path for image or video, verify embedding model "1123 error_msg += "has callable functions 'embed_image' or 'embed_video'."1124 raise ValueError(error_msg)1125 1126 docs = self.max_marginal_relevance_search_by_vector(1127 embedding_vector,1128 k,1129 fetch_k,1130 lambda_mult=lambda_mult,1131 filter=filter,1132 )1133 return docs1134 1135 def max_marginal_relevance_search_by_vector(1136 self,1137 embedding: List[float],1138 k: int = DEFAULT_K,1139 fetch_k: int = DEFAULT_FETCH_K,1140 lambda_mult: float = 0.5,1141 filter: Optional[Dict[str, List]] = None,1142 **kwargs: Any,1143 ) -> List[Document]:1144 """Return docs selected using the maximal marginal relevance.1145 Maximal marginal relevance optimizes for similarity to query AND diversity1146 among selected documents.1147 1148 Args:1149 embedding: Embedding to look up documents similar to.1150 k: Number of Documents to return. Defaults to 4.1151 fetch_k: Number of Documents to fetch to pass to MMR algorithm.1152 lambda_mult: Number between 0 and 1 that determines the degree1153 of diversity among the results with 0 corresponding1154 to maximum diversity and 1 to minimum diversity.1155 Defaults to 0.5.1156 filter (Optional[Dict[str, str]]): Filter by metadata. Defaults to None.1157 1158 Returns:1159 List of Documents selected by maximal marginal relevance.1160 """1161 results = self.query_collection_embeddings(1162 query_embeddings=[embedding],1163 n_results=fetch_k,1164 filter=filter,1165 include=["metadatas", "documents", "distances", "embeddings"],1166 )1167 1168 if len(results[0][1]) == 0:1169 # No results returned1170 return []1171 else:1172 embedding_list = [1173 list(_bytes2embedding(result)) for result in results[0][1]1174 ]1175 1176 mmr_selected = maximal_marginal_relevance(1177 np.array(embedding, dtype=np.float32),1178 embedding_list,1179 k=k,1180 lambda_mult=lambda_mult,1181 )1182 1183 candidates = _results_to_docs(results)1184 1185 selected_results = [1186 r for i, r in enumerate(candidates) if i in mmr_selected1187 ]1188 return selected_results1189 1190 def max_marginal_relevance_search_with_score(1191 self,1192 query: str,1193 k: int = DEFAULT_K,1194 fetch_k: int = DEFAULT_FETCH_K,1195 lambda_mult: float = 0.5,1196 filter: Optional[Dict[str, List]] = None,1197 **kwargs: Any,1198 ) -> List[Tuple[Document, float]]:1199 """Return docs selected using the maximal marginal relevance.1200 Maximal marginal relevance optimizes for similarity to query AND diversity