Spaces:
Running
Running
File size: 5,355 Bytes
a55ea57 | 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 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 | """
File: /list_reduction.py
Created Date: Wednesday December 31st 2025
Author: Christian Nonis <alch.infoemail@gmail.com>
-----
Last Modified: Wednesday December 31st 2025 9:44:49 am
Modified By: Christian Nonis <alch.infoemail@gmail.com>
-----
"""
from typing import List, Optional, TypeVar
from src.utils.similarity.vectors import cosine_similarity
from pydantic import BaseModel
T = TypeVar("T")
def _get_nested_value(obj, key_path: str):
"""
Get a value from an object using a dot-separated key path.
Supports both object attributes and dictionary keys.
Example: _get_nested_value(obj, "a.b.c") returns obj.a.b.c or obj["a"]["b"]["c"]
"""
keys = key_path.split(".")
current = obj
for key in keys:
if isinstance(current, dict):
if key not in current:
return None
current = current[key]
else:
if not hasattr(current, key):
return None
current = getattr(current, key)
return current
def _is_vector(value) -> bool:
"""
Check if a value is a vector (list of numbers).
"""
if not isinstance(value, (list, tuple)):
return False
if len(value) == 0:
return False
return all(isinstance(x, (int, float)) for x in value)
class Rerank(BaseModel):
local: str
with_: str
def _fallback_embed_text(text: str) -> list[float]:
buckets = [0.0] * 8
normalized = str(text).strip().lower()
if not normalized:
return buckets
for idx, char in enumerate(normalized):
buckets[idx % 8] += float(ord(char))
divisor = float(len(normalized))
return [value / divisor for value in buckets]
def _resolve_embeddings_client():
try:
from src.core.instances import embeddings_small_adapter
return embeddings_small_adapter
except Exception:
return None
def _embed_text(embeddings_client, text: str) -> list[float]:
if embeddings_client is None:
return _fallback_embed_text(text)
result = embeddings_client.embed_text(text)
if hasattr(result, "embeddings"):
return list(result.embeddings)
return list(result)
def reduce_list(
list_: List[T],
access_key: Optional[str] = None,
similarity_threshold: float = 0.92,
by_vector: Optional[list[float]] = None,
rerank: Optional[Rerank] = None,
) -> List[T]:
"""
Reduce a list of items by grouping similar items together.
Returns a list with one representative item from each similarity group.
If access_key points to a vector (list of floats), it will be used directly.
If access_key points to a string, embeddings will be computed.
If by_vector is provided, items are compared to by_vector instead of each other.
"""
if not list_:
return []
embeddings_client = _resolve_embeddings_client()
reduced_list = []
item_embeddings = []
if rerank and isinstance(rerank, dict):
rerank = Rerank(**rerank)
for item in list_:
if access_key:
accessed_value = _get_nested_value(item, access_key)
if accessed_value is None:
continue
if _is_vector(accessed_value):
vector_item = list(accessed_value)
else:
text = str(accessed_value)
vector_item = _embed_text(embeddings_client, text)
else:
text = str(item)
vector_item = _embed_text(embeddings_client, text)
if by_vector is not None:
similarity = cosine_similarity(vector_item, by_vector)
rerank_similarity = None
if rerank and rerank.local:
txt_value = _get_nested_value(item, rerank.local)
if txt_value:
vector_txt_value = _embed_text(embeddings_client, txt_value)
v_comp_txt_value = _embed_text(embeddings_client, rerank.with_)
rerank_similarity = cosine_similarity(
vector_txt_value, v_comp_txt_value
)
if similarity >= similarity_threshold or (
rerank_similarity and rerank_similarity >= similarity_threshold
):
reduced_list.append(item)
else:
is_similar = False
for i, existing_vector in enumerate(item_embeddings):
similarity = cosine_similarity(vector_item, existing_vector)
rerank_similarity = None
if rerank and rerank.local:
txt_value = _get_nested_value(item, rerank.local)
if txt_value:
vector_txt_value = _embed_text(embeddings_client, txt_value)
v_comp_txt_value = _embed_text(embeddings_client, rerank.with_)
rerank_similarity = cosine_similarity(
vector_txt_value, v_comp_txt_value
)
if similarity >= similarity_threshold or (
rerank_similarity and rerank_similarity >= similarity_threshold
):
is_similar = True
break
if not is_similar:
reduced_list.append(item)
item_embeddings.append(vector_item)
return reduced_list
|