def rrf_search()

in src/lindorm_mcp_server/lindorm_vector_search.py [0:0]

• 36 lines of code
• 4 McCabe index (conditional complexity)

    def rrf_search(self, index_name: str, query_text: str, top_k: int, content_field: str, vector_field: str) -> list[str]:
        if not self._check_index_exist(index_name):
            return [f"{index_name} not exist"]
        vector = self._embedding_query(query_text)
        query = {
            "size": top_k,
            "_source": [content_field],
            "query": {
                "knn": {
                    vector_field: {
                        "vector": vector,
                        "filter": {
                            "match": {
                                content_field: query_text
                            }
                        },
                        "k": top_k
                    }
                }
            },
            "ext": {
                "lvector": {
                    "hybrid_search_type": "filter_rrf",
                    "rrf_rank_constant": "60",
                }
            }
        }
        try:
            response = self.client.search(
                body=query,
                index=index_name
            )
            return [hit["_source"][content_field] for hit in response['hits']['hits']]
        except Exception as e:
            logging.error(f"Error performing RRF search: {e}")
            return []