def _convert_queries_to_msearch_query()

in datastore/providers/elasticsearch_datastore.py [0:0]

• 17 lines of code
• 2 McCabe index (conditional complexity)

    def _convert_queries_to_msearch_query(self, queries: List[QueryWithEmbedding]):
        searches = []

        for query in queries:
            searches.append({"index": self.index_name})
            searches.append(
                {
                    "_source": True,
                    "knn": {
                        "field": "embedding",
                        "query_vector": query.embedding,
                        "k": query.top_k,
                        "num_candidates": query.top_k,
                    },
                    "size": query.top_k,
                }
            )

        return searches