-
Notifications
You must be signed in to change notification settings - Fork 0
/
searcher.py
46 lines (36 loc) · 1.48 KB
/
searcher.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
from typing import List
from document import Document
from index_segment import IndexSegment
class Searcher:
"""
Executes search queries against the indexed documents contained in index segments.
Attributes:
index_segments (List[IndexSegment]): A list of IndexSegments that the searcher can query against.
"""
def __init__(self,index_name):
"""
Initializes a new instance of the Searcher class.
"""
self.mq = marqo.Client(url='http://localhost:8882')
self.index_name = index_name
def search(self, query: str, top_k: int = 10) -> List[dict]:
"""
Performs a vector search using Marqo based on the query.
Parameters:
query (str): The search query.
top_k (int): Number of top results to retrieve.
Returns:
List[dict]: A list of dictionaries representing the matching documents.
"""
results = self.mq.index(self.index_name).search(q=query, top_k=top_k)
return results["hits"]
def retrieve_document(self, document_id: str) -> Document:
# Logic to retrieve a Document object by its ID
pass
def __repr__(self):
"""
Provides a string representation of the Searcher object, primarily for debugging purposes.
Returns:
str: A string representation of the Searcher, including the number of loaded index segments.
"""
return f"Searcher(Loaded Index: {len(self.index_name)})"