symfonic.memory.backends.postgres_graph

postgres_graph

PostgreSQL graph backend using JSONB storage.

Implements GraphBackend protocol with asyncpg for all database I/O. All operations enforce tenant isolation via WHERE tenant_id = $N.

asyncpg must be installed.

pip install symfonic-core[postgres]

PostgresGraphBackend

PostgresGraphBackend(pool: PostgresPoolManager)

GraphBackend backed by PostgreSQL with JSONB storage.

All operations enforce tenant isolation via WHERE tenant_id = $N.

Source code in src/symfonic/memory/backends/postgres_graph.py

def __init__(self, pool: PostgresPoolManager) -> None:
    _require_asyncpg("PostgresGraphBackend")
    self._pool = pool

backfill_scope_paths async

backfill_scope_paths() -> int

Eagerly backfill NULL scope_path rows (design §6.b migration).

Pre-v8.0 rows have scope_path IS NULL; they read as a 1-level root path tenant\x1f<tenant_id> via the query-time dual-read, so this eager backfill is OPTIONAL (online-safe, idempotent) — it just materialises the same value into the column so the B-tree index and the scorer's dual-read agree. Returns the number of rows updated. Uses a parameterised expression so adopter tenant_ids cannot inject.

Source code in src/symfonic/memory/backends/postgres_graph.py

async def backfill_scope_paths(self) -> int:
    """Eagerly backfill NULL ``scope_path`` rows (design §6.b migration).

    Pre-v8.0 rows have ``scope_path IS NULL``; they read as a 1-level root
    path ``tenant\\x1f<tenant_id>`` via the query-time dual-read, so this
    eager backfill is OPTIONAL (online-safe, idempotent) — it just
    materialises the same value into the column so the B-tree index and
    the scorer's dual-read agree.  Returns the number of rows updated.
    Uses a parameterised expression so adopter tenant_ids cannot inject.
    """
    # The 1-level root path is f"tenant\x1f{tenant_id}"; build it in SQL.
    delim = "\x1f"
    conn = await self._pool.acquire()
    try:
        result = await conn.execute(
            "UPDATE memory_nodes "
            "SET scope_path = 'tenant' || $1 || tenant_id "
            "WHERE scope_path IS NULL",
            delim,
        )
    finally:
        await self._pool.release(conn)
    # asyncpg returns e.g. "UPDATE 42"
    try:
        return int(str(result).split()[-1])
    except (ValueError, IndexError):
        return 0

ensure_schema async

ensure_schema() -> None

Create graph tables (nodes + edges) and their indexes.

Does NOT create the memory_vectors table — that is the responsibility of :class:PostgresVectorBackend.ensure_schema.

Source code in src/symfonic/memory/backends/postgres_graph.py

async def ensure_schema(self) -> None:
    """Create graph tables (nodes + edges) and their indexes.

    Does NOT create the ``memory_vectors`` table — that is the
    responsibility of :class:`PostgresVectorBackend.ensure_schema`.
    """
    conn = await self._pool.acquire()
    try:
        await conn.execute("CREATE EXTENSION IF NOT EXISTS vector;")
        await conn.execute(NODES_DDL)
        await conn.execute(EDGES_DDL)
        # v8.0 online-safe upgrade: add scope_path to pre-existing tables.
        await conn.execute(NODES_SCOPE_PATH_MIGRATION)
        # Online-safe upgrade: add edge model fields to pre-existing tables.
        await conn.execute(EDGES_FIELDS_MIGRATION)
        for idx in GRAPH_INDEXES_DDL:
            await conn.execute(idx)
    finally:
        await self._pool.release(conn)

query_edges async

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

Query edges directly from memory_edges table with pagination.

Supports optional relationship filter. All other keys in filters are ignored to avoid SQL injection risk.

Source code in src/symfonic/memory/backends/postgres_graph.py

async def query_edges(
    self,
    scope: TenantScope,
    filters: dict[str, Any] | None = None,
    limit: int = 50,
    offset: int = 0,
) -> list[MemoryEdge]:
    """Query edges directly from memory_edges table with pagination.

    Supports optional ``relationship`` filter. All other keys in ``filters``
    are ignored to avoid SQL injection risk.
    """
    conditions: list[str] = ["tenant_id=$1"]
    params: list[Any] = [scope.tenant_id]

    relationship = (filters or {}).get("relationship")
    if relationship is not None:
        params.append(relationship)
        conditions.append(f"relationship=${len(params)}")

    params.append(limit)
    limit_ph = f"${len(params)}"
    params.append(offset)
    offset_ph = f"${len(params)}"

    sql = (
        f"SELECT * FROM memory_edges WHERE {' AND '.join(conditions)}"
        f" ORDER BY created_at DESC"
        f" LIMIT {limit_ph} OFFSET {offset_ph}"
    )

    conn = await self._pool.acquire()
    try:
        rows = await conn.fetch(sql, *params)
    finally:
        await self._pool.release(conn)

    return [_row_to_edge(row) for row in rows]

traverse async

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

BFS graph traversal up to max_depth hops from start node.

Source code in src/symfonic/memory/backends/postgres_graph.py

async def traverse(
    self, scope: TenantScope, start: NodeId, max_depth: int
) -> list[MemoryNode]:
    """BFS graph traversal up to max_depth hops from start node."""
    visited: set[str] = set()
    queue: deque[tuple[str, int]] = deque([(str(start), 0)])
    result: list[MemoryNode] = []
    while queue:
        current_id, depth = queue.popleft()
        if current_id in visited:
            continue
        visited.add(current_id)
        node = await self.get_node(scope, NodeId(current_id))
        if node is not None:
            result.append(node)
        if depth < max_depth:
            neighbors = await self.get_neighbors(scope, NodeId(current_id))
            for nb in neighbors:
                if str(nb.id) not in visited:
                    queue.append((str(nb.id), depth + 1))
    return result

upsert_edge async

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

Insert edge or increment weight by 1 if the same edge already exists.

Uniqueness is determined by (tenant_id, source, target, relationship). The unique index idx_memory_edges_upsert must exist for ON CONFLICT to resolve correctly (created by ensure_schema).

Source code in src/symfonic/memory/backends/postgres_graph.py

async def upsert_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
    """Insert edge or increment weight by 1 if the same edge already exists.

    Uniqueness is determined by (tenant_id, source, target, relationship).
    The unique index ``idx_memory_edges_upsert`` must exist for ON CONFLICT
    to resolve correctly (created by ``ensure_schema``).
    """
    edge.tenant_id = scope.tenant_id
    conn = await self._pool.acquire()
    try:
        row = await conn.fetchrow(
            """
            INSERT INTO memory_edges
                (id, source, target, relationship, tenant_id, weight,
                 created_at, uses, provenance, source_episodic_ids,
                 scope_path)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::jsonb, $11)
            ON CONFLICT (tenant_id, source, target, relationship)
            DO UPDATE SET weight = memory_edges.weight + 1
            RETURNING *
            """,
            str(edge.id), str(edge.source), str(edge.target),
            edge.relationship, scope.tenant_id, edge.weight, edge.created_at,
            edge.uses, edge.provenance, json.dumps(edge.source_episodic_ids),
            materialise_scope_path(scope),
        )
    finally:
        await self._pool.release(conn)

    if row is None:  # pragma: no cover — RETURNING always yields a row
        return edge
    return _row_to_edge(row)

matches_extra

matches_extra(
    node: MemoryNode, filters: dict[str, Any]
) -> bool

Apply arbitrary property filters not handled by SQL.

Source code in src/symfonic/memory/backends/postgres_graph.py

def matches_extra(node: MemoryNode, filters: dict[str, Any]) -> bool:
    """Apply arbitrary property filters not handled by SQL."""
    for key, value in filters.items():
        if hasattr(node, key):
            if getattr(node, key) != value:
                return False
        elif key not in node.properties or node.properties[key] != value:
            return False
    return True

row_to_node

row_to_node(row: Any) -> MemoryNode

Convert an asyncpg Record to a MemoryNode.

v6.2 T02 note: spreading_access_count has no dedicated column (no schema migration was required by the v6.2 carve-out). It is round-tripped through the properties JSONB under the reserved key __spreading_access_count; on read we lift the field out of properties so the Python model remains clean, and on write _node_properties_jsonb serialises it back in.

v7.2.1 hotfix: the embedding pgvector column is now hydrated onto MemoryNode.embedding. Prior versions silently dropped the column on read, which collapsed RetrievalScorer.semantic_similarity (weight 0.4) to zero for every Postgres-loaded graph candidate.

Source code in src/symfonic/memory/backends/postgres_graph.py

def row_to_node(row: Any) -> MemoryNode:
    """Convert an asyncpg Record to a MemoryNode.

    v6.2 T02 note: ``spreading_access_count`` has no dedicated column
    (no schema migration was required by the v6.2 carve-out). It is
    round-tripped through the ``properties`` JSONB under the reserved
    key ``__spreading_access_count``; on read we lift the field out of
    properties so the Python model remains clean, and on write
    ``_node_properties_jsonb`` serialises it back in.

    v7.2.1 hotfix: the ``embedding`` pgvector column is now hydrated
    onto ``MemoryNode.embedding``. Prior versions silently dropped the
    column on read, which collapsed ``RetrievalScorer.semantic_similarity``
    (weight 0.4) to zero for every Postgres-loaded graph candidate.
    """
    props = row["properties"]
    if isinstance(props, str):
        props = json.loads(props)
    props = dict(props)
    spreading = int(props.pop("__spreading_access_count", 0) or 0)
    # v8.0: if the dedicated scope_path column is populated but the JSONB
    # bag lacks it (e.g. a row written pre-v8.0 then backfilled), lift the
    # column value so the scorer dual-read sees a single source of truth.
    col_scope_path = _row_get(row, SCOPE_PATH_KEY)
    if col_scope_path and not props.get(SCOPE_PATH_KEY):
        props[SCOPE_PATH_KEY] = col_scope_path
    embedding = _decode_embedding(_row_get(row, "embedding"))
    return MemoryNode(
        id=NodeId(row["id"]),
        layer=MemoryLayer(row["layer"]),
        tenant_id=row["tenant_id"],
        label=row["label"],
        properties=props,
        embedding=embedding,
        created_at=row["created_at"],
        updated_at=row["updated_at"],
        access_count=row["access_count"],
        spreading_access_count=spreading,
        importance=row["importance"],
    )