Code: Select all
def _get_relevant_documents(self, query: str) -> List[Document]:
query_vector = self.embedding_model.embed_query(query)
print("query_vector",type(query_vector))
if hasattr(query_vector, "tolist"):
query_vector = query_vector.tolist()
print("query_vector2",type(query_vector))
query_vector1 = [float(x) for x in query_vector]
query_vector = query_vector1[:100]
cosmos_search_stage = {
"$search": {
"cosmosSearch": {
"vector": {
"path": "embedding",
"queryVector": query_vector,
"k": self.top_k,
"similarity": "COS"
}
}
}
}
if self.keyword:
cosmos_search_stage["$search"]["cosmosSearch"]["text"] = {
"path": "description",
"query": self.keyword
}
pipeline = [cosmos_search_stage]
results = list(self.collection.aggregate(pipeline))