@@ -43,13 +43,11 @@ Flow: Search + Concurrent Inserts (reuses existing data)
43
43
44
44
** File 1: ` engine/base_client/search.py ` **
45
45
``` python
46
- def process_chunk (chunk , search_one , insert_one , insert_fraction = 0.1 , test_set = None ):
46
+ def process_chunk (chunk , search_one , insert_one , insert_fraction = 0.1 ):
47
47
results = []
48
48
for i, query in enumerate (chunk):
49
49
if random.random() < insert_fraction:
50
- # Insert: use a vector from test_set
51
- vector_id, vector, metadata = test_set[i % len (test_set)]
52
- result = insert_one(vector_id, vector, metadata)
50
+ result = insert_one(query)
53
51
else :
54
52
# Search
55
53
result = search_one(query)
@@ -59,35 +57,36 @@ def process_chunk(chunk, search_one, insert_one, insert_fraction=0.1, test_set=N
59
57
60
58
** File 2: ` worker_function ` **
61
59
``` python
62
- def worker_function (self , distance , search_one , insert_one , chunk , result_queue , insert_fraction = 0.1 , test_set = None ):
60
+ def worker_function (self , distance , search_one , insert_one , chunk , result_queue , insert_fraction = 0.1 ):
63
61
self .init_client(self .host, distance, self .connection_params, self .search_params)
64
62
self .setup_search()
65
63
start_time = time.perf_counter()
66
- results = process_chunk(chunk, search_one, insert_one, insert_fraction, test_set )
64
+ results = process_chunk(chunk, search_one, insert_one, insert_fraction)
67
65
result_queue.put((start_time, results))
68
66
```
69
67
70
68
** File 3: ` BaseSearcher.search_all() ` **
71
- - When creating worker processes, pass ` search_one ` , ` insert_one ` , ` insert_fraction ` , and ` test_set ` as arguments to each worker.
69
+ - When creating worker processes, pass ` search_one ` , ` insert_one ` , and ` insert_fraction ` as arguments to each worker.
72
70
73
71
** File 4: Engine-specific ` insert_one ` implementations** (~ 5 lines each)
74
72
``` python
75
73
# Example: engine/clients/redis/search.py
76
74
@ classmethod
77
- def insert_one (cls , vector_id : int , vector : List[float ], metadata : Optional[dict ] = None ):
78
- """ Redis-specific single vector insert"""
75
+ def insert_one (cls , query ):
76
+ """ Redis-specific single vector insert from a Query object"""
77
+ # Extract vector_id, vector, metadata from query as needed
79
78
cls .client.hset(
80
- str (vector_id ),
81
- mapping = {" vector" : np.array( vector).astype(np.float32).tobytes() , ** (metadata or {})}
79
+ str (query.id ),
80
+ mapping = {" vector" : query. vector, ** (getattr (query, ' metadata' , {}) or {})}
82
81
)
83
82
84
83
# Example: engine/clients/qdrant/search.py
85
84
@ classmethod
86
- def insert_one (cls , vector_id : int , vector : List[ float ], metadata : Optional[ dict ] = None ):
87
- """ Qdrant-specific single vector insert"""
85
+ def insert_one (cls , query ):
86
+ """ Qdrant-specific single vector insert from a Query object """
88
87
cls .client.upsert(
89
88
collection_name = QDRANT_COLLECTION_NAME ,
90
- points = [{" id" : vector_id , " vector" : vector, " payload" : metadata or {}}],
89
+ points = [{" id" : query.id , " vector" : query. vector, " payload" : getattr (query, ' metadata' , {}) or {}}],
91
90
wait = False ,
92
91
)
93
92
```
0 commit comments