Skip to content

symfonic.capabilities.memory.graph_admin

graph_admin

Reading the memory graph: edges, neighbourhoods, and a portable export.

MemoryAdminService answers what a scope remembers -- the published records. This answers how those records relate, which is a different question with a different backend call, and the one a graph view and a GDPR export both need.

Split rather than added because the two have different blast radii. Records are what a turn recalls; edges are what a maintenance job rewrites and a privacy request must include. An adopter mounting a read-only graph view should not have to hold the object that can publish and forget.

Why it takes a backend and not a store. Relationships live on the GraphBackend protocol -- query_edges, get_neighbors, traverse -- and the capability-native store deliberately exposes only the three memory ports. Reaching through the store for an edge would make every store owe a graph, when most deployments want one.

GraphAdminService 

GraphAdminService(graph: Any, records: Any = None)

Relationship reads for one deployment's graph backend.

Parameters:

Name Type Description Default
graph Any

the GraphBackend the memory store persists through.

 required
records Any

a MemoryAdminService, needed only by :meth:export. Absent, an export would contain relationships and no memories, which is a worse answer to a portability request than an error.

 None
Source code in src/symfonic/capabilities/memory/graph_admin.py 
53
54
55
56
57
58
59
60
61
62
def __init__(self, graph: Any, records: Any = None) -> None:
    """
    Args:
        graph: the ``GraphBackend`` the memory store persists through.
        records: a ``MemoryAdminService``, needed only by :meth:`export`.
            Absent, an export would contain relationships and no memories,
            which is a worse answer to a portability request than an error.
    """
    self._graph = graph
    self._records = records

edges async 

edges(scope: MemoryScope, *, limit: int = 500, offset: int = 0) -> list[dict[str, Any]]

One bounded page of relationships, in newest backend order.

Source code in src/symfonic/capabilities/memory/graph_admin.py 
64
65
66
67
68
69
70
71
async def edges(
    self, scope: MemoryScope, *, limit: int = 500, offset: int = 0
) -> list[dict[str, Any]]:
    """One bounded page of relationships, in newest backend order."""
    found = await self._graph.query_edges(
        tenant_scope(scope), {}, limit=limit, offset=offset
    )
    return [edge_body(edge) for edge in found]

export async 

export(scope: MemoryScope) -> dict[str, Any]

Everything this scope owns, as a portable document (GDPR Art. 20).

complete is part of the payload rather than an exception, because a partial export is still owed to the subject -- and a request that silently returned nine layers of ten would be a portability failure nobody could see. What could not be read is named.

Source code in src/symfonic/capabilities/memory/graph_admin.py 
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
async def export(self, scope: MemoryScope) -> dict[str, Any]:
    """Everything this scope owns, as a portable document (GDPR Art. 20).

    ``complete`` is part of the payload rather than an exception, because a
    partial export is still owed to the subject -- and a request that
    silently returned nine layers of ten would be a portability failure
    nobody could see. What could not be read is named.
    """
    if self._records is None:
        raise ValueError(
            "this GraphAdminService was built without a records service, so "
            "an export would carry relationships and no memories. Pass the "
            "MemoryAdminService that owns the same store."
        )
    failed: dict[str, str] = {}
    try:
        page = await self._records.record_page(scope, limit=MAX_CANDIDATE_LIMIT)
        memories = [_record_body(item.record) for item in page.memories]
        incomplete = page.dropped or page.degraded or page.unavailable
        if incomplete or len(memories) >= MAX_CANDIDATE_LIMIT:
            failed["memories"] = "bounded export is incomplete; use a paginated inventory"
    except Exception as unreadable:  # noqa: BLE001 - named, not swallowed
        memories, failed["memories"] = [], str(unreadable)[:200]
    try:
        relationships = await self.edges(scope, limit=100_000)
    except Exception as unreadable:  # noqa: BLE001
        relationships, failed["edges"] = [], str(unreadable)[:200]

    return {
        "scope_path": scope.path,
        "schema_version": EXPORT_SCHEMA_VERSION,
        "memories": memories,
        "edges": relationships,
        "complete": not failed,
        "failed": failed,
    }

neighborhood async 

neighborhood(scope: MemoryScope, node_id: str, *, depth: int = 1) -> dict[str, Any]

One node's neighbours, out to depth hops.

Depth 1 is the immediate neighbours, which is what a graph view expands on a click. Deeper traversals are the backend's job -- doing it here with repeated neighbour calls would issue a query per node and call it a traversal.

Source code in src/symfonic/capabilities/memory/graph_admin.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
async def neighborhood(
    self, scope: MemoryScope, node_id: str, *, depth: int = 1
) -> dict[str, Any]:
    """One node's neighbours, out to ``depth`` hops.

    Depth 1 is the immediate neighbours, which is what a graph view expands
    on a click. Deeper traversals are the backend's job -- doing it here
    with repeated neighbour calls would issue a query per node and call it
    a traversal.
    """
    if depth <= 1:
        found = await self._graph.get_neighbors(tenant_scope(scope), node_id)
    else:
        found = await self._graph.traverse(tenant_scope(scope), node_id, depth)
    return {
        "node_id": node_id,
        "depth": depth,
        "neighbors": [_node_body(node) for node in found],
    }

nodes async 

nodes(scope: MemoryScope, *, limit: int = 500, offset: int = 0) -> list[dict[str, Any]]

Drawable nodes owned by scope, including its descendants.

A graph browser and a recall answer are different views. Recall is ranked and ancestor-facing; a graph needs the endpoints of the edges it was given, including conversation descendants, or it silently drops almost every relationship as dangling.

Source code in src/symfonic/capabilities/memory/graph_admin.py 
 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
async def nodes(
    self, scope: MemoryScope, *, limit: int = 500, offset: int = 0
) -> list[dict[str, Any]]:
    """Drawable nodes owned by ``scope``, including its descendants.

    A graph browser and a recall answer are different views.  Recall is
    ranked and ancestor-facing; a graph needs the endpoints of the edges
    it was given, including conversation descendants, or it silently drops
    almost every relationship as dangling.
    """
    legacy = tenant_scope(scope)
    query = getattr(self._graph, "query_subtree", None)
    if limit <= 0:
        return []
    native_page = getattr(self._graph, "query_subtree_page", None)
    if callable(native_page):
        found = await native_page(legacy, {}, limit=limit, offset=offset)
        return [_node_body(node) for node in found]
    # GraphBackend's historical node contract has no offset. Fetch only
    # through the requested page and slice here; this bounds process memory
    # while keeping custom backends compatible. Native cursor pagination
    # can replace this without changing the public admin/API contract.
    through = offset + limit
    found = await (
        query(legacy, {}, limit=through)
        if callable(query)
        else self._graph.query_nodes(legacy, {}, limit=through)
    )
    return [_node_body(node) for node in found[offset:through]]

edge_body 

edge_body(edge: Any) -> dict[str, Any]

One relationship as JSON. Named fields, so a model gaining an internal one does not silently widen a privacy export.

Source code in src/symfonic/capabilities/memory/graph_admin.py 
35
36
37
38
39
40
41
42
43
44
45
def edge_body(edge: Any) -> dict[str, Any]:
    """One relationship as JSON. Named fields, so a model gaining an internal
    one does not silently widen a privacy export."""
    return {
        "id": str(getattr(edge, "id", "")),
        "source": str(getattr(edge, "source", "")),
        "target": str(getattr(edge, "target", "")),
        "relationship": str(getattr(edge, "relationship", "")),
        "weight": getattr(edge, "weight", None),
        "uses": getattr(edge, "uses", None),
    }