symfonic.memory.graph.traversal

traversal

GraphTraversal -- BFS traversal and path queries for graph proximity scoring.

Provides breadth-first search, shortest-path computation, and the graph_proximity metric used in the four-signal scoring formula.

v6.1 T03 (deprecated by v6.2 T02): bfs used to route through GraphMemoryStore.get_node (bumping access_count on every traversal hit). v6.2 decomposed the counter: BFS now bumps spreading_access_count via GraphMemoryStore.bump_spreading. Direct fetches keep using get_node (= direct access_count bump). This lets Phase 1 strengthen weight the two signals independently via FrameworkConfig.phase1_spreading_weight (default 0.5 -- a node that was spread-activated twice contributes like one direct fetch, matching the v6.1.x observable behaviour).

shortest_path / graph_proximity deliberately do NOT bump either counter -- they run inside the scoring loop, and bumping there would pollute the frequency signal with one-query-N-bumps noise.

GraphTraversal

GraphTraversal(store: GraphMemoryStore)

Graph traversal utilities operating on a GraphMemoryStore.

All methods enforce tenant isolation through the required TenantScope.

Source code in src/symfonic/memory/graph/traversal.py

def __init__(self, store: GraphMemoryStore) -> None:
    self._store = store

bfs async

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

Breadth-first traversal returning all reachable nodes within depth.

v6.2 T02: visited nodes are fetched via GraphMemoryStore.bump_spreading which bumps spreading_access_count on every hit. Phase 1 strengthen combines it with the direct access_count using FrameworkConfig.phase1_spreading_weight so callers can tune how strongly one-shot spreading activation contributes to recurrence.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
start	NodeId	Starting node ID.	required
max_depth	int	Maximum traversal depth (inclusive).	3

Returns:

Type	Description
list[MemoryNode]	List of reachable MemoryNodes (including the start node if it exists).

Source code in src/symfonic/memory/graph/traversal.py

async def bfs(
    self,
    scope: TenantScope,
    start: NodeId,
    max_depth: int = 3,
) -> list[MemoryNode]:
    """Breadth-first traversal returning all reachable nodes within depth.

    v6.2 T02: visited nodes are fetched via
    ``GraphMemoryStore.bump_spreading`` which bumps
    ``spreading_access_count`` on every hit. Phase 1
    ``strengthen`` combines it with the direct ``access_count``
    using ``FrameworkConfig.phase1_spreading_weight`` so callers
    can tune how strongly one-shot spreading activation contributes
    to recurrence.

    Args:
        scope: Tenant scope for isolation.
        start: Starting node ID.
        max_depth: Maximum traversal depth (inclusive).

    Returns:
        List of reachable MemoryNodes (including the start node if it exists).
    """
    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)

        # v6.2 T02: route through GraphMemoryStore.bump_spreading so
        # spreading_access_count is bumped on every traversal hit.
        # This decomposes the v6.1 signal: direct fetches keep
        # bumping access_count (via store.get_node); BFS visits
        # bump spreading_access_count only.
        node = await self._store.bump_spreading(scope, NodeId(current_id))
        if node is not None:
            result.append(node)

        if depth < max_depth:
            neighbors = await self._store.get_neighbors(
                scope, NodeId(current_id)
            )
            for neighbor in neighbors:
                nid = str(neighbor.id)
                if nid not in visited:
                    queue.append((nid, depth + 1))

    return result

graph_proximity async

graph_proximity(
    scope: TenantScope, node_a: NodeId, node_b: NodeId
) -> float

Compute graph proximity between two nodes.

Returns 1.0 / (1 + shortest_path_length), or 0.0 if no path exists. Same node returns 1.0. Direct neighbors return 0.5.

This metric feeds the graph_proximity component of the scoring formula.

Source code in src/symfonic/memory/graph/traversal.py

async def graph_proximity(
    self,
    scope: TenantScope,
    node_a: NodeId,
    node_b: NodeId,
) -> float:
    """Compute graph proximity between two nodes.

    Returns 1.0 / (1 + shortest_path_length), or 0.0 if no path exists.
    Same node returns 1.0. Direct neighbors return 0.5.

    This metric feeds the graph_proximity component of the scoring formula.
    """
    path = await self.shortest_path(scope, node_a, node_b)
    if path is None:
        return 0.0
    # path includes both endpoints, so hops = len(path) - 1
    hops = len(path) - 1
    return 1.0 / (1.0 + hops)

shortest_path async

shortest_path(
    scope: TenantScope, source: NodeId, target: NodeId
) -> list[NodeId] | None

Find the shortest path between two nodes using BFS.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
source	NodeId	Start node ID.	required
target	NodeId	End node ID.	required

Returns:

Type	Description
list[NodeId] | None	List of NodeIds forming the shortest path (inclusive of both
list[NodeId] | None	endpoints), or None if no path exists.

Source code in src/symfonic/memory/graph/traversal.py

async def shortest_path(
    self,
    scope: TenantScope,
    source: NodeId,
    target: NodeId,
) -> list[NodeId] | None:
    """Find the shortest path between two nodes using BFS.

    Args:
        scope: Tenant scope for isolation.
        source: Start node ID.
        target: End node ID.

    Returns:
        List of NodeIds forming the shortest path (inclusive of both
        endpoints), or None if no path exists.
    """
    if str(source) == str(target):
        return [source]

    visited: set[str] = set()
    queue: deque[list[str]] = deque([[str(source)]])

    while queue:
        path = queue.popleft()
        current = path[-1]

        if current in visited:
            continue
        visited.add(current)

        neighbors = await self._store.get_neighbors(scope, NodeId(current))
        for neighbor in neighbors:
            nid = str(neighbor.id)
            if nid == str(target):
                return [NodeId(p) for p in path] + [target]
            if nid not in visited:
                queue.append(path + [nid])

    return None