async def create_cooccurrence_edges(
graph: GraphMemoryStore,
scope: TenantScope,
recent_nodes: list[MemoryNode],
) -> int:
"""Create CO_OCCURRED edges between nodes updated within 1 hour.
Weight reinforcement: repeated co-occurrence increments the edge weight
rather than inserting a duplicate.
Degree cap: a node that already has >= MAX_CO_OCCUR_DEGREE CO_OCCURRED
neighbours will not receive new CO_OCCURRED edges (existing edges still
get reinforced).
"""
window = timedelta(hours=CO_OCCUR_WINDOW_HOURS)
created = 0
# Conversation transcripts are evidence from which durable memories are
# derived, not concepts in the knowledge graph. Linking both the working
# and episodic copy creates edges between whole paragraphs (and between
# two copies of the same sentence), producing a dense graph that cannot
# improve recall. Entity linking separately reads episodics and emits
# canonical entity relations; co-occurrence operates on durable nodes.
nodes_with_ts = [
n
for n in recent_nodes
if n.updated_at is not None
and n.layer not in {MemoryLayer.WORKING, MemoryLayer.EPISODIC}
# EntityLinker already gives canonical entities explicit MENTIONS
# relationships. Treating its outputs as raw co-occurrence inputs on
# the next cycle creates a redundant clique one cycle late, so a graph
# that was stable after DEEP mutates on an unchanged replay.
and not n.label.startswith(f"{ENTITY_LABEL_PREFIX}:")
]
# Cache degree counts to avoid redundant neighbour fetches inside the loop.
co_degree: dict[str, int] = {}
async def _co_degree(node_id: NodeId) -> int:
key = str(node_id)
if key not in co_degree:
from symfonic.capabilities.memory.phases.adjacency import neighbor_probe
saturated = await neighbor_probe(
graph, scope, node_id, minimum=MAX_CO_OCCUR_DEGREE, relationship="CO_OCCURRED"
)
co_degree[key] = MAX_CO_OCCUR_DEGREE if saturated else 0
return co_degree[key]
for i, a in enumerate(nodes_with_ts):
for b in nodes_with_ts[i + 1 :]:
assert a.updated_at is not None and b.updated_at is not None
if a.label.strip().casefold() == b.label.strip().casefold():
continue
if abs((a.updated_at - b.updated_at).total_seconds()) > window.total_seconds():
continue
try:
# Check degree cap for both endpoints before creating a new edge.
deg_a = await _co_degree(a.id)
if deg_a >= MAX_CO_OCCUR_DEGREE:
logger.info(
"CO_OCCURRED degree cap reached for %s (%d edges), skipping new edge to %s",
a.label,
deg_a,
b.label,
)
continue
deg_b = await _co_degree(b.id)
if deg_b >= MAX_CO_OCCUR_DEGREE:
logger.info(
"CO_OCCURRED degree cap reached for %s (%d edges), skipping new edge to %s",
b.label,
deg_b,
a.label,
)
continue
edge = MemoryEdge(
source=a.id,
target=b.id,
relationship="CO_OCCURRED",
tenant_id=scope.tenant_id,
weight=0.3,
)
from symfonic.capabilities.memory.growth import edge_or_defer
result = await edge_or_defer(graph, scope, edge, upsert=True)
if result is None:
return created
created += 1
# Invalidate cached degree so subsequent pairs see the updated count.
co_degree.pop(str(a.id), None)
co_degree.pop(str(b.id), None)
logger.info(
"CO_OCCURRED upsert: %s <-> %s (weight=%.1f)",
a.label,
b.label,
result.weight,
)
except Exception:
logger.debug("Failed CO_OCCURRED %s <-> %s", a.id, b.id, exc_info=True)
return created