symfonic.core.learning.phases_synthetic

phases_synthetic

Moved to :mod:symfonic.capabilities.memory.phases.synthetic.

These phases are on a roster the kernel composes, so they live in the capability now and are imported back here for the legacy consolidator to call. One phase, one implementation, for as long as both routes ship.

The private helpers travel with them: the shipped per-phase tests import them by name, and a shim that re-exported only the public surface would have moved the code away from the tests that watch it.

build_synthetic_links async

build_synthetic_links(graph: Any, episodic_layer: Any, scope: TenantScope, *, min_co_count: int = DEFAULT_MIN_CO_COUNT, init_weight: float = DEFAULT_SYNTHETIC_WEIGHT, max_pairs: int = 500, gate: DurabilityGate | None = None) -> int

Phase 11 -- Scan recent episodic turns for semantic-node co-occurrences, emit candidate edges with provenance='synthetic' for pairs that co-occur

= min_co_count times AND have no existing edge.

Returns the number of synthetic edges created.

Source code in src/symfonic/capabilities/memory/phases/synthetic.py

async def build_synthetic_links(
    graph: Any,
    episodic_layer: Any,
    scope: TenantScope,
    *,
    min_co_count: int = DEFAULT_MIN_CO_COUNT,
    init_weight: float = DEFAULT_SYNTHETIC_WEIGHT,
    max_pairs: int = 500,
    gate: DurabilityGate | None = None,
) -> int:
    """Phase 11 -- Scan recent episodic turns for semantic-node co-occurrences,
    emit candidate edges with provenance='synthetic' for pairs that co-occur
    >= min_co_count times AND have no existing edge.

    Returns the number of synthetic edges created.
    """
    co_map = await _build_cooccurrence_map(
        episodic_layer, scope, DEFAULT_TOP_K_EPISODIC, gate=gate,
    )

    # Resolve labels -> NodeIds in one pass
    try:
        all_nodes = await graph.query_nodes(scope, layer=MemoryLayer.SEMANTIC)
    except Exception:
        logger.debug("Failed to query nodes for synthetic linking", exc_info=True)
        return 0

    label_to_id = {n.label: n.id for n in all_nodes}

    bounded = callable(getattr(type(graph), "neighbor_probe", None))
    existing_edges = []
    if not bounded:
        try:
            existing_edges = await graph.list_edges(scope, limit=10_000)
        except Exception:
            existing_edges = []

    edge_set: set[tuple[str, str]] = set()
    for e in existing_edges:
        edge_set.add((str(e.source), str(e.target)))
        edge_set.add((str(e.target), str(e.source)))  # undirected

    from symfonic.memory.models.edge import MemoryEdge

    created = 0
    for (la, lb), episodic_ids in co_map.items():
        if len(episodic_ids) < min_co_count:
            continue
        sid = label_to_id.get(la)
        tid = label_to_id.get(lb)
        if bounded:
            for label in (la, lb):
                if label not in label_to_id:
                    matches = await graph.query_nodes(scope, label=label)
                    if matches:
                        label_to_id[label] = matches[0].id
            sid, tid = label_to_id.get(la), label_to_id.get(lb)
        if sid is None or tid is None:
            continue
        if (str(sid), str(tid)) in edge_set:
            continue
        if bounded and await graph.neighbor_probe(scope, sid, target=str(tid)):
            continue

        edge = MemoryEdge(
            source=sid,
            target=tid,
            relationship="SYNTHETIC_LINK",
            tenant_id=scope.tenant_id,
            weight=init_weight,
            provenance="synthetic",
            source_episodic_ids=episodic_ids[:20],
            uses=0,
        )
        try:
            from symfonic.capabilities.memory.growth import edge_or_defer
            if await edge_or_defer(graph, scope, edge) is None:
                return created
        except Exception:
            logger.debug("Failed to create synthetic edge %s->%s", la, lb)
            continue

        created += 1
        if created >= max_pairs:
            break

    return created