symfonic.tools.learning

learning

CommitToMemoryTool -- saves LearnedInsight to the semantic memory layer.

CommitToMemoryTool

CommitToMemoryTool(
    orchestrator: Any = None,
    *,
    importance_cap: float = INSIGHT_IMPORTANCE_CAP,
)

Async tool that persists LearnedInsights to the semantic memory layer.

Designed for fire-and-forget usage inside the self-reflection phase. Errors are logged and suppressed — this tool never raises from commit. Automatically stamps tenant_id and confidence from the insight.

Restated facts collapse into their canonical node: before writing, the commit path fuzzy-matches the insight against existing same-scope semantic nodes (see :func:symfonic.memory.dedup.find_canonical_node) and, on a hit, upserts the canonical row -- latest phrasing wins, importance never decreases -- instead of minting a near-duplicate.

Usage::

tool = CommitToMemoryTool(orchestrator=orchestrator)
await tool.commit(insight)
committed = await tool.commit_batch(insights)

Source code in src/symfonic/tools/learning.py

def __init__(
    self,
    orchestrator: Any = None,
    *,
    importance_cap: float = INSIGHT_IMPORTANCE_CAP,
) -> None:
    self._orchestrator = orchestrator
    self._importance_cap = max(1.0, min(10.0, importance_cap))

commit async

commit(insight: LearnedInsight) -> bool

Persist a single insight to the semantic layer.

Parameters:

Name	Type	Description	Default
insight	LearnedInsight	The LearnedInsight to store.	required

Returns:

Type	Description
bool	True on success, False on any failure (never raises).

Source code in src/symfonic/tools/learning.py

async def commit(self, insight: LearnedInsight) -> bool:
    """Persist a single insight to the semantic layer.

    Args:
        insight: The LearnedInsight to store.

    Returns:
        True on success, False on any failure (never raises).
    """
    if self._orchestrator is None:
        logger.debug("No orchestrator configured, skipping commit")
        return False

    try:
        from symfonic.memory.models.entry import MemoryEntry
        from symfonic.memory.types import MemoryLayer, TenantScope

        # T-7.21.7 (Slice B site #6): prefer the full TenantScope on
        # the insight when set so namespace + sub_tenant_id survive
        # the commit.  Pre-7.21 the bare reconstruction from
        # insight.tenant_id silently dropped both fields.  Fallback
        # path preserves the v7.20 contract for insights that do not
        # carry a scope (legacy adopter extractors).
        if insight.scope is not None:
            scope = insight.scope
        else:
            scope = TenantScope(tenant_id=insight.tenant_id or "default")

        # Build metadata with graph triple if available
        entry_meta: dict[str, Any] = {
            "category": insight.category,
            "tags": list(insight.tags),
            "source": "self_reflection",
            "confidence": insight.confidence,
        }
        # Include graph triple for spreading activation traversal
        if insight.metadata.get("subject"):
            entry_meta["subject"] = insight.metadata["subject"]
        if insight.metadata.get("relation"):
            entry_meta["relationship"] = insight.metadata["relation"]
        if insight.metadata.get("object"):
            entry_meta["object"] = insight.metadata["object"]

        entry = MemoryEntry(
            content=insight.fact,
            layer=MemoryLayer.SEMANTIC,
            tenant_id=scope.tenant_id,
            importance=max(1.0, min(self._importance_cap, insight.confidence * 10)),
            metadata=entry_meta,
        )

        semantic = self._orchestrator.get_layer(MemoryLayer.SEMANTIC)
        if semantic is None:
            logger.debug("Semantic layer not available, skipping commit")
            return False

        # Collapse a restated fact into its canonical node instead of
        # minting a near-duplicate.  Labels derive from the fact text
        # exactly as SemanticLayer.store_fact does (content[:100], the
        # insight metadata carries no explicit "label"), so the match
        # runs against the same string the write would store.  Reusing
        # the canonical node id turns the write into an upsert of that
        # row; importance is floored at the existing node's so a
        # low-confidence restatement never downgrades a learned fact.
        from symfonic.memory.dedup import (
            find_canonical_node,
            normalise_label,
        )

        canonical = await find_canonical_node(
            getattr(semantic, "_graph", None),
            scope,
            layer=MemoryLayer.SEMANTIC,
            label=insight.fact[:100],
        )
        if canonical is not None:
            # Richer-phrasing rule (v8.16): a TERSER restatement of a
            # fact whose stored phrasing already contains it must not
            # overwrite the richer node -- skip the write and keep the
            # canonical as-is.  Otherwise latest phrasing wins.
            new_norm = normalise_label(insight.fact[:100])
            can_norm = normalise_label(canonical.label)
            if new_norm != can_norm and new_norm in can_norm:
                logger.info(
                    "Insight dedup: %r is subsumed by richer node %s; keeping stored phrasing",
                    insight.fact[:60],
                    canonical.id,
                )
                return True
            entry.node_id = canonical.id
            entry.importance = max(entry.importance, float(canonical.importance or 1.0))
            logger.info(
                "Insight dedup: collapsing %r into existing node %s",
                insight.fact[:60],
                canonical.id,
            )

        await semantic.write(scope, entry)
        logger.info(
            "Committed insight: %s (category=%s, confidence=%.2f)",
            insight.fact[:60],
            insight.category,
            insight.confidence,
        )
        return True

    except Exception:
        logger.warning("Failed to commit insight", exc_info=True)
        return False

commit_batch async

commit_batch(insights: list[LearnedInsight]) -> int

Commit multiple insights sequentially.

Parameters:

Name	Type	Description	Default
insights	list[LearnedInsight]	List of LearnedInsight objects to persist.	required

Returns:

Type	Description
int	Count of successfully committed insights.

Source code in src/symfonic/tools/learning.py

async def commit_batch(self, insights: list[LearnedInsight]) -> int:
    """Commit multiple insights sequentially.

    Args:
        insights: List of LearnedInsight objects to persist.

    Returns:
        Count of successfully committed insights.
    """
    committed = 0
    for insight in insights:
        if await self.commit(insight):
            committed += 1
    return committed