| 982 | |
| 983 | |
| 984 | def run_qps_latency( |
| 985 | dense_query_vectors: np.ndarray, |
| 986 | text_queries: List[str], |
| 987 | collection, |
| 988 | top_k: int, |
| 989 | ): |
| 990 | subset_size = min(1000, len(dense_query_vectors)) |
| 991 | dense_subset = dense_query_vectors[:subset_size] |
| 992 | text_subset = text_queries[:subset_size] |
| 993 | |
| 994 | with ThreadPoolExecutor(max_workers=MAX_WORKERS) as ex: |
| 995 | start = time.time() |
| 996 | futures = [ |
| 997 | ex.submit( |
| 998 | collection.search.hybrid, |
| 999 | dense_query=dense_subset[i].tolist(), |
| 1000 | sparse_query=text_subset[i], |
| 1001 | top_k=top_k, |
| 1002 | dense_weight=0.5, |
| 1003 | sparse_weight=0.5, |
| 1004 | ) |
| 1005 | for i in range(subset_size) |
| 1006 | ] |
| 1007 | for f in as_completed(futures): |
| 1008 | try: |
| 1009 | f.result() |
| 1010 | except Exception as e: |
| 1011 | print(f"Sequential QPS query failed: {e}") |
| 1012 | elapsed = time.time() - start |
| 1013 | qps = subset_size / elapsed |
| 1014 | |
| 1015 | # P50 / P95 latency |
| 1016 | times = [] |
| 1017 | latency_subset_size = min(100, len(dense_query_vectors)) |
| 1018 | for i in tqdm(range(latency_subset_size), desc="Latency"): |
| 1019 | start = time.time() |
| 1020 | try: |
| 1021 | collection.search.hybrid( |
| 1022 | dense_query=dense_query_vectors[i].tolist(), |
| 1023 | sparse_query=text_queries[i], |
| 1024 | top_k=top_k, |
| 1025 | dense_weight=0.5, |
| 1026 | sparse_weight=0.5, |
| 1027 | ) |
| 1028 | times.append((time.time() - start) * 1000) |
| 1029 | except Exception as e: |
| 1030 | print(f"Latency query failed: {e}") |
| 1031 | times.sort() |
| 1032 | if not times: |
| 1033 | return qps, 0.0, 0.0 |
| 1034 | return qps, times[len(times) // 2], times[int(len(times) * 0.95)] |
| 1035 | |
| 1036 | |
| 1037 | def save_comprehensive_results( |