Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:26
📄 __init__.py # Licensed to Elasticsearch B.V. under one or more contributor # license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright # ownership. Elasticsearch B.V. licenses this file to you under # the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License.	787 B	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 _utils.py # Licensed to Elasticsearch B.V. under one or more contributor # license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright # ownership. Elasticsearch B.V. licenses this file to you under # the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. from elasticsearch import BadRequestError, Elasticsearch, NotFoundError def model_must_be_deployed(client: Elasticsearch, model_id: str) -> None: """ :raises [NotFoundError]: if the model is neither downloaded nor deployed. :raises [ConflictError]: if the model is downloaded but not yet deployed. """ doc = {"text_field": f"test if the model '{model_id}' is deployed"} try: client.ml.infer_trained_model(model_id=model_id, docs=[doc]) except BadRequestError: # The model is deployed but expects a different input field name. pass def model_is_deployed(client: Elasticsearch, model_id: str) -> bool: try: model_must_be_deployed(client, model_id) return True except NotFoundError: return False	1.53 KB	2025-05-25 12:26
📄 embedding_service.py # Licensed to Elasticsearch B.V. under one or more contributor # license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright # ownership. Elasticsearch B.V. licenses this file to you under # the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. from abc import ABC, abstractmethod from typing import List from elasticsearch import Elasticsearch from elasticsearch._version import __versionstr__ as lib_version class EmbeddingService(ABC): @abstractmethod def embed_documents(self, texts: List[str]) -> List[List[float]]: """Generate embeddings for a list of documents. :param texts: A list of document strings to generate embeddings for. :return: A list of embeddings, one for each document in the input. """ @abstractmethod def embed_query(self, query: str) -> List[float]: """Generate an embedding for a single query text. :param text: The query text to generate an embedding for. :return: The embedding for the input query text. """ class ElasticsearchEmbeddings(EmbeddingService): """Elasticsearch as a service for embedding model inference. You need to have an embedding model downloaded and deployed in Elasticsearch: - https://www.elastic.co/guide/en/elasticsearch/reference/current/infer-trained-model.html - https://www.elastic.co/guide/en/machine-learning/current/ml-nlp-deploy-models.html """ # noqa: E501 def __init__( self, , client: Elasticsearch, model_id: str, input_field: str = "text_field", user_agent: str = f"elasticsearch-py-es/{lib_version}", ): """ :param agent_header: user agent header specific to the 3rd party integration. Used for usage tracking in Elastic Cloud. :param model_id: The model_id of the model deployed in the Elasticsearch cluster. :param input_field: The name of the key for the input text field in the document. Defaults to 'text_field'. :param client: Elasticsearch client connection. Alternatively specify the Elasticsearch connection with the other es_ parameters. """ # Add integration-specific usage header for tracking usage in Elastic Cloud. # client.options preserves existing (non-user-agent) headers. client = client.options(headers={"User-Agent": user_agent}) self.client = client self.model_id = model_id self.input_field = input_field def embed_documents(self, texts: List[str]) -> List[List[float]]: return self._embedding_func(texts) def embed_query(self, text: str) -> List[float]: result = self._embedding_func([text]) return result[0] def _embedding_func(self, texts: List[str]) -> List[List[float]]: response = self.client.ml.infer_trained_model( model_id=self.model_id, docs=[{self.input_field: text} for text in texts] ) return [doc["predicted_value"] for doc in response["inference_results"]]	3.5 KB	2025-05-25 12:26
📄 strategies.py # Licensed to Elasticsearch B.V. under one or more contributor # license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright # ownership. Elasticsearch B.V. licenses this file to you under # the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. from abc import ABC, abstractmethod from typing import Any, Dict, List, Optional, Tuple, Union, cast from elasticsearch import Elasticsearch from elasticsearch.helpers.vectorstore._sync._utils import model_must_be_deployed from elasticsearch.helpers.vectorstore._utils import DistanceMetric class RetrievalStrategy(ABC): @abstractmethod def es_query( self, , query: Optional[str], query_vector: Optional[List[float]], text_field: str, vector_field: str, k: int, num_candidates: int, filter: List[Dict[str, Any]] = [], ) -> Dict[str, Any]: """ Returns the Elasticsearch query body for the given parameters. The store will execute the query. :param query: The text query. Can be None if query_vector is given. :param k: The total number of results to retrieve. :param num_candidates: The number of results to fetch initially in knn search. :param filter: List of filter clauses to apply to the query. :param query_vector: The query vector. Can be None if a query string is given. :return: The Elasticsearch query body. """ @abstractmethod def es_mappings_settings( self, , text_field: str, vector_field: str, num_dimensions: Optional[int], ) -> Tuple[Dict[str, Any], Dict[str, Any]]: """ Create the required index and do necessary preliminary work, like creating inference pipelines or checking if a required model was deployed. :param client: Elasticsearch client connection. :param text_field: The field containing the text data in the index. :param vector_field: The field containing the vector representations in the index. :param num_dimensions: If vectors are indexed, how many dimensions do they have. :return: Dictionary with field and field type pairs that describe the schema. """ def before_index_creation( self, , client: Elasticsearch, text_field: str, vector_field: str ) -> None: """ Executes before the index is created. Used for setting up any required Elasticsearch resources like a pipeline. Defaults to a no-op. :param client: The Elasticsearch client. :param text_field: The field containing the text data in the index. :param vector_field: The field containing the vector representations in the index. """ pass def needs_inference(self) -> bool: """ Some retrieval strategies index embedding vectors and allow search by embedding vector, for example the `DenseVectorStrategy` strategy. Mapping a user input query string to an embedding vector is called inference. Inference can be applied in Elasticsearch (using a `model_id`) or outside of Elasticsearch (using an `EmbeddingService` defined on the `VectorStore`). In the latter case, this method has to return True. """ return False class SparseVectorStrategy(RetrievalStrategy): """Sparse retrieval strategy using the `sparse_vector` processor.""" def __init__(self, model_id: str = ".elser_model_2"): self.model_id = model_id self._tokens_field = "tokens" self._pipeline_name = f"{self.model_id}_sparse_embedding" def es_query( self, , query: Optional[str], query_vector: Optional[List[float]], text_field: str, vector_field: str, k: int, num_candidates: int, filter: List[Dict[str, Any]] = [], ) -> Dict[str, Any]: if query_vector: raise ValueError( "Cannot do sparse retrieval with a query_vector. " "Inference is currently always applied in Elasticsearch." ) if query is None: raise ValueError("please specify a query string") return { "query": { "bool": { "must": [ { "sparse_vector": { "field": f"{vector_field}.{self._tokens_field}", "inference_id": self.model_id, "query": query, } } ], "filter": filter, } } } def es_mappings_settings( self, , text_field: str, vector_field: str, num_dimensions: Optional[int], ) -> Tuple[Dict[str, Any], Dict[str, Any]]: mappings: Dict[str, Any] = { "properties": { vector_field: { "properties": {self._tokens_field: {"type": "sparse_vector"}} } } } settings = {"default_pipeline": self._pipeline_name} return mappings, settings def before_index_creation( self, , client: Elasticsearch, text_field: str, vector_field: str ) -> None: if self.model_id: model_must_be_deployed(client, self.model_id) # Create a pipeline for the model client.ingest.put_pipeline( id=self._pipeline_name, description="Embedding pipeline for Python VectorStore", processors=[ { "inference": { "model_id": self.model_id, "input_output": [ { "input_field": text_field, "output_field": f"{vector_field}.{self._tokens_field}", }, ], } } ], ) class DenseVectorStrategy(RetrievalStrategy): """K-nearest-neighbors retrieval.""" def __init__( self, , distance: DistanceMetric = DistanceMetric.COSINE, model_id: Optional[str] = None, hybrid: bool = False, rrf: Union[bool, Dict[str, Any]] = True, text_field: Optional[str] = "text_field", ): if hybrid and not text_field: raise ValueError( "to enable hybrid you have to specify a text_field (for BM25Strategy matching)" ) self.distance = distance self.model_id = model_id self.hybrid = hybrid self.rrf = rrf self.text_field = text_field def es_query( self, , query: Optional[str], query_vector: Optional[List[float]], text_field: str, vector_field: str, k: int, num_candidates: int, filter: List[Dict[str, Any]] = [], ) -> Dict[str, Any]: knn = { "filter": filter, "field": vector_field, "k": k, "num_candidates": num_candidates, } if query_vector is not None: knn["query_vector"] = query_vector else: # Inference in Elasticsearch. When initializing we make sure to always have # a model_id if don't have an embedding_service. knn["query_vector_builder"] = { "text_embedding": { "model_id": self.model_id, "model_text": query, } } if self.hybrid: return self._hybrid(query=cast(str, query), knn=knn, filter=filter) return {"knn": knn} def es_mappings_settings( self, , text_field: str, vector_field: str, num_dimensions: Optional[int], ) -> Tuple[Dict[str, Any], Dict[str, Any]]: if self.distance is DistanceMetric.COSINE: similarity = "cosine" elif self.distance is DistanceMetric.EUCLIDEAN_DISTANCE: similarity = "l2_norm" elif self.distance is DistanceMetric.DOT_PRODUCT: similarity = "dot_product" elif self.distance is DistanceMetric.MAX_INNER_PRODUCT: similarity = "max_inner_product" else: raise ValueError(f"Similarity {self.distance} not supported.") mappings: Dict[str, Any] = { "properties": { vector_field: { "type": "dense_vector", "dims": num_dimensions, "index": True, "similarity": similarity, }, } } return mappings, {} def before_index_creation( self, , client: Elasticsearch, text_field: str, vector_field: str ) -> None: if self.model_id: model_must_be_deployed(client, self.model_id) def _hybrid( self, query: str, knn: Dict[str, Any], filter: List[Dict[str, Any]] ) -> Dict[str, Any]: # Add a query to the knn query. # RRF is used to even the score from the knn query and text query # RRF has two optional parameters: {'rank_constant':int, 'rank_window_size':int} # https://www.elastic.co/guide/en/elasticsearch/reference/current/rrf.html standard_query = { "query": { "bool": { "must": [ { "match": { self.text_field: { "query": query, } } } ], "filter": filter, } } } if self.rrf is False: query_body = { "knn": knn, standard_query, } else: rrf_options = {} if isinstance(self.rrf, Dict): if "rank_constant" in self.rrf: rrf_options["rank_constant"] = self.rrf["rank_constant"] if "window_size" in self.rrf: # 'window_size' was renamed to 'rank_window_size', but we support # the older name for backwards compatibility rrf_options["rank_window_size"] = self.rrf["window_size"] if "rank_window_size" in self.rrf: rrf_options["rank_window_size"] = self.rrf["rank_window_size"] query_body = { "retriever": { "rrf": { "retrievers": [ {"standard": standard_query}, {"knn": knn}, ], rrf_options, }, }, } return query_body def needs_inference(self) -> bool: return not self.model_id class DenseVectorScriptScoreStrategy(RetrievalStrategy): """Exact nearest neighbors retrieval using the `script_score` query.""" def __init__(self, distance: DistanceMetric = DistanceMetric.COSINE) -> None: self.distance = distance def es_query( self, , query: Optional[str], query_vector: Optional[List[float]], text_field: str, vector_field: str, k: int, num_candidates: int, filter: List[Dict[str, Any]] = [], ) -> Dict[str, Any]: if not query_vector: raise ValueError("specify a query_vector") if self.distance is DistanceMetric.COSINE: similarity_algo = ( f"cosineSimilarity(params.query_vector, '{vector_field}') + 1.0" ) elif self.distance is DistanceMetric.EUCLIDEAN_DISTANCE: similarity_algo = f"1 / (1 + l2norm(params.query_vector, '{vector_field}'))" elif self.distance is DistanceMetric.DOT_PRODUCT: similarity_algo = f""" double value = dotProduct(params.query_vector, '{vector_field}'); return sigmoid(1, Math.E, -value); """ elif self.distance is DistanceMetric.MAX_INNER_PRODUCT: similarity_algo = f""" double value = dotProduct(params.query_vector, '{vector_field}'); if (dotProduct < 0) {{ return 1 / (1 + -1 dotProduct); }} return dotProduct + 1; """ else: raise ValueError(f"Similarity {self.distance} not supported.") query_bool: Dict[str, Any] = {"match_all": {}} if filter: query_bool = {"bool": {"filter": filter}} return { "query": { "script_score": { "query": query_bool, "script": { "source": similarity_algo, "params": {"query_vector": query_vector}, }, }, } } def es_mappings_settings( self, , text_field: str, vector_field: str, num_dimensions: Optional[int], ) -> Tuple[Dict[str, Any], Dict[str, Any]]: mappings = { "properties": { vector_field: { "type": "dense_vector", "dims": num_dimensions, "index": False, } } } return mappings, {} def needs_inference(self) -> bool: return True class BM25Strategy(RetrievalStrategy): def __init__( self, k1: Optional[float] = None, b: Optional[float] = None, ): self.k1 = k1 self.b = b def es_query( self, , query: Optional[str], query_vector: Optional[List[float]], text_field: str, vector_field: str, k: int, num_candidates: int, filter: List[Dict[str, Any]] = [], ) -> Dict[str, Any]: return { "query": { "bool": { "must": [ { "match": { text_field: { "query": query, } }, }, ], "filter": filter, }, }, } def es_mappings_settings( self, *, text_field: str, vector_field: str, num_dimensions: Optional[int], ) -> Tuple[Dict[str, Any], Dict[str, Any]]: similarity_name = "custom_bm25" mappings: Dict[str, Any] = { "properties": { text_field: { "type": "text", "similarity": similarity_name, }, }, } bm25: Dict[str, Any] = { "type": "BM25", } if self.k1 is not None: bm25["k1"] = self.k1 if self.b is not None: bm25["b"] = self.b settings = { "similarity": { similarity_name: bm25, } } return mappings, settings	15.7 KB	2025-05-25 12:26
📄 vectorstore.py # Licensed to Elasticsearch B.V. under one or more contributor # license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright # ownership. Elasticsearch B.V. licenses this file to you under # the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. import logging import uuid from typing import Any, Callable, Dict, List, Optional from elasticsearch import Elasticsearch from elasticsearch._version import __versionstr__ as lib_version from elasticsearch.helpers import BulkIndexError, bulk from elasticsearch.helpers.vectorstore import ( EmbeddingService, RetrievalStrategy, ) from elasticsearch.helpers.vectorstore._utils import maximal_marginal_relevance logger = logging.getLogger(__name__) class VectorStore: """ VectorStore is a higher-level abstraction of indexing and search. Users can pick from available retrieval strategies. Documents have up to 3 fields: - text_field: the text to be indexed and searched. - metadata: additional information about the document, either schema-free or defined by the supplied metadata_mappings. - vector_field (usually not filled by the user): the embedding vector of the text. Depending on the strategy, vector embeddings are - created by the user beforehand - created by this AsyncVectorStore class in Python - created in-stack by inference pipelines. """ def __init__( self, client: Elasticsearch, , index: str, retrieval_strategy: RetrievalStrategy, embedding_service: Optional[EmbeddingService] = None, num_dimensions: Optional[int] = None, text_field: str = "text_field", vector_field: str = "vector_field", metadata_mappings: Optional[Dict[str, Any]] = None, user_agent: str = f"elasticsearch-py-vs/{lib_version}", custom_index_settings: Optional[Dict[str, Any]] = None, ) -> None: """ :param user_header: user agent header specific to the 3rd party integration. Used for usage tracking in Elastic Cloud. :param index: The name of the index to query. :param retrieval_strategy: how to index and search the data. See the strategies module for availble strategies. :param text_field: Name of the field with the textual data. :param vector_field: For strategies that perform embedding inference in Python, the embedding vector goes in this field. :param client: Elasticsearch client connection. Alternatively specify the Elasticsearch connection with the other es_ parameters. :param custom_index_settings: A dictionary of custom settings for the index. This can include configurations like the number of shards, number of replicas, analysis settings, and other index-specific settings. If not provided, default settings will be used. Note that if the same setting is provided by both the user and the strategy, will raise an error. """ # Add integration-specific usage header for tracking usage in Elastic Cloud. # client.options preserves existing (non-user-agent) headers. client = client.options(headers={"User-Agent": user_agent}) if hasattr(retrieval_strategy, "text_field"): retrieval_strategy.text_field = text_field if hasattr(retrieval_strategy, "vector_field"): retrieval_strategy.vector_field = vector_field self.client = client self.index = index self.retrieval_strategy = retrieval_strategy self.embedding_service = embedding_service self.num_dimensions = num_dimensions self.text_field = text_field self.vector_field = vector_field self.metadata_mappings = metadata_mappings self.custom_index_settings = custom_index_settings def close(self) -> None: return self.client.close() def add_texts( self, texts: List[str], , metadatas: Optional[List[Dict[str, Any]]] = None, vectors: Optional[List[List[float]]] = None, ids: Optional[List[str]] = None, refresh_indices: bool = True, create_index_if_not_exists: bool = True, bulk_kwargs: Optional[Dict[str, Any]] = None, ) -> List[str]: """Add documents to the Elasticsearch index. :param texts: List of text documents. :param metadata: Optional list of document metadata. Must be of same length as texts. :param vectors: Optional list of embedding vectors. Must be of same length as texts. :param ids: Optional list of ID strings. Must be of same length as texts. :param refresh_indices: Whether to refresh the index after deleting documents. Defaults to True. :param create_index_if_not_exists: Whether to create the index if it does not exist. Defaults to True. :param bulk_kwargs: Arguments to pass to the bulk function when indexing (for example chunk_size). :return: List of IDs of the created documents, either echoing the provided one or returning newly created ones. """ bulk_kwargs = bulk_kwargs or {} ids = ids or [str(uuid.uuid4()) for _ in texts] requests = [] if create_index_if_not_exists: self._create_index_if_not_exists() if self.embedding_service and not vectors: vectors = self.embedding_service.embed_documents(texts) for i, text in enumerate(texts): metadata = metadatas[i] if metadatas else {} request: Dict[str, Any] = { "_op_type": "index", "_index": self.index, self.text_field: text, "metadata": metadata, "_id": ids[i], } if vectors: request[self.vector_field] = vectors[i] requests.append(request) if len(requests) > 0: try: success, failed = bulk( self.client, requests, stats_only=True, refresh=refresh_indices, bulk_kwargs, ) logger.debug(f"added texts {ids} to index") return ids except BulkIndexError as e: logger.error(f"Error adding texts: {e}") firstError = e.errors[0].get("index", {}).get("error", {}) logger.error(f"First error reason: {firstError.get('reason')}") raise e else: logger.debug("No texts to add to index") return [] def delete( # type: ignore[no-untyped-def] self, , ids: Optional[List[str]] = None, query: Optional[Dict[str, Any]] = None, refresh_indices: bool = True, delete_kwargs, ) -> bool: """Delete documents from the Elasticsearch index. :param ids: List of IDs of documents to delete. :param refresh_indices: Whether to refresh the index after deleting documents. Defaults to True. :return: True if deletion was successful. """ if ids is not None and query is not None: raise ValueError("one of ids or query must be specified") elif ids is None and query is None: raise ValueError("either specify ids or query") try: if ids: body = [ {"_op_type": "delete", "_index": self.index, "_id": _id} for _id in ids ] bulk( self.client, body, refresh=refresh_indices, ignore_status=404, delete_kwargs, ) logger.debug(f"Deleted {len(body)} texts from index") else: self.client.delete_by_query( index=self.index, query=query, refresh=refresh_indices, *delete_kwargs, ) except BulkIndexError as e: logger.error(f"Error deleting texts: {e}") firstError = e.errors[0].get("index", {}).get("error", {}) logger.error(f"First error reason: {firstError.get('reason')}") raise e return True def search( self, , query: Optional[str] = None, query_vector: Optional[List[float]] = None, k: int = 4, num_candidates: int = 50, fields: Optional[List[str]] = None, filter: Optional[List[Dict[str, Any]]] = None, custom_query: Optional[ Callable[[Dict[str, Any], Optional[str]], Dict[str, Any]] ] = None, ) -> List[Dict[str, Any]]: """ :param query: Input query string. :param query_vector: Input embedding vector. If given, input query string is ignored. :param k: Number of returned results. :param num_candidates: Number of candidates to fetch from data nodes in knn. :param fields: List of field names to return. :param filter: Elasticsearch filters to apply. :param custom_query: Function to modify the Elasticsearch query body before it is sent to Elasticsearch. :return: List of document hits. Includes _index, _id, _score and _source. """ if fields is None: fields = [] if "metadata" not in fields: fields.append("metadata") if self.text_field not in fields: fields.append(self.text_field) if self.embedding_service and not query_vector: if not query: raise ValueError("specify a query or a query_vector to search") query_vector = self.embedding_service.embed_query(query) query_body = self.retrieval_strategy.es_query( query=query, query_vector=query_vector, text_field=self.text_field, vector_field=self.vector_field, k=k, num_candidates=num_candidates, filter=filter or [], ) if custom_query is not None: query_body = custom_query(query_body, query) logger.debug(f"Calling custom_query, Query body now: {query_body}") response = self.client.search( index=self.index, query_body, size=k, source=True, source_includes=fields, ) hits: List[Dict[str, Any]] = response["hits"]["hits"] return hits def _create_index_if_not_exists(self) -> None: exists = self.client.indices.exists(index=self.index) if exists.meta.status == 200: logger.debug(f"Index {self.index} already exists. Skipping creation.") return if self.retrieval_strategy.needs_inference(): if not self.num_dimensions and not self.embedding_service: raise ValueError( "retrieval strategy requires embeddings; either embedding_service " "or num_dimensions need to be specified" ) if not self.num_dimensions and self.embedding_service: vector = self.embedding_service.embed_query("get num dimensions") self.num_dimensions = len(vector) mappings, settings = self.retrieval_strategy.es_mappings_settings( text_field=self.text_field, vector_field=self.vector_field, num_dimensions=self.num_dimensions, ) if self.custom_index_settings: conflicting_keys = set(self.custom_index_settings.keys()) & set( settings.keys() ) if conflicting_keys: raise ValueError(f"Conflicting settings: {conflicting_keys}") else: settings.update(self.custom_index_settings) if self.metadata_mappings: metadata = mappings["properties"].get("metadata", {"properties": {}}) for key in self.metadata_mappings.keys(): if key in metadata: raise ValueError(f"metadata key {key} already exists in mappings") metadata = dict(metadata["properties"], *self.metadata_mappings) mappings["properties"]["metadata"] = {"properties": metadata} self.retrieval_strategy.before_index_creation( client=self.client, text_field=self.text_field, vector_field=self.vector_field, ) self.client.indices.create( index=self.index, mappings=mappings, settings=settings ) def max_marginal_relevance_search( self, , query: Optional[str] = None, query_embedding: Optional[List[float]] = None, embedding_service: Optional[EmbeddingService] = None, vector_field: str, k: int = 4, num_candidates: int = 20, lambda_mult: float = 0.5, fields: Optional[List[str]] = None, custom_query: Optional[ Callable[[Dict[str, Any], Optional[str]], Dict[str, Any]] ] = None, ) -> List[Dict[str, Any]]: """Return docs selected using the maximal marginal relevance. Maximal marginal relevance optimizes for similarity to query AND diversity among selected documents. :param query (str): Text to look up documents similar to. :param query_embedding: Input embedding vector. If given, input query string is ignored. :param k (int): Number of Documents to return. Defaults to 4. :param fetch_k (int): Number of Documents to fetch to pass to MMR algorithm. :param lambda_mult (float): Number between 0 and 1 that determines the degree of diversity among the results with 0 corresponding to maximum diversity and 1 to minimum diversity. Defaults to 0.5. :param fields: Other fields to get from elasticsearch source. These fields will be added to the document metadata. :return: A list of Documents selected by maximal marginal relevance. """ remove_vector_query_field_from_metadata = True if fields is None: fields = [vector_field] elif vector_field not in fields: fields.append(vector_field) else: remove_vector_query_field_from_metadata = False # Embed the query if query_embedding: query_vector = query_embedding else: if not query: raise ValueError("specify either query or query_embedding to search") elif embedding_service: query_vector = embedding_service.embed_query(query) elif self.embedding_service: query_vector = self.embedding_service.embed_query(query) else: raise ValueError("specify embedding_service to search with query") # Fetch the initial documents got_hits = self.search( query=None, query_vector=query_vector, k=num_candidates, fields=fields, custom_query=custom_query, ) # Get the embeddings for the fetched documents got_embeddings = [hit["_source"][vector_field] for hit in got_hits] # Select documents using maximal marginal relevance selected_indices = maximal_marginal_relevance( query_vector, got_embeddings, lambda_mult=lambda_mult, k=k ) selected_hits = [got_hits[i] for i in selected_indices] if remove_vector_query_field_from_metadata: for hit in selected_hits: del hit["_source"][vector_field] return selected_hits	16.14 KB	2025-05-25 12:26

Terminal