symfonic.capabilities.memory.linking

linking

Entity linking: the same surface, mentioned repeatedly, becomes a node.

A memory that says "we met at Cafe Milano" and another that says "Cafe Milano again" are two episodes about one thing. Linking mints that thing as a semantic memory and records that the two co-mentioned surfaces belong together, so a later retrieval can reach one through the other.

Three refusals shape the output, and each is a false-positive tax the shipped phase already pays:

• A single mention is not an entity. min_mention_count exists because every capitalised word in one sentence would otherwise become a permanent fact about the tenant.
• The heuristic extractor is opt-in by threshold. It reports every capitalised token at 0.4, under the 0.5 default threshold, so wiring it changes nothing until a deployment lowers the bar deliberately. That is the shipped behaviour (design §16.0 Q2) and it is preserved on purpose: a regex that promotes proper nouns by default fills a graph with sentence starts.
• Reading is budgeted. max_records bounds how much history one pass scans, because linking runs inside a consolidation cycle that has to end.

Surfaces are slugged into ids rather than used as ids. A record id lives in a rendered charset; "Cafe Milano/Berlin" would forge a delimiter, and an accented surface would not be expressible at all.

EntityLink dataclass

EntityLink(subject: str, object: str, relationship: str = MENTIONS, weight: float = 0.5)

Two entities mentioned together in the same memory.

EntityMention dataclass

EntityMention(surface: str, kind: str = 'other', confidence: float = HEURISTIC_CONFIDENCE)

One surface form found in one memory.

HeuristicEntityExtractor

Capitalised-token extraction with a stoplist. The zero-dependency default.

extract

extract(text: str) -> tuple[EntityMention, ...]

Every capitalised surface in text that survives the stoplist.

Source code in src/symfonic/capabilities/memory/linking.py

def extract(self, text: str) -> tuple[EntityMention, ...]:
    """Every capitalised surface in ``text`` that survives the stoplist."""
    mentions: list[EntityMention] = []
    seen: set[str] = set()
    for match in self._TOKEN.finditer(text):
        surface = match.group(0).strip()
        surface = _strip_stopwords(surface, self.STOPLIST)
        if not surface or surface in self.STOPLIST or surface in seen:
            continue
        seen.add(surface)
        mentions.append(EntityMention(surface=surface))
    return tuple(mentions)

LinkingResult dataclass

LinkingResult(entities: tuple[MemoryRecord, ...] = (), links: tuple[EntityLink, ...] = (), dropped: tuple[tuple[str, str], ...] = ())

The entities one pass minted, the links between them, and the refusals.

link_entities

link_entities(records: Sequence[MemoryRecord], *, scope: MemoryScope, extractor: HeuristicEntityExtractor | None = None, min_mention_count: int = 2, confidence_threshold: float = 0.5, max_records: int = 200) -> LinkingResult

Mint entities for surfaces mentioned often enough, and link co-mentions.

Source code in src/symfonic/capabilities/memory/linking.py

def link_entities(
    records: Sequence[MemoryRecord],
    *,
    scope: MemoryScope,
    extractor: HeuristicEntityExtractor | None = None,
    min_mention_count: int = 2,
    confidence_threshold: float = 0.5,
    max_records: int = 200,
) -> LinkingResult:
    """Mint entities for surfaces mentioned often enough, and link co-mentions."""
    if len(records) > max_records:
        return LinkingResult(
            dropped=(
                (
                    "",
                    f"{len(records)} memories exceed the {max_records}-memory read "
                    "budget for one linking pass",
                ),
            )
        )
    reader = extractor if extractor is not None else HeuristicEntityExtractor()
    counts: dict[str, int] = {}
    per_record: list[tuple[str, ...]] = []
    dropped: list[tuple[str, str]] = []

    for record in records:
        kept: list[str] = []
        for mention in reader.extract(record.text):
            if mention.confidence < confidence_threshold:
                dropped.append(
                    (
                        mention.surface,
                        f"confidence {mention.confidence} is below the "
                        f"{confidence_threshold} threshold",
                    )
                )
                continue
            counts[mention.surface] = counts.get(mention.surface, 0) + 1
            kept.append(mention.surface)
        per_record.append(tuple(kept))

    minted = {
        surface for surface, count in counts.items() if count >= min_mention_count
    }
    for surface, count in sorted(counts.items()):
        if surface not in minted:
            dropped.append(
                (surface, f"mentioned {count} time(s), below the {min_mention_count} floor")
            )

    entities = tuple(
        _entity_record(surface, scope, counts[surface]) for surface in sorted(minted)
    )
    return LinkingResult(
        entities=entities,
        links=_links(per_record, minted),
        dropped=tuple(dropped),
    )