Skip to content

symfonic.memory.backends.in_memory_vector

in_memory_vector

In-memory VectorBackend and the brute-force similarity it searches with.

Split out of :mod:symfonic.memory.backends.in_memory (378 lines against the 300-line budget). That module implemented two unrelated protocols in one file: GraphBackend, which stores nodes and edges, and VectorBackend, which stores embeddings. Only the vector half ever used _cosine_similarity, so the two leave with nothing shared between them.

in_memory re-exports both names, so existing imports are unchanged.

InMemoryVectorBackend 

InMemoryVectorBackend()

In-memory implementation of the VectorBackend protocol.

Stores embeddings in a list and uses brute-force cosine similarity for search. Suitable for testing only.

Source code in src/symfonic/memory/backends/in_memory_vector.py 
55
56
def __init__(self) -> None:
    self._store: dict[str, list[dict[str, Any]]] = defaultdict(list)

add async 

add(scope: StorageScope, ids: list[str], embeddings: list[list[float]], metadatas: list[dict[str, Any]], documents: list[str]) -> None

Add vectors with metadata to the store.

v8.0: stamps the materialised scope_path into the metadata bag so the prefix-isolation filter (§5.d) can run on every search.

Source code in src/symfonic/memory/backends/in_memory_vector.py 
 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
async def add(
    self,
    scope: StorageScope,
    ids: list[str],
    embeddings: list[list[float]],
    metadatas: list[dict[str, Any]],
    documents: list[str],
) -> None:
    """Add vectors with metadata to the store.

    v8.0: stamps the materialised ``scope_path`` into the metadata bag so
    the prefix-isolation filter (§5.d) can run on every search.
    """
    scope_path = materialise_scope_path(scope)
    from symfonic.memory.backends.in_memory_ownership import native_owner
    native = _is_native_scope(scope)
    wanted = set(ids)
    for entry in self._store.get(scope.tenant_id, []):
        owner = metadata_scope_path(entry["metadata"], scope.tenant_id, scope=scope)
        if entry["id"] in wanted and (
            owner != scope_path if native else native_owner(owner)
        ):
            from symfonic.memory.errors import VectorOwnershipConflict
            raise VectorOwnershipConflict("Vector ID already belongs to a different scope")
    if not native and any(native_owner(meta.get(SCOPE_PATH_KEY)) for meta in metadatas):
        raise ValueError("Legacy metadata cannot declare a native scope owner")
    added = []
    for i, doc_id in enumerate(ids):
        meta = dict(metadatas[i]) if i < len(metadatas) else {}
        if native:
            meta[SCOPE_PATH_KEY] = scope_path
        else:
            meta.setdefault(SCOPE_PATH_KEY, scope_path)
        entries = added if native else self._store[scope.tenant_id]
        entries.append({
            "id": doc_id,
            "embedding": embeddings[i],
            "metadata": meta,
            "document": documents[i] if i < len(documents) else "",
        })
    if native:
        replacements = {row["id"]: row for row in added}
        self._store[scope.tenant_id] = [
            row for row in self._store.get(scope.tenant_id, ())
            if row["id"] not in replacements
        ] + list(replacements.values())

add_owned_batch async 

add_owned_batch(entries: list[tuple[Any, str, list[float], dict, str]]) -> None

Validate every native owner before changing vectors; never awaits.

Source code in src/symfonic/memory/backends/in_memory_vector.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
async def add_owned_batch(self, entries: list[tuple[Any, str, list[float], dict, str]]) -> None:
    """Validate every native owner before changing vectors; never awaits."""
    from symfonic.memory.errors import VectorOwnershipConflict
    proposed: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for scope, identifier, embedding, metadata, document in entries:
        if not _is_native_scope(scope):
            raise ValueError("owned vector batch requires native scopes")
        path = materialise_scope_path(scope)
        for row in (*self._store.get(scope.tenant_id, ()), *proposed[scope.tenant_id]):
            if row["id"] == identifier and metadata_scope_path(
                row["metadata"], scope.tenant_id, scope=scope,
            ) != path:
                raise VectorOwnershipConflict("Vector ID already belongs to a different scope")
        proposed[scope.tenant_id].append({
            "id": identifier, "embedding": embedding,
            "metadata": {**metadata, SCOPE_PATH_KEY: path}, "document": document,
        })
    for tenant, rows in proposed.items():
        replacements = {row["id"]: row for row in rows}
        self._store[tenant] = [
            row for row in self._store.get(tenant, ())
            if row["id"] not in replacements
        ] + list(replacements.values())

count async 

count(scope: StorageScope) -> int

Return the total number of stored vectors for the tenant.

Source code in src/symfonic/memory/backends/in_memory_vector.py 
193
194
195
196
197
198
199
200
201
202
203
async def count(self, scope: StorageScope) -> int:
    """Return the total number of stored vectors for the tenant."""
    entries = self._store.get(scope.tenant_id, [])
    if not _is_native_scope(scope):
        from symfonic.memory.backends.in_memory_ownership import native_owner
        return sum(not native_owner(entry["metadata"].get(SCOPE_PATH_KEY))
                   for entry in entries)
    return sum(is_visible(
        metadata_scope_path(entry["metadata"], scope.tenant_id, scope=scope),
        scope.ancestor_prefix_paths(),
    ) for entry in entries)

delete async 

delete(scope: StorageScope, ids: list[str]) -> None

Delete vectors by their IDs.

Source code in src/symfonic/memory/backends/in_memory_vector.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
async def delete(self, scope: StorageScope, ids: list[str]) -> None:
    """Delete vectors by their IDs."""
    from symfonic.memory.backends.in_memory_ownership import native_owner
    id_set = set(ids)
    if scope.tenant_id in self._store:
        self._store[scope.tenant_id] = [
            e for e in self._store[scope.tenant_id] if e["id"] not in id_set
            or (not _is_native_scope(scope) and native_owner(
                e["metadata"].get(SCOPE_PATH_KEY),
            ))
            or (_is_native_scope(scope) and not is_visible(
                metadata_scope_path(e["metadata"], scope.tenant_id, scope=scope),
                scope.ancestor_prefix_paths(),
            ))
        ]

search async 

search(scope: StorageScope, query_embedding: list[float], top_k: int = 5, *, allowed_scope_paths: tuple[str, ...] | None = None) -> list[dict[str, Any]]

Brute-force cosine similarity search.

v8.0: ALWAYS-ON prefix-isolation (§5.d) — only entries whose stored scope_path is a prefix of the query path are scored. Dual-reads pre-v8.0 metadata (no scope_path) as a 1-level root path.

Source code in src/symfonic/memory/backends/in_memory_vector.py 
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
async def search(
    self,
    scope: StorageScope,
    query_embedding: list[float],
    top_k: int = 5,
    *,
    allowed_scope_paths: tuple[str, ...] | None = None,
) -> list[dict[str, Any]]:
    """Brute-force cosine similarity search.

    v8.0: ALWAYS-ON prefix-isolation (§5.d) — only entries whose stored
    ``scope_path`` is a prefix of the query path are scored.  Dual-reads
    pre-v8.0 metadata (no ``scope_path``) as a 1-level root path.
    """
    entries = self._store.get(scope.tenant_id, [])
    prefixes = scope.ancestor_prefix_paths()
    from symfonic.memory.backends.in_memory_ownership import native_owner
    scored = []
    native = _is_native_scope(scope)
    for entry in entries:
        entry_path = metadata_scope_path(entry["metadata"], scope.tenant_id, scope=scope)
        if not native and native_owner(entry_path):
            continue
        if (native or allowed_scope_paths is None) and not is_visible(entry_path, prefixes):
            continue
        if allowed_scope_paths is not None and entry_path not in allowed_scope_paths:
            continue
        sim = _cosine_similarity(query_embedding, entry["embedding"])
        scored.append({
            "id": entry["id"],
            "score": sim,
            "metadata": entry["metadata"],
            "document": entry["document"],
        })
    scored.sort(key=lambda x: x["score"], reverse=True)
    return scored[:top_k]