GraphMemoryStore(backend: GraphBackend, embedding_provider: EmbeddingProvider | None = None, embedding_cache: EmbeddingCache | None = None)
Bases: GraphQueryMixin
Graph-aware memory store wrapping a GraphBackend.
All operations enforce tenant isolation via TenantScope.
Access counts are automatically incremented on reads to support
frequency-based retrieval scoring.
v7.3 Item 13.1: optional auto-embed at the store level.
The graph-backed memory LAYERS (semantic / procedural /
prospective) wrap their own writes through
:func:~symfonic.memory.embeddings.auto_embed.maybe_embed so
layer-API callers benefit even when constructing layers without
the orchestrator. Consolidation phases that bypass the layers
(Item 12 EntityLinker calls GraphMemoryStore.add_node
directly) need the same auto-embed pathway -- that's what the
constructor kwargs here are for.
The double wrap (layer -> store) is idempotent: maybe_embed
short-circuits when node.embedding is already populated, so
a node that went through SemanticLayer.store_fact only
triggers one provider call regardless of how many layers of
wrapping it traverses.
Both kwargs default to None so legacy callers that build
GraphMemoryStore directly are byte-identical to v7.2.
Source code in symfonic/memory/graph/store.py
| def __init__(
self,
backend: GraphBackend,
embedding_provider: EmbeddingProvider | None = None,
embedding_cache: EmbeddingCache | None = None,
) -> None:
"""v7.3 Item 13.1: optional auto-embed at the store level.
The graph-backed memory LAYERS (semantic / procedural /
prospective) wrap their own writes through
:func:`~symfonic.memory.embeddings.auto_embed.maybe_embed` so
layer-API callers benefit even when constructing layers without
the orchestrator. Consolidation phases that bypass the layers
(Item 12 EntityLinker calls ``GraphMemoryStore.add_node``
directly) need the same auto-embed pathway -- that's what the
constructor kwargs here are for.
The double wrap (layer -> store) is idempotent: ``maybe_embed``
short-circuits when ``node.embedding`` is already populated, so
a node that went through ``SemanticLayer.store_fact`` only
triggers one provider call regardless of how many layers of
wrapping it traverses.
Both kwargs default to ``None`` so legacy callers that build
``GraphMemoryStore`` directly are byte-identical to v7.2.
"""
self._backend = backend
self._embedding_provider = embedding_provider
self._embedding_cache = embedding_cache
|
backend
property
What this store persists through. Named, so callers stop reaching.
add_edge
async
add_edge(scope: TenantScope, edge: MemoryEdge) -> MemoryEdge
Add an edge to the graph. Returns the persisted edge.
Source code in symfonic/memory/graph/store.py
| async def add_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
"""Add an edge to the graph. Returns the persisted edge."""
edge.tenant_id = scope.tenant_id
return await self._backend.add_edge(scope, edge)
|
add_node
async
add_node(scope: TenantScope, node: MemoryNode) -> MemoryNode
Add a node to the graph. Returns the persisted node.
v7.3 Item 13.1: when an embedding_provider is wired into
this store, nodes the caller left with embedding=None
auto-embed before the backend write. Caller-provided embeddings
always win (see :func:maybe_embed).
Source code in symfonic/memory/graph/store.py
| async def add_node(self, scope: TenantScope, node: MemoryNode) -> MemoryNode:
"""Add a node to the graph. Returns the persisted node.
v7.3 Item 13.1: when an ``embedding_provider`` is wired into
this store, nodes the caller left with ``embedding=None``
auto-embed before the backend write. Caller-provided embeddings
always win (see :func:`maybe_embed`).
"""
node.tenant_id = scope.tenant_id
if self._embedding_provider is not None:
# Late import keeps the symfonic.memory.embeddings package an
# optional surface for callers that never construct one.
from symfonic.memory.embeddings.auto_embed import maybe_embed
node = await maybe_embed(
node, self._embedding_provider, self._embedding_cache,
)
return await self._backend.add_node(scope, node)
|
bump_spreading
async
bump_spreading(scope: TenantScope, node_id: NodeId, *, include_retracted: bool = False) -> MemoryNode | None
Fetch a node and increment its spreading_access_count.
v6.2 T02: BFS-induced reads funnel through this helper instead
of :meth:get_node, so the Phase 1 recurrence signal can
distinguish direct fetches from one-shot spreading-activation
visits. Both counters carry independent monotonic semantics:
- access_count grows on get_node
- spreading_access_count grows on bump_spreading
The method is intentionally named bump_spreading (rather than
get_node_via_spreading or a private _bump_spreading) so
alternative traversal strategies in future sprints can call it
directly without reaching into private API.
Returns None if the node does not exist or belongs to another
tenant; otherwise returns the updated node.
Soft-retracted nodes return None by default (and are not bumped) so
a retracted false memory cannot re-enter context via spreading
activation -- notably as a live node's neighbor. Pass
include_retracted=True only for audit/maintenance reads.
Source code in symfonic/memory/graph/store.py
| async def bump_spreading(
self, scope: TenantScope, node_id: NodeId,
*, include_retracted: bool = False,
) -> MemoryNode | None:
"""Fetch a node and increment its spreading_access_count.
v6.2 T02: BFS-induced reads funnel through this helper instead
of :meth:`get_node`, so the Phase 1 recurrence signal can
distinguish direct fetches from one-shot spreading-activation
visits. Both counters carry independent monotonic semantics:
- access_count grows on ``get_node``
- spreading_access_count grows on ``bump_spreading``
The method is intentionally named ``bump_spreading`` (rather than
``get_node_via_spreading`` or a private ``_bump_spreading``) so
alternative traversal strategies in future sprints can call it
directly without reaching into private API.
Returns None if the node does not exist or belongs to another
tenant; otherwise returns the updated node.
Soft-retracted nodes return ``None`` by default (and are not bumped) so
a retracted false memory cannot re-enter context via spreading
activation -- notably as a live node's neighbor. Pass
``include_retracted=True`` only for audit/maintenance reads.
"""
node = await self._backend.get_node(scope, node_id)
if node is None:
return None
if not include_retracted and is_retracted(node.properties):
return None
updated = await self._backend.update_node(
scope,
node_id,
{
"spreading_access_count": node.spreading_access_count + 1,
"updated_at": datetime.now(UTC).isoformat(),
},
)
return updated
|
delete_edge
async
delete_edge(scope: TenantScope, edge_id: EdgeId) -> None
Delete a single edge by its ID.
Source code in symfonic/memory/graph/store.py
| async def delete_edge(self, scope: TenantScope, edge_id: EdgeId) -> None:
"""Delete a single edge by its ID."""
await self._backend.delete_edge(scope, edge_id)
|
delete_node
async
delete_node(scope: TenantScope, node_id: NodeId) -> None
Delete a node and cascade-delete all connected edges.
Source code in symfonic/memory/graph/store.py
| async def delete_node(self, scope: TenantScope, node_id: NodeId) -> None:
"""Delete a node and cascade-delete all connected edges."""
await self._backend.delete_node(scope, node_id, cascade=True)
|
get_neighbors
async
get_neighbors(scope: TenantScope, node_id: NodeId, relationship: str | None = None, *, include_retracted: bool = False) -> list[MemoryNode]
Get neighboring nodes, optionally filtered by relationship type.
Soft-retracted neighbours are excluded by default so a retracted node
cannot re-enter context as a graph neighbour of a live node. Audit /
maintenance callers pass include_retracted=True.
Source code in symfonic/memory/graph/store.py
| async def get_neighbors(
self,
scope: TenantScope,
node_id: NodeId,
relationship: str | None = None,
*,
include_retracted: bool = False,
) -> list[MemoryNode]:
"""Get neighboring nodes, optionally filtered by relationship type.
Soft-retracted neighbours are excluded by default so a retracted node
cannot re-enter context as a graph neighbour of a live node. Audit /
maintenance callers pass ``include_retracted=True``.
"""
neighbors = await self._backend.get_neighbors(
scope, node_id, relationship,
)
if include_retracted:
return neighbors
return [n for n in neighbors if not is_retracted(n.properties)]
|
get_node
async
get_node(scope: TenantScope, node_id: NodeId, *, include_retracted: bool = False) -> MemoryNode | None
Get a node by ID, incrementing its access count.
Returns None if the node does not exist or belongs to another tenant.
Soft-retracted nodes (see symfonic.memory.retraction for the
namespaced marker contract) return None by default so a
corrected/false-positive memory never re-enters context through a
direct fetch; the access count is NOT bumped for them. Audit /
erasure / prune callers pass include_retracted=True to reach them.
Source code in symfonic/memory/graph/store.py
| async def get_node(
self, scope: TenantScope, node_id: NodeId,
*, include_retracted: bool = False,
) -> MemoryNode | None:
"""Get a node by ID, incrementing its access count.
Returns None if the node does not exist or belongs to another tenant.
Soft-retracted nodes (see ``symfonic.memory.retraction`` for the
namespaced marker contract) return ``None`` by default so a
corrected/false-positive memory never re-enters context through a
direct fetch; the access count is NOT bumped for them. Audit /
erasure / prune callers pass ``include_retracted=True`` to reach them.
"""
node = await self._backend.get_node(scope, node_id)
if node is None:
return None
if not include_retracted and is_retracted(node.properties):
return None
# Increment access_count for frequency scoring
updated = await self._backend.update_node(
scope,
node_id,
{
"access_count": node.access_count + 1,
"updated_at": datetime.now(UTC).isoformat(),
},
)
return updated
|
list_edges
async
list_edges(scope: TenantScope, *, limit: int = 50, offset: int = 0, relationship: str | None = None) -> list[MemoryEdge]
List edges for the tenant with optional filtering and pagination.
Delegates directly to GraphBackend.query_edges — no node iteration.
Parameters:
| Name |
Type |
Description |
Default |
scope
|
TenantScope
|
|
required
|
limit
|
int
|
Maximum number of edges to return (default 50).
|
50
|
offset
|
int
|
Number of edges to skip for pagination (default 0).
|
0
|
relationship
|
str | None
|
If provided, only edges of this type are returned.
|
None
|
Source code in symfonic/memory/graph/store.py
| async def list_edges(
self,
scope: TenantScope,
*,
limit: int = 50,
offset: int = 0,
relationship: str | None = None,
) -> list[MemoryEdge]:
"""List edges for the tenant with optional filtering and pagination.
Delegates directly to ``GraphBackend.query_edges`` — no node iteration.
Args:
scope: Tenant isolation scope.
limit: Maximum number of edges to return (default 50).
offset: Number of edges to skip for pagination (default 0).
relationship: If provided, only edges of this type are returned.
"""
filters: dict[str, Any] | None = None
if relationship is not None:
filters = {"relationship": relationship}
return await self._backend.query_edges(scope, filters, limit, offset)
|
neighbor_probe
async
neighbor_probe(scope: TenantScope, node_id: NodeId, minimum: int = 1, relationship: str | None = None, target: str | None = None) -> bool
Answer a threshold/existence question without loading a neighborhood.
Source code in symfonic/memory/graph/store.py
| async def neighbor_probe(self, scope: TenantScope, node_id: NodeId, minimum: int = 1,
relationship: str | None = None, target: str | None = None) -> bool:
"""Answer a threshold/existence question without loading a neighborhood."""
probe = getattr(self._backend, "neighbor_probe", None)
if callable(probe):
return bool(await probe(scope, str(node_id), minimum, relationship, target))
neighbors = await self.get_neighbors(scope, node_id, relationship)
return sum(target is None or str(node.id) == target for node in neighbors) >= minimum
|
related_candidates(scope: TenantScope, cue: str) -> list[MemoryNode]
Bounded exact-owner semantic candidates for offline comparison.
Source code in symfonic/memory/graph/store.py
| async def related_candidates(self, scope: TenantScope, cue: str) -> list[MemoryNode]:
"""Bounded exact-owner semantic candidates for offline comparison."""
select = getattr(self._backend, "select_candidates", None)
if not callable(select):
return []
from symfonic.memory.candidates import CandidateRequest
page = await select(scope, CandidateRequest(
cue=cue, layers=("semantic",), limit=64, profile_slots=0, exact_scope=True,
))
return list(page.nodes)
|
update_node
async
update_node(scope: TenantScope, node_id: NodeId, properties: dict[str, object]) -> MemoryNode
Partial update of node properties.
Source code in symfonic/memory/graph/store.py
| async def update_node(
self, scope: TenantScope, node_id: NodeId, properties: dict[str, object]
) -> MemoryNode:
"""Partial update of node properties."""
return await self._backend.update_node(scope, node_id, properties)
|
upsert_edge
async
upsert_edge(scope: TenantScope, edge: MemoryEdge) -> MemoryEdge
Insert edge or increment weight if it already exists.
Uniqueness key: (tenant_id, source, target, relationship).
Each repeated call on the same pair increments weight by 1.
Source code in symfonic/memory/graph/store.py
| async def upsert_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
"""Insert edge or increment weight if it already exists.
Uniqueness key: (tenant_id, source, target, relationship).
Each repeated call on the same pair increments weight by 1.
"""
edge.tenant_id = scope.tenant_id
return await self._backend.upsert_edge(scope, edge)
|
upsert_edge_props
async
upsert_edge_props(scope: TenantScope, edge_id: EdgeId, props: dict[str, Any]) -> MemoryEdge | None
Update edge properties in-place.
Attempts backend.update_edge when available; otherwise falls back
to delete+re-add. Returns the updated edge, or None on failure.
Source code in symfonic/memory/graph/store.py
| async def upsert_edge_props(
self,
scope: TenantScope,
edge_id: EdgeId,
props: dict[str, Any],
) -> MemoryEdge | None:
"""Update edge properties in-place.
Attempts backend.update_edge when available; otherwise falls back
to delete+re-add. Returns the updated edge, or None on failure.
"""
if hasattr(self._backend, "update_edge"):
return await self._backend.update_edge(scope, edge_id, props)
# Fallback: find edge, apply props, delete, re-add
edges = await self._backend.query_edges(scope, None, 10_000, 0)
target_edge = None
for e in edges:
if str(e.id) == str(edge_id):
target_edge = e
break
if target_edge is None:
return None
valid_props = {k: v for k, v in props.items() if hasattr(target_edge, k)}
updated_edge = target_edge.model_copy(update=valid_props)
# Delete-then-add (v8.7.2): the previous add-then-delete order re-added
# the edge under the SAME id and then deleted that id, permanently
# destroying the edge — on InMemory the delete popped the just-written
# key, and on Postgres ``add_edge`` is ``ON CONFLICT DO NOTHING`` so the
# write was a no-op before the delete removed the original. Deleting
# first, then re-adding the mutated copy, is correct on every backend.
await self._backend.delete_edge(scope, edge_id)
await self._backend.add_edge(scope, updated_edge)
return updated_edge
|