| from typing import Any, Dict | |
| from haystack import Document | |
| from fastrag.rankers import IPEXBiEncoderSimilarityRanker | |
| class EndpointHandler: | |
| def __init__(self, path=""): | |
| model_id = "Intel/bge-large-en-v1.5-rag-int8-static" | |
| self.ranker = IPEXBiEncoderSimilarityRanker(model=model_id) | |
| self.ranker.warm_up() | |
| def __call__(self, data: Dict[str, Any]): | |
| query = data.get("query") | |
| documents = data.get("documents") | |
| top_k = data.get("top_k", None) | |
| documents = [Document.from_dict(doc) for doc in documents] | |
| return self.ranker.run(query, documents, top_k) | |