HNSW index configuration.
| 34 | |
| 35 | @dataclass |
| 36 | class HNSWConfig: |
| 37 | """HNSW index configuration.""" |
| 38 | metric: str = os.getenv("HNSW_METRIC", "euclidean") |
| 39 | max_levels: int = int(os.getenv("HNSW_MAX_LEVELS", "3")) |
| 40 | ef_search: int = int(os.getenv("HNSW_EF_SEARCH", "40")) |
| 41 | ef_construction: int = int(os.getenv("HNSW_EF_CONSTRUCTION", "100")) |
| 42 | # Index type allows switching between standard hnsw and experimental variants |
| 43 | # such as the new "partionedhnsw" (sic) index. |
| 44 | index_type: str = os.getenv("HNSW_INDEX_TYPE", "hnsw") |
| 45 | # Optional number of clusters for partitioned HNSW experimental variants. Interpreted as a |
| 46 | # string in the index definition to match existing HNSW parameters. Only set |
| 47 | # when HNSW_NUM_CLUSTERS is provided; otherwise default to 0 (disabled). |
| 48 | num_clusters: int = int(os.getenv("HNSW_NUM_CLUSTERS")) if os.getenv("HNSW_NUM_CLUSTERS") is not None else 0 |
| 49 | # Optional vector dimension; when using partionedhnsw (sic) this should match the |
| 50 | # length of the stored embeddings. Experimental, this is not yet released. |
| 51 | vector_dim: int = 0 |
| 52 | |
| 53 | def to_index_string(self) -> str: |
| 54 | """Generate index string for Dgraph schema.""" |
| 55 | # Base index name (e.g., "hnsw" or "partionedhnsw") |
| 56 | index_name = self.index_type |
| 57 | |
| 58 | # Common parameters shared by both standard and partitioned HNSW |
| 59 | params = [ |
| 60 | f'metric:"{self.metric}"', |
| 61 | f'maxLevels:"{self.max_levels}"', |
| 62 | f'efSearch:"{self.ef_search}"', |
| 63 | f'efConstruction:"{self.ef_construction}"', |
| 64 | ] |
| 65 | |
| 66 | # Partitioned HNSW adds numClusters; we include it whenever explicitly |
| 67 | # configured, regardless of the index name, to keep behavior simple. |
| 68 | if self.num_clusters > 0: |
| 69 | params.append(f'numClusters:"{self.num_clusters}"') |
| 70 | |
| 71 | # Some partitioned HNSW variants require the vector dimension to be |
| 72 | # specified explicitly in the index configuration. |
| 73 | if self.index_type == "partionedhnsw" and self.vector_dim > 0: |
| 74 | params.append(f'vectorDimension:"{self.vector_dim}"') |
| 75 | |
| 76 | joined = ",".join(params) |
| 77 | return f"{index_name}({joined})" |
| 78 | |
| 79 | |
| 80 | @dataclass |