-
Notifications
You must be signed in to change notification settings - Fork 9
/
Copy pathdb-explorer.py
68 lines (54 loc) · 1.9 KB
/
db-explorer.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
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
from pymilvus import MilvusClient
from pymilvus import model
from rich import print
from rich.panel import Panel
# Initialize the Milvus client
client = MilvusClient("milvusdemo.db")
# Initialize the embedding function
embedding_fn = model.DefaultEmbeddingFunction()
def semantic_search(query: str, limit: int = 5):
"""
Perform a semantic search on the 'lore' collection.
Args:
query (str): The search query.
limit (int): The maximum number of results to return.
Returns:
list: A list of search results.
"""
# Embed the query
query_vector = embedding_fn.encode_documents([query])[0]
# Perform the search
results = client.search(
collection_name="lore",
data=[query_vector],
output_fields=["name", "content", "keywords"],
limit=limit,
)
return results[0] # Return the first (and only) query result
def main():
# Get collection statistics
stats = client.get_collection_stats("lore")
total_records = stats["row_count"]
print(f"Welcome to the Lore Explorer!")
print(f"Total records in the database: {total_records}\n")
while True:
# Get user input
query = input("Enter your search query (or 'quit' to exit): ")
if query.lower() == 'quit':
break
# Perform the search
results = semantic_search(query)
# Display results
print(f"\nSearch results for: '{query}'\n")
for i, result in enumerate(results, 1):
entity = result['entity']
print(Panel.fit(
f"[bold]Name:[/bold] {entity['name']}\n\n"
f"[bold]Content:[/bold] {entity['content']}\n\n"
f"[bold]Keywords:[/bold] {', '.join(entity['keywords'])}\n\n"
f"[bold]Distance:[/bold] {result['distance']}",
title=f"Result {i}"
))
print("\n")
if __name__ == "__main__":
main()