Skip to content

teff.rag.base

teff.rag.base

Vector store abstract base and similarity utilities.

Classes:

Name Description
VectorStore

Abstract interface for vector storage and similarity search.

Functions:

Name Description
blend_scores

Blend a cosine score with a lexical overlap score.
cosine_similarity

Compute cosine similarity between two vectors.
finalize_results

Apply the filter, optional hybrid blending, rank, and cap at k.
match_filter

Return True if metadata satisfies the filter DSL.

VectorStore

Bases: ABC

Abstract interface for vector storage and similarity search.

Implementations must provide add, search, and delete. The extended operations (count, list, get, update_metadata, clear) default to :class:NotImplementedError and are implemented by the built-in stores.

Methods:

Name Description
add

Store vectors with IDs and metadata.
clear

Remove all stored vectors.
count

Return the number of stored vectors.
delete

Remove vectors by ID.
entries

Return (id, metadata) pairs with pagination.
get

Return (id, metadata) pairs for existing IDs.
search

Search for the k nearest neighbours.
update_metadata

Merge metadata into the metadata of an existing ID.
Source code in teff/rag/base.py 
 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
class VectorStore(ABC):
    """Abstract interface for vector storage and similarity search.

    Implementations must provide *add*, *search*, and *delete*.  The
    extended operations (*count*, *list*, *get*, *update_metadata*,
    *clear*) default to :class:`NotImplementedError` and are implemented
    by the built-in stores.
    """

    @abstractmethod
    async def add(self, vectors: list[tuple[str, list[float], dict]]) -> None:
        """Store vectors with IDs and metadata.

        Args:
            vectors: List of ``(id, embedding, metadata)`` tuples.
        """
        ...

    @abstractmethod
    async def search(
        self,
        query: list[float],
        k: int = 10,
        filter: dict | None = None,
        hybrid: bool = False,
        query_text: str | None = None,
    ) -> list[tuple[str, float, dict]]:
        """Search for the *k* nearest neighbours.

        Args:
            query: Query embedding.
            k: Maximum number of results.
            filter: Optional metadata filter DSL (see :func:`match_filter`).
            hybrid: When ``True``, blend a lexical keyword score with the
                cosine score (stores that support it; others ignore it).
            query_text: Original query text, required for ``hybrid``.

        Returns:
            List of ``(id, score, metadata)`` tuples sorted by score
            descending.  Scores are similarity-like (higher = more similar).
        """
        ...

    @abstractmethod
    async def delete(self, ids: list[str]) -> None:
        """Remove vectors by ID."""
        ...

    async def count(self) -> int:
        """Return the number of stored vectors."""
        raise NotImplementedError(f"{type(self).__name__} does not implement count()")

    async def entries(
        self, limit: int = 100, offset: int = 0
    ) -> list[tuple[str, dict]]:
        """Return ``(id, metadata)`` pairs with pagination."""
        raise NotImplementedError(f"{type(self).__name__} does not implement list()")

    async def get(self, ids: list[str]) -> list[tuple[str, dict]]:
        """Return ``(id, metadata)`` pairs for existing IDs."""
        raise NotImplementedError(f"{type(self).__name__} does not implement get()")

    async def update_metadata(self, id: str, metadata: dict) -> None:
        """Merge *metadata* into the metadata of an existing ID."""
        raise NotImplementedError(
            f"{type(self).__name__} does not implement update_metadata()"
        )

    async def clear(self) -> None:
        """Remove all stored vectors."""
        raise NotImplementedError(f"{type(self).__name__} does not implement clear()")

add abstractmethod async 

add(vectors)

Store vectors with IDs and metadata.

Parameters:

Name Type Description Default
vectors list[tuple[str, list[float], dict]]

List of (id, embedding, metadata) tuples.

 required
Source code in teff/rag/base.py 
92
93
94
95
96
97
98
99
@abstractmethod
async def add(self, vectors: list[tuple[str, list[float], dict]]) -> None:
    """Store vectors with IDs and metadata.

    Args:
        vectors: List of ``(id, embedding, metadata)`` tuples.
    """
    ...

clear async 

clear()

Remove all stored vectors.

Source code in teff/rag/base.py 
151
152
153
async def clear(self) -> None:
    """Remove all stored vectors."""
    raise NotImplementedError(f"{type(self).__name__} does not implement clear()")

count async 

count()

Return the number of stored vectors.

Source code in teff/rag/base.py 
131
132
133
async def count(self) -> int:
    """Return the number of stored vectors."""
    raise NotImplementedError(f"{type(self).__name__} does not implement count()")

delete abstractmethod async 

delete(ids)

Remove vectors by ID.

Source code in teff/rag/base.py 
126
127
128
129
@abstractmethod
async def delete(self, ids: list[str]) -> None:
    """Remove vectors by ID."""
    ...

entries async 

entries(limit=100, offset=0)

Return (id, metadata) pairs with pagination.

Source code in teff/rag/base.py 
135
136
137
138
139
async def entries(
    self, limit: int = 100, offset: int = 0
) -> list[tuple[str, dict]]:
    """Return ``(id, metadata)`` pairs with pagination."""
    raise NotImplementedError(f"{type(self).__name__} does not implement list()")

get async 

get(ids)

Return (id, metadata) pairs for existing IDs.

Source code in teff/rag/base.py 
141
142
143
async def get(self, ids: list[str]) -> list[tuple[str, dict]]:
    """Return ``(id, metadata)`` pairs for existing IDs."""
    raise NotImplementedError(f"{type(self).__name__} does not implement get()")

search abstractmethod async 

search(query, k=10, filter=None, hybrid=False, query_text=None)

Search for the k nearest neighbours.

Parameters:

Name Type Description Default
query list[float]

Query embedding.

 required
k int

Maximum number of results.

 10
filter dict | None

Optional metadata filter DSL (see :func:match_filter).

 None
hybrid bool

When True, blend a lexical keyword score with the cosine score (stores that support it; others ignore it).

 False
query_text str | None

Original query text, required for hybrid.

 None

Returns:

Type Description
list[tuple[str, float, dict]]

List of (id, score, metadata) tuples sorted by score
list[tuple[str, float, dict]]

descending. Scores are similarity-like (higher = more similar).
Source code in teff/rag/base.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
@abstractmethod
async def search(
    self,
    query: list[float],
    k: int = 10,
    filter: dict | None = None,
    hybrid: bool = False,
    query_text: str | None = None,
) -> list[tuple[str, float, dict]]:
    """Search for the *k* nearest neighbours.

    Args:
        query: Query embedding.
        k: Maximum number of results.
        filter: Optional metadata filter DSL (see :func:`match_filter`).
        hybrid: When ``True``, blend a lexical keyword score with the
            cosine score (stores that support it; others ignore it).
        query_text: Original query text, required for ``hybrid``.

    Returns:
        List of ``(id, score, metadata)`` tuples sorted by score
        descending.  Scores are similarity-like (higher = more similar).
    """
    ...

update_metadata async 

update_metadata(id, metadata)

Merge metadata into the metadata of an existing ID.

Source code in teff/rag/base.py 
145
146
147
148
149
async def update_metadata(self, id: str, metadata: dict) -> None:
    """Merge *metadata* into the metadata of an existing ID."""
    raise NotImplementedError(
        f"{type(self).__name__} does not implement update_metadata()"
    )

blend_scores 

blend_scores(cosine, text, query, alpha=0.4)

Blend a cosine score with a lexical overlap score.

Used for hybrid search: alpha weights the lexical (keyword) share, 1 - alpha the semantic (cosine) share. When the query has no alphabetic tokens the cosine score is returned unchanged.

Source code in teff/rag/base.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def blend_scores(cosine: float, text: str, query: str, alpha: float = 0.4) -> float:
    """Blend a cosine score with a lexical overlap score.

    Used for hybrid search: ``alpha`` weights the lexical (keyword) share,
    ``1 - alpha`` the semantic (cosine) share.  When the query has no
    alphabetic tokens the cosine score is returned unchanged.
    """
    tokens = {t for t in query.lower().split() if t}
    if not tokens:
        return cosine
    ltext = text.lower()
    hits = sum(1 for t in tokens if t in ltext)
    lexical = hits / len(tokens)
    return (1 - alpha) * cosine + alpha * lexical

cosine_similarity 

cosine_similarity(a, b)

Compute cosine similarity between two vectors.

Source code in teff/rag/base.py 
156
157
158
159
160
161
162
163
def cosine_similarity(a: list[float], b: list[float]) -> float:
    """Compute cosine similarity between two vectors."""
    dot = sum(x * y for x, y in zip(a, b))
    norm_a = math.sqrt(sum(x * x for x in a))
    norm_b = math.sqrt(sum(x * x for x in b))
    if norm_a == 0 or norm_b == 0:
        return 0.0
    return dot / (norm_a * norm_b)

finalize_results 

finalize_results(candidates, k, filter=None, hybrid=False, query_text=None)

Apply the filter, optional hybrid blending, rank, and cap at k.

Stores that retrieve candidates (e.g. brute-force scans) use this to post-process results consistently: drop non-matching metadata, blend a lexical score for hybrid search, sort descending, and trim.

Source code in teff/rag/base.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def finalize_results(
    candidates: list[tuple[str, float, dict]],
    k: int,
    filter: dict | None = None,
    hybrid: bool = False,
    query_text: str | None = None,
) -> list[tuple[str, float, dict]]:
    """Apply the filter, optional hybrid blending, rank, and cap at *k*.

    Stores that retrieve candidates (e.g. brute-force scans) use this to
    post-process results consistently: drop non-matching metadata, blend
    a lexical score for ``hybrid`` search, sort descending, and trim.
    """
    out: list[tuple[str, float, dict]] = []
    for vid, score, meta in candidates:
        if not match_filter(meta, filter):
            continue
        if hybrid and query_text:
            score = blend_scores(score, meta.get("text", ""), query_text)
        out.append((vid, score, meta))
    out.sort(key=lambda x: x[1], reverse=True)
    return out[:k]

match_filter 

match_filter(metadata, filter)

Return True if metadata satisfies the filter DSL.

A filter is a dict of field -> condition pairs:

  • scalar value — equality: {"category": "news"}
  • list value — membership: {"category": ["news", "tech"]}; when the stored field value is itself a list, any shared element matches
  • "$and" / "$or" keys combine sub-filters (lists of filters).

A missing field never matches a scalar or list condition.

Source code in teff/rag/base.py 
 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
def match_filter(metadata: dict, filter: dict | None) -> bool:
    """Return ``True`` if *metadata* satisfies the filter DSL.

    A filter is a dict of field -> condition pairs:

    - scalar value — equality: ``{"category": "news"}``
    - list value — membership: ``{"category": ["news", "tech"]}``; when the
      stored field value is itself a list, any shared element matches
    - ``"$and"`` / ``"$or"`` keys combine sub-filters (lists of filters).

    A missing field never matches a scalar or list condition.
    """
    if not filter:
        return True
    for key, cond in filter.items():
        if key == "$and":
            if not all(match_filter(metadata, sub) for sub in cond):
                return False
        elif key == "$or":
            if not any(match_filter(metadata, sub) for sub in cond):
                return False
        else:
            value = metadata.get(key)
            if isinstance(cond, list):
                if isinstance(value, list):
                    if not set(value).intersection(cond):
                        return False
                elif value not in cond:
                    return False
            elif value != cond:
                return False
    return True