Spaces:
Running
Running
File size: 3,513 Bytes
09361ac | 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 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 | 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 {}
|