Skip to content

symfonic.memory.backends.in_memory

in_memory

In-memory backends for testing and prototyping.

Zero-dependency GraphBackend and VectorBackend over plain dicts, scoped by tenant_id.

InMemoryGraphBackend 

InMemoryGraphBackend()

GraphBackend over dicts keyed by (tenant_id, node_id/edge_id).

Testing and local development only.

Source code in src/symfonic/memory/backends/in_memory.py 
43
44
45
46
def __init__(self) -> None:
    self._nodes: dict[str, dict[str, MemoryNode]] = defaultdict(dict)
    self._edges: dict[str, dict[str, MemoryEdge]] = defaultdict(dict)
    self._workset_cursors: dict[Any, Any] = {}

add_edge async 

add_edge(scope: TenantScope, edge: MemoryEdge) -> MemoryEdge

Add an edge to the in-memory store.

Source code in src/symfonic/memory/backends/in_memory.py 
111
112
113
114
115
async def add_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
    """Add an edge to the in-memory store."""
    edge.tenant_id = scope.tenant_id
    self._edges[scope.tenant_id][str(edge.id)] = edge
    return edge

add_node async 

add_node(scope: TenantScope, node: MemoryNode) -> MemoryNode

Add a node to the in-memory store.

v8.0: stamps the materialised scope_path into the node's property bag (zero-migration) so the prefix-isolation filter (§5.d) can be enforced on every read.

Source code in src/symfonic/memory/backends/in_memory.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
async def add_node(self, scope: TenantScope, node: MemoryNode) -> MemoryNode:
    """Add a node to the in-memory store.

    v8.0: stamps the materialised ``scope_path`` into the node's
    property bag (zero-migration) so the prefix-isolation filter (§5.d)
    can be enforced on every read.
    """
    node.tenant_id = scope.tenant_id
    node.properties[SCOPE_PATH_KEY] = materialise_scope_path(scope)
    self._nodes[scope.tenant_id][str(node.id)] = node
    return node

delete_edge async 

delete_edge(scope: TenantScope, edge_id: EdgeId) -> None

Delete a single edge by its ID.

Source code in src/symfonic/memory/backends/in_memory.py 
239
240
241
242
async def delete_edge(self, scope: TenantScope, edge_id: EdgeId) -> None:
    """Delete a single edge by its ID."""
    tenant_edges = self._edges.get(scope.tenant_id, {})
    tenant_edges.pop(str(edge_id), None)

delete_node async 

delete_node(scope: TenantScope, node_id: NodeId, cascade: bool = False) -> None

Delete a node, optionally cascading to connected edges.

Source code in src/symfonic/memory/backends/in_memory.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
async def delete_node(
    self, scope: TenantScope, node_id: NodeId, cascade: bool = False
) -> None:
    """Delete a node, optionally cascading to connected edges."""
    tenant_nodes = self._nodes.get(scope.tenant_id, {})
    node = tenant_nodes.get(str(node_id))
    if node is None:
        return
    # v8.7.7: scope-gated by prefix visibility like the PG delete_node and
    # the get_node read filter, so a sibling sub-scope cannot delete this.
    stored = stored_scope_path(node.properties, scope.tenant_id)
    if not is_visible(stored, scope.ancestor_prefix_paths()):
        return
    tenant_nodes.pop(str(node_id), None)
    if cascade:
        tenant_edges = self._edges.get(scope.tenant_id, {})
        to_remove = [
            eid
            for eid, edge in tenant_edges.items()
            if str(edge.source) == str(node_id)
            or str(edge.target) == str(node_id)
        ]
        for eid in to_remove:
            del tenant_edges[eid]

delete_subtree async 

delete_subtree(scope: TenantScope) -> int

Erase owned subtree, edges and cursors in one non-yielding operation.

Source code in src/symfonic/memory/backends/in_memory.py 
244
245
246
247
async def delete_subtree(self, scope: TenantScope) -> int:
    """Erase owned subtree, edges and cursors in one non-yielding operation."""
    from symfonic.memory.backends.in_memory_deletion import delete_subtree
    return delete_subtree(self, scope)

get_neighbors async 

get_neighbors(scope: TenantScope, node_id: NodeId, relationship: str | None = None) -> list[MemoryNode]

Get nodes connected to the given node by edges.

Source code in src/symfonic/memory/backends/in_memory.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
async def get_neighbors(
    self,
    scope: TenantScope,
    node_id: NodeId,
    relationship: str | None = None,
) -> list[MemoryNode]:
    """Get nodes connected to the given node by edges."""
    tenant_edges = self._edges.get(scope.tenant_id, {})
    tenant_nodes = self._nodes.get(scope.tenant_id, {})
    neighbor_ids: set[str] = set()
    for edge in tenant_edges.values():
        if relationship and edge.relationship != relationship:
            continue
        if str(edge.source) == str(node_id):
            neighbor_ids.add(str(edge.target))
        elif str(edge.target) == str(node_id):
            neighbor_ids.add(str(edge.source))
    return [
        tenant_nodes[nid]
        for nid in neighbor_ids
        if nid in tenant_nodes and self._node_visible(scope, tenant_nodes[nid])
    ]

get_node async 

get_node(scope: TenantScope, node_id: NodeId) -> MemoryNode | None

Get a node by ID within the scope's visible prefix chain.

Source code in src/symfonic/memory/backends/in_memory.py 
117
118
119
120
121
122
123
124
async def get_node(
    self, scope: TenantScope, node_id: NodeId
) -> MemoryNode | None:
    """Get a node by ID within the scope's visible prefix chain."""
    node = self._nodes.get(scope.tenant_id, {}).get(str(node_id))
    if node is None or not self._node_visible(scope, node):
        return None
    return node

query_edges async 

query_edges(scope: TenantScope, filters: dict[str, Any] | None = None, limit: int = 50, offset: int = 0) -> list[MemoryEdge]

Query edges with optional relationship filter and pagination.

Source code in src/symfonic/memory/backends/in_memory.py 
264
265
266
267
268
269
270
271
272
273
async def query_edges(
    self,
    scope: TenantScope,
    filters: dict[str, Any] | None = None,
    limit: int = 50,
    offset: int = 0,
) -> list[MemoryEdge]:
    """Query edges with optional relationship filter and pagination."""
    from symfonic.memory.backends.in_memory_queries import query_edges
    return query_edges(self._edges.get(scope.tenant_id, {}).values(), filters, limit, offset)

query_nodes async 

query_nodes(scope: TenantScope, filters: dict[str, Any], limit: int | None = 50) -> list[MemoryNode]

Filter nodes by properties within the tenant scope.

limit=None returns every matching node (no truncation); limit=0 returns no rows (v8.7.1 — unified across backends).

Source code in src/symfonic/memory/backends/in_memory.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
async def query_nodes(
    self, scope: TenantScope, filters: dict[str, Any], limit: int | None = 50
) -> list[MemoryNode]:
    """Filter nodes by properties within the tenant scope.

    ``limit=None`` returns every matching node (no truncation);
    ``limit=0`` returns no rows (v8.7.1 — unified across backends).
    """
    if limit == 0:
        return []
    tenant_nodes = self._nodes.get(scope.tenant_id, {})
    results: list[MemoryNode] = []
    for node in tenant_nodes.values():
        if not self._node_visible(scope, node):
            continue
        if self._matches_filters(node, filters):
            results.append(node)
            if limit is not None and len(results) >= limit:
                break
    return results

query_subtree async 

query_subtree(scope: TenantScope, filters: dict[str, Any], limit: int | None = None) -> list[MemoryNode]

Filter nodes at scope or below it (the descendant read).

Source code in src/symfonic/memory/backends/in_memory.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
async def query_subtree(
    self, scope: TenantScope, filters: dict[str, Any], limit: int | None = None
) -> list[MemoryNode]:
    """Filter nodes at ``scope`` or below it (the descendant read)."""
    if limit == 0:
        return []
    root = materialise_scope_path(scope)
    results: list[MemoryNode] = []
    for node in self._nodes.get(scope.tenant_id, {}).values():
        stored = stored_scope_path(node.properties, node.tenant_id)
        if not is_in_subtree(stored, root):
            continue
        if self._matches_filters(node, filters):
            results.append(node)
            if limit is not None and len(results) >= limit:
                break
    return results

transaction 

transaction() -> Any

All of a batch, or none of it -- see :mod:.in_memory_transaction.

Source code in src/symfonic/memory/backends/in_memory.py 
38
39
40
41
def transaction(self) -> Any:
    """All of a batch, or none of it -- see :mod:`.in_memory_transaction`."""
    from symfonic.memory.backends.in_memory_transaction import snapshot
    return snapshot(self)

traverse async 

traverse(scope: TenantScope, start: NodeId, max_depth: int) -> list[MemoryNode]

BFS traversal from start node up to max_depth.

Source code in src/symfonic/memory/backends/in_memory.py 
149
150
151
152
153
154
async def traverse(
    self, scope: TenantScope, start: NodeId, max_depth: int
) -> list[MemoryNode]:
    """BFS traversal from start node up to max_depth."""
    from symfonic.memory.backends.in_memory_queries import traverse
    return await traverse(self, scope, start, max_depth)

update_node async 

update_node(scope: TenantScope, node_id: NodeId, updates: dict[str, Any]) -> MemoryNode

Partial update of node properties.

Source code in src/symfonic/memory/backends/in_memory.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
async def update_node(
    self, scope: TenantScope, node_id: NodeId, updates: dict[str, Any]
) -> MemoryNode:
    """Partial update of node properties."""
    tenant_nodes = self._nodes.get(scope.tenant_id, {})
    node = tenant_nodes.get(str(node_id))
    if node is None:
        raise KeyError(f"Node {node_id} not found for tenant {scope.tenant_id}")
    for key, value in updates.items():
        if key == "updated_at" and isinstance(value, str):
            value = datetime.fromisoformat(value)
        if hasattr(node, key):
            object.__setattr__(node, key, value)
        else:
            node.properties[key] = value
    if "updated_at" not in updates:
        node.updated_at = datetime.now(UTC)
    return node

upsert_edge async 

upsert_edge(scope: TenantScope, edge: MemoryEdge) -> MemoryEdge

Insert edge or increment weight by 1 if (tenant, source, target, rel) matches.

Source code in src/symfonic/memory/backends/in_memory.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
async def upsert_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
    """Insert edge or increment weight by 1 if (tenant, source, target, rel) matches."""
    edge.tenant_id = scope.tenant_id
    tenant_edges = self._edges[scope.tenant_id]
    for existing in tenant_edges.values():
        if (
            str(existing.source) == str(edge.source)
            and str(existing.target) == str(edge.target)
            and existing.relationship == edge.relationship
        ):
            object.__setattr__(existing, "weight", existing.weight + 1)
            return existing
    tenant_edges[str(edge.id)] = edge
    return edge

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 
49
50
def __init__(self) -> None:
    self._store: dict[str, list[dict[str, Any]]] = defaultdict(list)

add async 

add(scope: TenantScope, 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 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
async def add(
    self,
    scope: TenantScope,
    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)
    for i, doc_id in enumerate(ids):
        meta = dict(metadatas[i]) if i < len(metadatas) else {}
        meta.setdefault(SCOPE_PATH_KEY, scope_path)
        self._store[scope.tenant_id].append({
            "id": doc_id,
            "embedding": embeddings[i],
            "metadata": meta,
            "document": documents[i] if i < len(documents) else "",
        })

count async 

count(scope: TenantScope) -> int

Return the total number of stored vectors for the tenant.

Source code in src/symfonic/memory/backends/in_memory_vector.py 
122
123
124
async def count(self, scope: TenantScope) -> int:
    """Return the total number of stored vectors for the tenant."""
    return len(self._store.get(scope.tenant_id, []))

delete async 

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

Delete vectors by their IDs.

Source code in src/symfonic/memory/backends/in_memory_vector.py 
105
106
107
108
109
110
111
async def delete(self, scope: TenantScope, ids: list[str]) -> None:
    """Delete vectors by their IDs."""
    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
        ]

search async 

search(scope: TenantScope, query_embedding: list[float], top_k: int = 5) -> 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 
 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 search(
    self,
    scope: TenantScope,
    query_embedding: list[float],
    top_k: int = 5,
) -> 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()
    scored = []
    for entry in entries:
        entry_path = metadata_scope_path(entry["metadata"], scope.tenant_id)
        if not is_visible(entry_path, prefixes):
            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]