codekingpro/portable-devtools

1from __future__ import annotations2 3import datetime4import os5from typing import (6    TYPE_CHECKING,7    Any,8    Callable,9    Dict,10    Iterable,11    List,12    Optional,13    Tuple,14)15from uuid import uuid416 17import numpy as np18from langchain_core._api import deprecated19from langchain_core.documents import Document20from langchain_core.embeddings import Embeddings21from langchain_core.vectorstores import VectorStore22 23from langchain_community.vectorstores.utils import maximal_marginal_relevance24 25if TYPE_CHECKING:26    import weaviate27 28 29def _default_schema(index_name: str, text_key: str) -> Dict:30    return {31        "class": index_name,32        "properties": [33            {34                "name": text_key,35                "dataType": ["text"],36            }37        ],38    }39 40 41def _create_weaviate_client(42    url: Optional[str] = None,43    api_key: Optional[str] = None,44    **kwargs: Any,45) -> weaviate.Client:46    try:47        import weaviate48    except ImportError:49        raise ImportError(50            "Could not import weaviate python  package. "51            "Please install it with `pip install weaviate-client`"52        )53    url = url or os.environ.get("WEAVIATE_URL")54    api_key = api_key or os.environ.get("WEAVIATE_API_KEY")55    auth = weaviate.auth.AuthApiKey(api_key=api_key) if api_key else None56    return weaviate.Client(url=url, auth_client_secret=auth, **kwargs)57 58 59def _default_score_normalizer(val: float) -> float:60    return 1 - 1 / (1 + np.exp(val))61 62 63def _json_serializable(value: Any) -> Any:64    if isinstance(value, datetime.datetime):65        return value.isoformat()66    return value67 68 69@deprecated(70    since="0.3.18",71    removal="1.0",72    alternative_import="langchain_weaviate.WeaviateVectorStore",73)74class Weaviate(VectorStore):75    """`Weaviate` vector store.76 77    To use, you should have the ``weaviate-client`` python package installed.78 79    Example:80        .. code-block:: python81 82            import weaviate83            from langchain_community.vectorstores import Weaviate84 85            client = weaviate.Client(url=os.environ["WEAVIATE_URL"], ...)86            weaviate = Weaviate(client, index_name, text_key)87 88    """89 90    def __init__(91        self,92        client: Any,93        index_name: str,94        text_key: str,95        embedding: Optional[Embeddings] = None,96        attributes: Optional[List[str]] = None,97        relevance_score_fn: Optional[98            Callable[[float], float]99        ] = _default_score_normalizer,100        by_text: bool = True,101    ):102        """Initialize with Weaviate client."""103        try:104            import weaviate105        except ImportError:106            raise ImportError(107                "Could not import weaviate python package. "108                "Please install it with `pip install weaviate-client`."109            )110        if not isinstance(client, weaviate.Client):111            raise ValueError(112                f"client should be an instance of weaviate.Client, got {type(client)}"113            )114        self._client = client115        self._index_name = index_name116        self._embedding = embedding117        self._text_key = text_key118        self._query_attrs = [self._text_key]119        self.relevance_score_fn = relevance_score_fn120        self._by_text = by_text121        if attributes is not None:122            self._query_attrs.extend(attributes)123 124    @property125    def embeddings(self) -> Optional[Embeddings]:126        return self._embedding127 128    def _select_relevance_score_fn(self) -> Callable[[float], float]:129        return (130            self.relevance_score_fn131            if self.relevance_score_fn132            else _default_score_normalizer133        )134 135    def add_texts(136        self,137        texts: Iterable[str],138        metadatas: Optional[List[dict]] = None,139        **kwargs: Any,140    ) -> List[str]:141        """Upload texts with metadata (properties) to Weaviate."""142        from weaviate.util import get_valid_uuid143 144        ids = []145        embeddings: Optional[List[List[float]]] = None146        if self._embedding:147            if not isinstance(texts, list):148                texts = list(texts)149            embeddings = self._embedding.embed_documents(texts)150 151        with self._client.batch as batch:152            for i, text in enumerate(texts):153                data_properties = {self._text_key: text}154                if metadatas is not None:155                    for key, val in metadatas[i].items():156                        data_properties[key] = _json_serializable(val)157 158                # Allow for ids (consistent w/ other methods)159                # # Or uuids (backwards compatible w/ existing arg)160                # If the UUID of one of the objects already exists161                # then the existing object will be replaced by the new object.162                _id = get_valid_uuid(uuid4())163                if "uuids" in kwargs:164                    _id = kwargs["uuids"][i]165                elif "ids" in kwargs:166                    _id = kwargs["ids"][i]167 168                batch.add_data_object(169                    data_object=data_properties,170                    class_name=self._index_name,171                    uuid=_id,172                    vector=embeddings[i] if embeddings else None,173                    tenant=kwargs.get("tenant"),174                )175                ids.append(_id)176        return ids177 178    def similarity_search(179        self, query: str, k: int = 4, **kwargs: Any180    ) -> List[Document]:181        """Return docs most similar to query.182 183        Args:184            query: Text to look up documents similar to.185            k: Number of Documents to return. Defaults to 4.186 187        Returns:188            List of Documents most similar to the query.189        """190        if self._by_text:191            return self.similarity_search_by_text(query, k, **kwargs)192        else:193            if self._embedding is None:194                raise ValueError(195                    "_embedding cannot be None for similarity_search when "196                    "_by_text=False"197                )198            embedding = self._embedding.embed_query(query)199            return self.similarity_search_by_vector(embedding, k, **kwargs)200 201    def similarity_search_by_text(202        self, query: str, k: int = 4, **kwargs: Any203    ) -> List[Document]:204        """Return docs most similar to query.205 206        Args:207            query: Text to look up documents similar to.208            k: Number of Documents to return. Defaults to 4.209 210        Returns:211            List of Documents most similar to the query.212        """213        content: Dict[str, Any] = {"concepts": [query]}214        if kwargs.get("search_distance"):215            content["certainty"] = kwargs.get("search_distance")216        query_obj = self._client.query.get(self._index_name, self._query_attrs)217        if kwargs.get("where_filter"):218            query_obj = query_obj.with_where(kwargs.get("where_filter"))219        if kwargs.get("tenant"):220            query_obj = query_obj.with_tenant(kwargs.get("tenant"))221        if kwargs.get("additional"):222            query_obj = query_obj.with_additional(kwargs.get("additional"))223        result = query_obj.with_near_text(content).with_limit(k).do()224        if "errors" in result:225            raise ValueError(f"Error during query: {result['errors']}")226        docs = []227        for res in result["data"]["Get"][self._index_name]:228            text = res.pop(self._text_key)229            docs.append(Document(page_content=text, metadata=res))230        return docs231 232    def similarity_search_by_vector(233        self, embedding: List[float], k: int = 4, **kwargs: Any234    ) -> List[Document]:235        """Look up similar documents by embedding vector in Weaviate."""236        vector = {"vector": embedding}237        query_obj = self._client.query.get(self._index_name, self._query_attrs)238        if kwargs.get("where_filter"):239            query_obj = query_obj.with_where(kwargs.get("where_filter"))240        if kwargs.get("tenant"):241            query_obj = query_obj.with_tenant(kwargs.get("tenant"))242        if kwargs.get("additional"):243            query_obj = query_obj.with_additional(kwargs.get("additional"))244        result = query_obj.with_near_vector(vector).with_limit(k).do()245        if "errors" in result:246            raise ValueError(f"Error during query: {result['errors']}")247        docs = []248        for res in result["data"]["Get"][self._index_name]:249            text = res.pop(self._text_key)250            docs.append(Document(page_content=text, metadata=res))251        return docs252 253    def max_marginal_relevance_search(254        self,255        query: str,256        k: int = 4,257        fetch_k: int = 20,258        lambda_mult: float = 0.5,259        **kwargs: Any,260    ) -> List[Document]:261        """Return docs selected using the maximal marginal relevance.262 263        Maximal marginal relevance optimizes for similarity to query AND diversity264        among selected documents.265 266        Args:267            query: Text to look up documents similar to.268            k: Number of Documents to return. Defaults to 4.269            fetch_k: Number of Documents to fetch to pass to MMR algorithm.270            lambda_mult: Number between 0 and 1 that determines the degree271                        of diversity among the results with 0 corresponding272                        to maximum diversity and 1 to minimum diversity.273                        Defaults to 0.5.274 275        Returns:276            List of Documents selected by maximal marginal relevance.277        """278        if self._embedding is not None:279            embedding = self._embedding.embed_query(query)280        else:281            raise ValueError(282                "max_marginal_relevance_search requires a suitable Embeddings object"283            )284 285        return self.max_marginal_relevance_search_by_vector(286            embedding, k=k, fetch_k=fetch_k, lambda_mult=lambda_mult, **kwargs287        )288 289    def max_marginal_relevance_search_by_vector(290        self,291        embedding: List[float],292        k: int = 4,293        fetch_k: int = 20,294        lambda_mult: float = 0.5,295        **kwargs: Any,296    ) -> List[Document]:297        """Return docs selected using the maximal marginal relevance.298 299        Maximal marginal relevance optimizes for similarity to query AND diversity300        among selected documents.301 302        Args:303            embedding: Embedding to look up documents similar to.304            k: Number of Documents to return. Defaults to 4.305            fetch_k: Number of Documents to fetch to pass to MMR algorithm.306            lambda_mult: Number between 0 and 1 that determines the degree307                        of diversity among the results with 0 corresponding308                        to maximum diversity and 1 to minimum diversity.309                        Defaults to 0.5.310 311        Returns:312            List of Documents selected by maximal marginal relevance.313        """314        vector = {"vector": embedding}315        query_obj = self._client.query.get(self._index_name, self._query_attrs)316        if kwargs.get("where_filter"):317            query_obj = query_obj.with_where(kwargs.get("where_filter"))318        if kwargs.get("tenant"):319            query_obj = query_obj.with_tenant(kwargs.get("tenant"))320        results = (321            query_obj.with_additional("vector")322            .with_near_vector(vector)323            .with_limit(fetch_k)324            .do()325        )326 327        payload = results["data"]["Get"][self._index_name]328        embeddings = [result["_additional"]["vector"] for result in payload]329        mmr_selected = maximal_marginal_relevance(330            np.array(embedding), embeddings, k=k, lambda_mult=lambda_mult331        )332 333        docs = []334        for idx in mmr_selected:335            text = payload[idx].pop(self._text_key)336            payload[idx].pop("_additional")337            meta = payload[idx]338            docs.append(Document(page_content=text, metadata=meta))339        return docs340 341    def similarity_search_with_score(342        self, query: str, k: int = 4, **kwargs: Any343    ) -> List[Tuple[Document, float]]:344        """345        Return list of documents most similar to the query346        text and cosine distance in float for each.347        Lower score represents more similarity.348        """349        if self._embedding is None:350            raise ValueError(351                "_embedding cannot be None for similarity_search_with_score"352            )353        content: Dict[str, Any] = {"concepts": [query]}354        if kwargs.get("search_distance"):355            content["certainty"] = kwargs.get("search_distance")356        query_obj = self._client.query.get(self._index_name, self._query_attrs)357        if kwargs.get("where_filter"):358            query_obj = query_obj.with_where(kwargs.get("where_filter"))359        if kwargs.get("tenant"):360            query_obj = query_obj.with_tenant(kwargs.get("tenant"))361 362        embedded_query = self._embedding.embed_query(query)363        if not self._by_text:364            vector = {"vector": embedded_query}365            result = (366                query_obj.with_near_vector(vector)367                .with_limit(k)368                .with_additional("vector")369                .do()370            )371        else:372            result = (373                query_obj.with_near_text(content)374                .with_limit(k)375                .with_additional("vector")376                .do()377            )378 379        if "errors" in result:380            raise ValueError(f"Error during query: {result['errors']}")381 382        docs_and_scores = []383        for res in result["data"]["Get"][self._index_name]:384            text = res.pop(self._text_key)385            score = np.dot(res["_additional"]["vector"], embedded_query)386            docs_and_scores.append((Document(page_content=text, metadata=res), score))387        return docs_and_scores388 389    @classmethod390    def from_texts(391        cls,392        texts: List[str],393        embedding: Embeddings,394        metadatas: Optional[List[dict]] = None,395        *,396        client: Optional[weaviate.Client] = None,397        weaviate_url: Optional[str] = None,398        weaviate_api_key: Optional[str] = None,399        batch_size: Optional[int] = None,400        index_name: Optional[str] = None,401        text_key: str = "text",402        by_text: bool = False,403        relevance_score_fn: Optional[404            Callable[[float], float]405        ] = _default_score_normalizer,406        **kwargs: Any,407    ) -> Weaviate:408        """Construct Weaviate wrapper from raw documents.409 410        This is a user-friendly interface that:411            1. Embeds documents.412            2. Creates a new index for the embeddings in the Weaviate instance.413            3. Adds the documents to the newly created Weaviate index.414 415        This is intended to be a quick way to get started.416 417        Args:418            texts: Texts to add to vector store.419            embedding: Text embedding model to use.420            metadatas: Metadata associated with each text.421            client: weaviate.Client to use.422            weaviate_url: The Weaviate URL. If using Weaviate Cloud Services get it423                from the ``Details`` tab. Can be passed in as a named param or by424                setting the environment variable ``WEAVIATE_URL``. Should not be425                specified if client is provided.426            weaviate_api_key: The Weaviate API key. If enabled and using Weaviate Cloud427                Services, get it from ``Details`` tab. Can be passed in as a named param428                or by setting the environment variable ``WEAVIATE_API_KEY``. Should429                not be specified if client is provided.430            batch_size: Size of batch operations.431            index_name: Index name.432            text_key: Key to use for uploading/retrieving text to/from vectorstore.433            by_text: Whether to search by text or by embedding.434            relevance_score_fn: Function for converting whatever distance function the435                vector store uses to a relevance score, which is a normalized similarity436                score (0 means dissimilar, 1 means similar).437            kwargs: Additional named parameters to pass to ``Weaviate.__init__()``.438 439        Example:440            .. code-block:: python441 442                from langchain_community.embeddings import OpenAIEmbeddings443                from langchain_community.vectorstores import Weaviate444 445                embeddings = OpenAIEmbeddings()446                weaviate = Weaviate.from_texts(447                    texts,448                    embeddings,449                    weaviate_url="http://localhost:8080"450                )451        """452 453        try:454            from weaviate.util import get_valid_uuid455        except ImportError as e:456            raise ImportError(457                "Could not import weaviate python  package. "458                "Please install it with `pip install weaviate-client`"459            ) from e460 461        client = client or _create_weaviate_client(462            url=weaviate_url,463            api_key=weaviate_api_key,464        )465        if batch_size:466            client.batch.configure(batch_size=batch_size)467 468        index_name = index_name or f"LangChain_{uuid4().hex}"469        schema = _default_schema(index_name, text_key)470        # check whether the index already exists471        if not client.schema.exists(index_name):472            client.schema.create_class(schema)473 474        embeddings = embedding.embed_documents(texts) if embedding else None475        attributes = list(metadatas[0].keys()) if metadatas else None476 477        # If the UUID of one of the objects already exists478        # then the existing object will be replaced by the new object.479        if "uuids" in kwargs:480            uuids = kwargs.pop("uuids")481        else:482            uuids = [get_valid_uuid(uuid4()) for _ in range(len(texts))]483 484        with client.batch as batch:485            for i, text in enumerate(texts):486                data_properties = {487                    text_key: text,488                }489                if metadatas is not None:490                    for key in metadatas[i].keys():491                        data_properties[key] = metadatas[i][key]492 493                _id = uuids[i]494 495                # if an embedding strategy is not provided, we let496                # weaviate create the embedding. Note that this will only497                # work if weaviate has been installed with a vectorizer module498                # like text2vec-contextionary for example499                params = {500                    "uuid": _id,501                    "data_object": data_properties,502                    "class_name": index_name,503                }504                if embeddings is not None:505                    params["vector"] = embeddings[i]506 507                batch.add_data_object(**params)508 509            batch.flush()510 511        return cls(512            client,513            index_name,514            text_key,515            embedding=embedding,516            attributes=attributes,517            relevance_score_fn=relevance_score_fn,518            by_text=by_text,519            **kwargs,520        )521 522    def delete(self, ids: Optional[List[str]] = None, **kwargs: Any) -> None:523        """Delete by vector IDs.524 525        Args:526            ids: List of ids to delete.527        """528 529        if ids is None:530            raise ValueError("No ids provided to delete.")531 532        # TODO: Check if this can be done in bulk533        for id in ids:534            self._client.data_object.delete(uuid=id)535