symfonic.capabilities.memory.phases.entity_helpers

entity_helpers

Internal helpers for phases_entity.build_entity_links.

Split out of phases_entity.py to keep each module under the project's 300-line modularity budget. These helpers are not part of the public API; the phase entrypoint re-exports them only for tests.

confidence_log_get

confidence_log_get(key: tuple[str, str]) -> list[float]

Public accessor for the module-level confidence log.

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

def confidence_log_get(key: tuple[str, str]) -> list[float]:
    """Public accessor for the module-level confidence log."""
    return _confidence_log.get(key, [])

load_existing_entity_nodes async

load_existing_entity_nodes(graph: Any, scope: TenantScope) -> dict[tuple[str, str], MemoryNode]

Load existing Entity:* semantic nodes keyed by (content, kind).

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

async def load_existing_entity_nodes(
    graph: Any,
    scope: TenantScope,
) -> dict[tuple[str, str], MemoryNode]:
    """Load existing ``Entity:*`` semantic nodes keyed by (content, kind)."""
    try:
        all_nodes = await graph.query_nodes(scope, layer=MemoryLayer.SEMANTIC)
    except Exception:
        logger.debug("entity_links: failed to query semantic nodes", exc_info=True)
        return {}
    out: dict[tuple[str, str], MemoryNode] = {}
    for node in all_nodes:
        if not node.label.startswith(ENTITY_LABEL_PREFIX + ":"):
            continue
        props = node.properties or {}
        kind = props.get("kind")
        content = props.get("content")
        if isinstance(kind, str) and isinstance(content, str):
            out[(content, kind)] = node
    return out

resolve_or_create_entity_node async

resolve_or_create_entity_node(graph: Any, scope: TenantScope, canonical_surface: str, kind: EntityKind, *, episodic_ids: list[str], extractor_name: str, max_confidence: float, existing_entity_nodes: dict[tuple[str, str], MemoryNode]) -> tuple[MemoryNode | None, bool]

Return (node, created_new). Implements Decision 2's hybrid identity rule with exact-match -> SequenceMatcher fallback.

existing_entity_nodes is keyed by (canonical_surface, kind) so exact match is O(1). Fuzzy match is only attempted for the proper-noun bucket and only against same-kind nodes (Decision 2).

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

async def resolve_or_create_entity_node(
    graph: Any,
    scope: TenantScope,
    canonical_surface: str,
    kind: EntityKind,
    *,
    episodic_ids: list[str],
    extractor_name: str,
    max_confidence: float,
    existing_entity_nodes: dict[tuple[str, str], MemoryNode],
) -> tuple[MemoryNode | None, bool]:
    """Return ``(node, created_new)``. Implements Decision 2's hybrid
    identity rule with exact-match -> SequenceMatcher fallback.

    ``existing_entity_nodes`` is keyed by ``(canonical_surface, kind)``
    so exact match is O(1). Fuzzy match is only attempted for the
    proper-noun bucket and only against same-kind nodes (Decision 2).
    """
    # A bounded cycle's initial inventory is not the whole scope. Resolve
    # the canonical label directly, then enrich fuzzy candidates separately.
    if callable(getattr(type(graph), "related_candidates", None)):
        existing_entity_nodes = dict(existing_entity_nodes)
        exact_rows = await graph.query_nodes(
            scope, label=f"{ENTITY_LABEL_PREFIX}: {kind}:{canonical_surface}",
        )
        related = await graph.related_candidates(scope, canonical_surface)
        for node in (*exact_rows, *related):
            props = node.properties or {}
            if node.label.startswith(ENTITY_LABEL_PREFIX + ":"):
                surface, found_kind = props.get("content"), props.get("kind")
                if isinstance(surface, str) and isinstance(found_kind, str):
                    existing_entity_nodes.setdefault((surface, found_kind), node)
    # Exact match fast path.
    exact = existing_entity_nodes.get((canonical_surface, kind))
    if exact is not None:
        await _bump_mention_count(
            graph,
            scope,
            exact,
            episodic_ids,
            max_confidence,
        )
        return exact, False

    # Fuzzy fallback for the proper-noun bucket only.
    if kind in PROPER_NOUN_KINDS:
        best_node: MemoryNode | None = None
        best_ratio = 0.0
        for (other_surface, other_kind), other_node in existing_entity_nodes.items():
            if other_kind != kind:
                continue
            ratio = SequenceMatcher(
                None,
                canonical_surface,
                other_surface,
            ).ratio()
            if ratio >= FUZZY_MATCH_THRESHOLD and ratio > best_ratio:
                best_ratio = ratio
                best_node = other_node
        if best_node is not None:
            await _bump_mention_count(
                graph,
                scope,
                best_node,
                episodic_ids,
                max_confidence,
            )
            return best_node, False

    # No match -- mint a new entity node.
    label = f"{ENTITY_LABEL_PREFIX}: {kind}:{canonical_surface}"
    new_node = MemoryNode(
        layer=MemoryLayer.SEMANTIC,
        tenant_id=scope.tenant_id,
        label=label,
        properties={
            "content": canonical_surface,
            "kind": kind,
            "label_prefix": ENTITY_LABEL_PREFIX,
            "first_seen_episodic_id": episodic_ids[0] if episodic_ids else "",
            "mention_count": len(set(episodic_ids)),
            "extractor": extractor_name,
            "confidence": max_confidence,
        },
        importance=5.0,
    )
    try:
        persisted = await graph.add_node(scope, new_node)
    except Exception:
        logger.debug(
            "entity_links: add_node failed for %s",
            label,
            exc_info=True,
        )
        return None, False
    return persisted, True

scan_episodics async

scan_episodics(episodic_layer: Any, scope: TenantScope, extractor: EntityExtractor, *, confidence_threshold: float, max_episodics: int) -> dict[tuple[str, str], list[str]]

Return {(canonical_surface, kind): [episodic_id, ...]}.

The mention-count gate is NOT applied here -- caller decides based on min_mention_count. Drops candidates below confidence_threshold.

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

async def scan_episodics(
    episodic_layer: Any,
    scope: TenantScope,
    extractor: EntityExtractor,
    *,
    confidence_threshold: float,
    max_episodics: int,
) -> dict[tuple[str, str], list[str]]:
    """Return ``{(canonical_surface, kind): [episodic_id, ...]}``.

    The mention-count gate is NOT applied here -- caller decides
    based on ``min_mention_count``. Drops candidates below
    ``confidence_threshold``.
    """
    _confidence_log.clear()
    try:
        entries = await episodic_layer.list_events(scope, limit=max_episodics)
    except Exception:
        logger.debug("entity_links: list_events failed", exc_info=True)
        return {}

    out: dict[tuple[str, str], list[str]] = defaultdict(list)
    seen_episodics: set[str] = set()
    for entry in entries:
        ep_id = str(getattr(entry, "id", getattr(entry, "node_id", "")))
        if not ep_id or ep_id in seen_episodics:
            continue
        seen_episodics.add(ep_id)
        content = getattr(entry, "content", "") or str(
            getattr(entry, "properties", {}).get("content", "")
        )
        if not content:
            continue
        try:
            candidates = await extractor.extract(content)
        except Exception:
            logger.debug(
                "entity_links: extractor raised for episodic %s",
                ep_id,
                exc_info=True,
            )
            continue
        for cand in candidates:
            if cand.confidence < confidence_threshold:
                continue
            kind: EntityKind = cand.kind
            lemmatise = kind not in PROPER_NOUN_KINDS
            canonical = canonicalise_surface(
                cand.surface_form,
                kind,
                lemmatise=lemmatise,
            )
            if not canonical:
                continue
            key = (canonical, kind)
            out[key].append(ep_id)
            _confidence_log.setdefault(key, []).append(cand.confidence)
    return out