Spaces:
Running
Running
| from __future__ import annotations | |
| from typing import Any, Optional | |
| import numpy as np | |
| def normalize(v: np.ndarray) -> np.ndarray: | |
| norm = np.linalg.norm(v, axis=-1, keepdims=True) | |
| norm = np.where(norm == 0, 1.0, norm) | |
| return v / norm | |
| def score_vectors(query: np.ndarray, matrix: np.ndarray, distance: str) -> np.ndarray: | |
| """Return similarity scores (higher is better) for ranking.""" | |
| if matrix.size == 0: | |
| return np.array([], dtype=np.float32) | |
| if distance == "Cosine": | |
| q = normalize(query.astype(np.float32)) | |
| m = normalize(matrix.astype(np.float32)) | |
| return m @ q | |
| if distance == "Dot": | |
| return matrix.astype(np.float32) @ query.astype(np.float32) | |
| # Euclid → convert distance to similarity: -distance | |
| diffs = matrix.astype(np.float32) - query.astype(np.float32) | |
| dists = np.linalg.norm(diffs, axis=1) | |
| return -dists | |
| def match_payload(payload: Optional[dict[str, Any]], flt: Optional[dict[str, Any]]) -> bool: | |
| """Minimal Qdrant-like filter: must / should / must_not with match & range.""" | |
| if not flt: | |
| return True | |
| payload = payload or {} | |
| must = flt.get("must") or [] | |
| should = flt.get("should") or [] | |
| must_not = flt.get("must_not") or [] | |
| for cond in must: | |
| if not _match_condition(payload, cond): | |
| return False | |
| for cond in must_not: | |
| if _match_condition(payload, cond): | |
| return False | |
| if should: | |
| return any(_match_condition(payload, cond) for cond in should) | |
| return True | |
| def _match_condition(payload: dict[str, Any], cond: dict[str, Any]) -> bool: | |
| if "must" in cond or "should" in cond or "must_not" in cond: | |
| return match_payload(payload, cond) | |
| key = cond.get("key") | |
| if key is None: | |
| return True | |
| value = _nested_get(payload, key) | |
| if "match" in cond: | |
| m = cond["match"] | |
| if "value" in m: | |
| return value == m["value"] | |
| if "any" in m: | |
| return value in m["any"] | |
| if "except" in m: | |
| return value not in m["except"] | |
| if "text" in m: | |
| return isinstance(value, str) and m["text"].lower() in value.lower() | |
| if "range" in cond: | |
| r = cond["range"] | |
| if value is None: | |
| return False | |
| try: | |
| num = float(value) | |
| except (TypeError, ValueError): | |
| return False | |
| if "gt" in r and not (num > r["gt"]): | |
| return False | |
| if "gte" in r and not (num >= r["gte"]): | |
| return False | |
| if "lt" in r and not (num < r["lt"]): | |
| return False | |
| if "lte" in r and not (num <= r["lte"]): | |
| return False | |
| return True | |
| if "is_empty" in cond: | |
| empty = value is None or value == "" or value == [] or value == {} | |
| return empty if cond["is_empty"] else not empty | |
| return True | |
| def _nested_get(payload: dict[str, Any], key: str) -> Any: | |
| cur: Any = payload | |
| for part in key.split("."): | |
| if not isinstance(cur, dict) or part not in cur: | |
| return None | |
| cur = cur[part] | |
| return cur | |
| def filter_payload_fields( | |
| payload: Optional[dict[str, Any]], with_payload: bool | list[str] | |
| ) -> Optional[dict[str, Any]]: | |
| if with_payload is False: | |
| return None | |
| if with_payload is True: | |
| return payload or {} | |
| if isinstance(with_payload, list): | |
| src = payload or {} | |
| return {k: src.get(k) for k in with_payload} | |
| return payload or {} | |