`symfonic.core.learning`¶

learning ¶

symfonic.core.learning -- self-reflection, insight extraction, and consolidation.

ConsolidationReport `dataclass` ¶

ConsolidationReport(tenant_id: str, started_at: datetime = (lambda: datetime.now(UTC))(), finished_at: datetime | None = None, nodes_strengthened: int = 0, nodes_pruned: int = 0, risk_nodes_tagged: int = 0, meta_nodes_created: int = 0, soul_updates: int = 0, profile_updates: int = 0, edges_created: int = 0, working_ttl_pruned: int = 0, nodes_decayed: int = 0, nodes_expired: int = 0, retracted_nodes_pruned: int = 0, episodes_summarized: int = 0, synthetic_edges_created: int = 0, draft_skills_promoted: int = 0, entity_nodes_created: int = 0, mentions_edges_created: int = 0, transient_rejected: int = 0, semantic_merges: int = 0, errors: list[str] = list(), phases_run: list[str] = list())

Summary of a single consolidation run.

phases_run is an observability-only list populated by each phase implementation in phases.py / phases_maintenance.py / phases_episodic.py etc. when the phase actually executes (even if its mutation count is zero). It exists to disambiguate "the phase ran and found nothing" from "the phase was short-circuited and never ran" -- both produce a 0 counter, but only the former appears in phases_run. Purely additive; existing consumers reading the counter fields continue to work unchanged.

InsightExtractor ¶

InsightExtractor(llm: Any = None, custom_prompt: str | None = None, min_confidence: float = 0.5, *, observability_hook: ObservabilityHook | None = None)

Extracts LearnedInsight objects from conversation text using an LLM.

The extraction prompt can be customized with domain-specific instructions via custom_prompt. The default prompt targets general business insights.

Graceful degradation: when no LLM is configured, extract() returns an empty list without raising.

Source code in symfonic/core/learning/refiner.py

def __init__(
    self,
    llm: Any = None,
    custom_prompt: str | None = None,
    min_confidence: float = 0.5,
    *,
    observability_hook: ObservabilityHook | None = None,
) -> None:
    self._llm = llm
    self._prompt_template = custom_prompt or _DEFAULT_PROMPT
    self._min_confidence = min_confidence
    # v8.3.2 observability-bug fix (G2): the insight-extractor LLM call
    # emitted ONLY on the merge-dependent callback manager and made ZERO
    # ``ObservabilityHook`` calls -- the same bug class as the v8.3.1
    # critic.  The engine now threads in the always-on hook
    # (``deps.require(ObservabilityHook)``); ``None`` resolves to a
    # ``NoOpObservabilityHook`` so the default is byte-identical
    # (additive only).  Attribution is MODEL-BASED (no node_name on the
    # hook signature); node-level disambiguation is deferred to Build 2.
    if observability_hook is None:
        from symfonic.core.observability.hooks import NoOpObservabilityHook

        observability_hook = NoOpObservabilityHook()
    self._observability_hook = observability_hook

extract `async` ¶

extract(conversation: str, tenant_id: str = '', *, callback_manager: CallbackManager | None = None, run_id: str = '', model_name: str = '') -> list[LearnedInsight]

Extract insights from conversation text.

Parameters:

Name	Type	Description	Default
`conversation`	`str`	The raw conversation text to analyze.	required
`tenant_id`	`str`	Tenant identifier to stamp on each insight.	`''`
`callback_manager`	`CallbackManager \| None`	v7.4.3 -- when present, fires `on_llm_end(node_name="insight_extractor")` after the LLM call (mirrors react.py:303-311). `None` preserves pre-7.4.3 behaviour.	`None`
`run_id`	`str`	Engine run identifier for callback correlation.	`''`
`model_name`	`str`	Optional model identifier to stamp on the LLMEndEvent. When empty, falls back to `"insight_extractor"`.	`''`

Returns:

Type	Description
`list[LearnedInsight]`	List of LearnedInsight objects above the min_confidence threshold.
`list[LearnedInsight]`	Returns empty list when no LLM is configured or the conversation
`list[LearnedInsight]`	is blank (never raises).

Source code in symfonic/core/learning/refiner.py

async def extract(
    self,
    conversation: str,
    tenant_id: str = "",
    *,
    callback_manager: CallbackManager | None = None,
    run_id: str = "",
    model_name: str = "",
) -> list[LearnedInsight]:
    """Extract insights from conversation text.

    Args:
        conversation: The raw conversation text to analyze.
        tenant_id: Tenant identifier to stamp on each insight.
        callback_manager: v7.4.3 -- when present, fires
            ``on_llm_end(node_name="insight_extractor")`` after the
            LLM call (mirrors react.py:303-311).  ``None`` preserves
            pre-7.4.3 behaviour.
        run_id: Engine run identifier for callback correlation.
        model_name: Optional model identifier to stamp on the
            LLMEndEvent.  When empty, falls back to ``"insight_extractor"``.

    Returns:
        List of LearnedInsight objects above the min_confidence threshold.
        Returns empty list when no LLM is configured or the conversation
        is blank (never raises).
    """
    if not self._llm or not conversation.strip():
        return []

    prompt = self._prompt_template.format(conversation=conversation)

    try:
        raw_text = await self._invoke_llm(
            prompt,
            callback_manager=callback_manager,
            run_id=run_id,
            model_name=model_name or "insight_extractor",
        )
        return self._parse_insights(raw_text, tenant_id)
    except Exception:
        logger.warning("Insight extraction failed", exc_info=True)
        return []

LearnedInsight `dataclass` ¶

LearnedInsight(fact: str, confidence: float, category: str, tags: tuple[str, ...] = (), tenant_id: str = '', timestamp: datetime = (lambda: datetime.now(UTC))(), metadata: dict[str, Any] = dict(), scope: TenantScope | None = None)

A business fact extracted from conversation by the InsightExtractor.

Immutable by design — once extracted, an insight is a fixed record that can be safely passed across async boundaries without copying.

Attributes:

Name	Type	Description
`fact`	`str`	The specific business observation (1-2 sentences).
`confidence`	`float`	Certainty score between 0.0 and 1.0.
`category`	`str`	Domain bucket (e.g. "customer_behavior", "inventory").
`tags`	`tuple[str, ...]`	Optional keyword tuple for filtering and retrieval.
`tenant_id`	`str`	Owning tenant; empty string means unscoped.
`timestamp`	`datetime`	UTC creation time; auto-set on construction.
`metadata`	`dict[str, Any]`	Arbitrary extra data from the extraction process.
`scope`	`TenantScope \| None`	T-7.21.7 -- optional full `TenantScope` carrying `sub_tenant_id` / `namespace` (and any future field). Preferred by :class:`CommitToMemoryTool` over the bare `tenant_id` reconstruction path because the latter silently dropped `namespace`. Backward compatible: when `None` (the v7.20 default), the tool falls back to the bare- tenant_id construction.

LearningPolicy `dataclass` ¶

LearningPolicy(lookback_hours: int = 24, promotion_min_pattern_count: int = 3, promotion_recency_days: int = 30, promotion_max_drafts_per_run: int = 5, promotion_use_tool_calls_fallback: bool = False, promotion_promote_assistant_content: bool = False, phase_12_use_llm_extractor: bool = False, phase_12_llm_model: str = 'claude-haiku-4-5', phase_12_llm_max_episodes_per_run: int = 100, phase_12_llm_max_drafts_per_run: int = 5, episodic_summarization_max_entries: int = 100, episodic_summarization_batch_size: int = 50, phase1_spreading_weight: float = 0.5, synthetic_link_min_co_count: int = 2, enable_entity_linker: bool = False, entity_linker_extractor_kind: str = 'regex', entity_linker_min_mention_count: int = 2, entity_linker_max_episodics_per_run: int = 200, entity_linker_confidence_threshold: float = 0.5)

What a consolidation run should promote, summarise and link.

Frozen: a run reads its policy once, and one that could change mid-pass would produce a result nobody can reproduce.

as_kwargs ¶

as_kwargs() -> dict[str, Any]

The keyword arguments a consolidation run takes.

One call site instead of nineteen. Every getattr(config, name, default) it replaces was a place the framework's default and the template's copy could drift apart with nothing to notice.

Source code in symfonic/capabilities/memory/policy.py

def as_kwargs(self) -> dict[str, Any]:
    """The keyword arguments a consolidation run takes.

    One call site instead of nineteen. Every
    ``getattr(config, name, default)`` it replaces was a place the
    framework's default and the template's copy could drift apart with
    nothing to notice.
    """
    return asdict(self)

as_phase_kwargs ¶

as_phase_kwargs() -> dict[str, Any]

The same knobs, named as the phase factories name them.

A translation table rather than a rename, because the two vocabularies are genuinely different: the shipped consolidator takes nineteen keyword arguments on one constructor, and the factories take them where the phase that reads each one is built. Written once, here, next to the fields it maps -- a worker doing this inline would be a second copy of every default in the place this class exists to remove them from.

enable_entity_linker and phase_12_use_llm_extractor are absent on purpose: in the factories a phase runs when it was given the collaborator it needs, so "enabled" is not a flag but the presence of an extractor. A deployment that set the flag and composed nothing would otherwise have a phase that reports zero rather than declining.

Source code in symfonic/capabilities/memory/policy.py

def as_phase_kwargs(self) -> dict[str, Any]:
    """The same knobs, named as the phase factories name them.

    A translation table rather than a rename, because the two vocabularies
    are genuinely different: the shipped consolidator takes nineteen
    keyword arguments on one constructor, and the factories take them where
    the phase that reads each one is built. Written once, here, next to the
    fields it maps -- a worker doing this inline would be a second copy of
    every default in the place this class exists to remove them from.

    ``enable_entity_linker`` and ``phase_12_use_llm_extractor`` are absent
    on purpose: in the factories a phase runs when it was given the
    collaborator it needs, so "enabled" is not a flag but the presence of
    an extractor. A deployment that set the flag and composed nothing would
    otherwise have a phase that reports zero rather than declining.
    """
    return {
        "lookback_hours": float(self.lookback_hours),
        "spreading_weight": self.phase1_spreading_weight,
        "episodic_max_entries": self.episodic_summarization_max_entries,
        "episodic_summarize_batch": self.episodic_summarization_batch_size,
        "entity_min_mention_count": self.entity_linker_min_mention_count,
        "entity_max_episodics_per_run": self.entity_linker_max_episodics_per_run,
        "entity_confidence_threshold": self.entity_linker_confidence_threshold,
        "procedural_model_name": self.phase_12_llm_model,
    }

from_settings `classmethod` ¶

from_settings(settings: Any) -> LearningPolicy

Read a deployment's own settings object, falling back per field.

Absent names take the shipped default rather than raising: a settings object carries what that deployment chose to configure and nothing else, and requiring all nineteen would put a second copy of every default back in the place this removes it from.

Source code in symfonic/capabilities/memory/policy.py

@classmethod
def from_settings(cls, settings: Any) -> LearningPolicy:
    """Read a deployment's own settings object, falling back per field.

    Absent names take the shipped default rather than raising: a settings
    object carries what that deployment chose to configure and nothing
    else, and requiring all nineteen would put a second copy of every
    default back in the place this removes it from.
    """
    supplied = {
        field.name: getattr(settings, field.name)
        for field in fields(cls)
        if hasattr(settings, field.name)
        # Pydantic settings commonly use ``None`` to mean "not
        # configured".  Passing it through replaces this value object's
        # typed default and fails later in validation (or arithmetic),
        # which made a stock generated worker unable to start DEEP.
        and getattr(settings, field.name) is not None
    }
    # The scaffold's environment spelling predates this value object's
    # explicit ``_kind`` suffix.  It is the same choice, not a second
    # setting, so translate it at the boundary.
    if "entity_linker_extractor_kind" not in supplied:
        alias = getattr(settings, "entity_linker_extractor", None)
        if alias is not None:
            supplied["entity_linker_extractor_kind"] = alias
    return cls(**supplied)

SleepConsolidator ¶

SleepConsolidator(graph_store: GraphMemoryStore, *, lookback_hours: int = 24, llm_summariser: Any | None = None, episodic_layer: Any | None = None, procedural_layer: Any | None = None, promotion_min_pattern_count: int = 3, promotion_recency_days: int = 30, promotion_max_drafts_per_run: int = 5, promotion_use_tool_calls_fallback: bool = False, promotion_promote_assistant_content: bool = False, phase_12_use_llm_extractor: bool = False, phase_12_llm_max_episodes_per_run: int = 100, phase_12_llm_max_drafts_per_run: int = 5, phase_12_llm_model: str = 'claude-haiku-4-5', chat_model: Any | None = None, episodic_summarization_max_entries: int = 100, episodic_summarization_batch_size: int = 50, phase1_spreading_weight: float = 0.5, synthetic_link_min_co_count: int = 2, enable_entity_linker: bool = False, entity_linker_extractor_kind: Literal['regex', 'spacy', 'llm'] = 'regex', entity_linker_min_mention_count: int = 2, entity_linker_max_episodics_per_run: int = 200, entity_linker_confidence_threshold: float = 0.5, entity_linker_extractor: Any | None = None, enable_semantic_merge: bool = False, semantic_merge_embedding_provider: Any | None = None, semantic_merge_embedding_threshold: float = 0.9, semantic_merge_lexical_threshold: float = 0.6, semantic_merge_max_pairs_per_run: int = 10)

Offline memory consolidation engine.

Phases: 1. Strengthen -- boost importance for recently recurring nodes. 1.5 Co-occurrence edges -- create CO_OCCURRED edges between nodes updated within the same 1-hour window that have no direct edge. 2. Tag risk -- mark nodes with negative-sentiment context. 2.5 Pending edges -- write inferred pending_connections as real edges. 3. Prune orphans -- delete stale low-confidence orphan nodes. 4. Generate meta-nodes -- cluster nodes by label prefix. 5. Profile-fact promotion -- promote user-corrected profile facts (name/role/etc, scoped by profile_fields) onto the tenant's SOUL node in the graph. See phases_profile.py docstring. 8. Working TTL cleanup -- prune WORKING-layer nodes with expired TTL. 9. Semantic importance decay -- reduce importance of stale nodes. 10. Episodic summarization -- compress old episodic entries into a semantic meta-node.

Source code in symfonic/core/learning/consolidation.py

def __init__(
    self,
    graph_store: GraphMemoryStore,
    *,
    lookback_hours: int = 24,
    llm_summariser: Any | None = None,
    episodic_layer: Any | None = None,
    procedural_layer: Any | None = None,
    promotion_min_pattern_count: int = 3,
    promotion_recency_days: int = 30,
    promotion_max_drafts_per_run: int = 5,
    # v7.4.7 adopter Ask 9 PR-A: opt-in tool-call fallback for the
    # regex-based pattern extractor. Off by default so existing
    # consumers see no behaviour change.
    promotion_use_tool_calls_fallback: bool = False,
    # v7.7.2 adopter fabrication-promotion ask: default-off
    # promotion of assistant-narrated content.  When ``False``
    # (default), the regex AND LLM extractor paths filter out
    # episodic rows whose ``metadata['speaker'] == 'assistant'``
    # and legacy rows without a speaker discriminator.
    promotion_promote_assistant_content: bool = False,
    # v7.6 adopter Ask 9 Option A: opt-in LLM-backed procedural
    # extractor.  ``phase_12_use_llm_extractor=False`` (default) runs
    # the regex extractor unchanged.  When True AND ``chat_model`` is
    # non-None, ``_build_procedural_extractor`` instantiates
    # :class:`LlmProceduralExtractor` and the regex path is skipped
    # (either-or dispatch -- the eval rejected merged streams).
    phase_12_use_llm_extractor: bool = False,
    phase_12_llm_max_episodes_per_run: int = 100,
    phase_12_llm_max_drafts_per_run: int = 5,
    phase_12_llm_model: str = "claude-haiku-4-5",
    chat_model: Any | None = None,
    episodic_summarization_max_entries: int = 100,
    episodic_summarization_batch_size: int = 50,
    phase1_spreading_weight: float = 0.5,
    synthetic_link_min_co_count: int = 2,
    # v7.2 Roadmap Item 12 -- EntityLinker phase (default-off).
    enable_entity_linker: bool = False,
    entity_linker_extractor_kind: Literal["regex", "spacy", "llm"] = "regex",
    entity_linker_min_mention_count: int = 2,
    entity_linker_max_episodics_per_run: int = 200,
    entity_linker_confidence_threshold: float = 0.5,
    entity_linker_extractor: Any | None = None,
    # v8.16 Phase 13 -- LLM-judged semantic-duplicate merge
    # (default-off).  Catches the restatement classes the write-time
    # dedup documents as misses (front reword, clause reorder; see
    # tests/memory/test_restatement_corpus.py).  Requires
    # ``chat_model``; ``semantic_merge_embedding_provider`` upgrades
    # the candidate prefilter from lexical ratio to cosine
    # similarity.  The judge is the correctness gate either way.
    enable_semantic_merge: bool = False,
    semantic_merge_embedding_provider: Any | None = None,
    semantic_merge_embedding_threshold: float = 0.90,
    semantic_merge_lexical_threshold: float = 0.60,
    semantic_merge_max_pairs_per_run: int = 10,
) -> None:
    self._graph = graph_store
    self._lookback_hours = lookback_hours
    self._llm_summarise = llm_summariser
    self._episodic = episodic_layer
    self._procedural = procedural_layer
    self._promotion_min_pattern_count = promotion_min_pattern_count
    self._promotion_recency_days = promotion_recency_days
    self._promotion_max_drafts_per_run = promotion_max_drafts_per_run
    self._promotion_use_tool_calls_fallback = (
        promotion_use_tool_calls_fallback
    )
    self._promotion_promote_assistant_content = (
        promotion_promote_assistant_content
    )
    # v7.6 LLM extractor knobs -- stored for the factory in
    # ``_build_procedural_extractor`` to consume.
    self._phase_12_use_llm_extractor = phase_12_use_llm_extractor
    self._phase_12_llm_max_episodes_per_run = (
        phase_12_llm_max_episodes_per_run
    )
    self._phase_12_llm_max_drafts_per_run = (
        phase_12_llm_max_drafts_per_run
    )
    self._phase_12_llm_model = phase_12_llm_model
    self._chat_model = chat_model
    self._episodic_summarization_max_entries = episodic_summarization_max_entries
    self._episodic_summarization_batch_size = episodic_summarization_batch_size
    # v6.2 T02: weight applied to spreading_access_count inside
    # Phase 1 strengthen. 0.5 preserves v6.1.x observable behaviour.
    self._phase1_spreading_weight = phase1_spreading_weight
    # v6.2 T03: minimum co-occurrence count for Phase 11 synthetic
    # edges. 2 matches the hardcoded v6.1 floor.
    self._synthetic_link_min_co_count = synthetic_link_min_co_count
    # v7.2 Roadmap Item 12 -- EntityLinker phase configuration.
    # When ``enable_entity_linker`` is False the phase is a strict
    # no-op (not appended to ``report.phases_run``).
    self._enable_entity_linker = enable_entity_linker
    self._entity_linker_extractor_kind = entity_linker_extractor_kind
    self._entity_linker_min_mention_count = entity_linker_min_mention_count
    self._entity_linker_max_episodics_per_run = (
        entity_linker_max_episodics_per_run
    )
    self._entity_linker_confidence_threshold = (
        entity_linker_confidence_threshold
    )
    # Test escape hatch: pre-built extractor instance trumps the
    # ``_kind`` knob so tests can drop in mocks without spinning up
    # spaCy or the LLM provider.
    self._entity_linker_extractor = entity_linker_extractor
    # v8.16 Phase 13 -- semantic-merge configuration.  When
    # ``enable_semantic_merge`` is False the phase is a strict no-op
    # (not appended to ``report.phases_run``).
    self._enable_semantic_merge = enable_semantic_merge
    self._semantic_merge_embedding_provider = (
        semantic_merge_embedding_provider
    )
    self._semantic_merge_embedding_threshold = (
        semantic_merge_embedding_threshold
    )
    self._semantic_merge_lexical_threshold = (
        semantic_merge_lexical_threshold
    )
    self._semantic_merge_max_pairs_per_run = (
        semantic_merge_max_pairs_per_run
    )

bind_layers ¶

bind_layers(*, episodic_layer: Any | None = None, procedural_layer: Any | None = None, overwrite: bool = False) -> None

Late-bind the episodic and/or procedural layers after construction.

The constructor accepts episodic_layer and procedural_layer as kwargs (see :meth:__init__). Callers that build the :class:SymfonicAgent first and only obtain layer references via agent._orchestrator.get_layer(...) afterwards can use this method to wire the layers without resorting to direct attribute assignment.

Direct consolidator._episodic = ... mutation silently creates new attributes on a typo (_episodic_layer vs _episodic) because the class is not slotted. This method validates the kwarg names and the binding state, so a typo raises TypeError at the call site (Python's keyword-only enforcement) instead of silently no-op'ing.

Parameters:

Name	Type	Description	Default
`episodic_layer`	`Any \| None`	Optional EpisodicLayer to bind. When None, leaves the existing binding unchanged.	`None`
`procedural_layer`	`Any \| None`	Optional ProceduralLayer to bind. When None, leaves the existing binding unchanged.	`None`
`overwrite`	`bool`	When False (default), refuses to overwrite an already-bound layer (raises `RuntimeError`). Set True to replace an existing binding (e.g. test rebinding).	`False`

Raises:

Type	Description
`TypeError`	When an unknown kwarg is passed (typo-safety). The keyword-only parameters above mean Python raises `TypeError` naturally at the call site for unknown kwargs.
`RuntimeError`	When `overwrite=False` (default) and an attempt is made to rebind a layer that's already set.

Example::

agent = SymfonicAgent(model_provider=provider, config=config)
consolidator = SleepConsolidator(graph_store=store)
consolidator.bind_layers(
    episodic_layer=agent._orchestrator.get_layer(
        MemoryLayer.EPISODIC,
    ),
    procedural_layer=agent._orchestrator.get_layer(
        MemoryLayer.PROCEDURAL,
    ),
)

Source code in symfonic/core/learning/consolidation.py

def bind_layers(
    self,
    *,
    episodic_layer: Any | None = None,
    procedural_layer: Any | None = None,
    overwrite: bool = False,
) -> None:
    """Late-bind the episodic and/or procedural layers after construction.

    The constructor accepts ``episodic_layer`` and ``procedural_layer``
    as kwargs (see :meth:`__init__`).  Callers that build the
    :class:`SymfonicAgent` first and only obtain layer references via
    ``agent._orchestrator.get_layer(...)`` afterwards can use this
    method to wire the layers without resorting to direct attribute
    assignment.

    Direct ``consolidator._episodic = ...`` mutation silently creates
    new attributes on a typo (``_episodic_layer`` vs ``_episodic``)
    because the class is not slotted.  This method validates the
    kwarg names and the binding state, so a typo raises ``TypeError``
    at the call site (Python's keyword-only enforcement) instead of
    silently no-op'ing.

    Args:
        episodic_layer: Optional EpisodicLayer to bind.  When None,
            leaves the existing binding unchanged.
        procedural_layer: Optional ProceduralLayer to bind.  When
            None, leaves the existing binding unchanged.
        overwrite: When False (default), refuses to overwrite an
            already-bound layer (raises ``RuntimeError``).  Set
            True to replace an existing binding (e.g. test
            rebinding).

    Raises:
        TypeError: When an unknown kwarg is passed (typo-safety).
            The keyword-only parameters above mean Python raises
            ``TypeError`` naturally at the call site for unknown
            kwargs.
        RuntimeError: When ``overwrite=False`` (default) and an
            attempt is made to rebind a layer that's already set.

    Example::

        agent = SymfonicAgent(model_provider=provider, config=config)
        consolidator = SleepConsolidator(graph_store=store)
        consolidator.bind_layers(
            episodic_layer=agent._orchestrator.get_layer(
                MemoryLayer.EPISODIC,
            ),
            procedural_layer=agent._orchestrator.get_layer(
                MemoryLayer.PROCEDURAL,
            ),
        )
    """
    if episodic_layer is not None:
        if self._episodic is not None and not overwrite:
            raise RuntimeError(
                "SleepConsolidator.episodic_layer is already bound. "
                "Pass overwrite=True to replace it.",
            )
        self._episodic = episodic_layer
    if procedural_layer is not None:
        if self._procedural is not None and not overwrite:
            raise RuntimeError(
                "SleepConsolidator.procedural_layer is already bound. "
                "Pass overwrite=True to replace it.",
            )
        self._procedural = procedural_layer

nightly_nap `async` ¶

nightly_nap(scope: TenantScope, *, soul_schema: dict[str, Any] | None = None, profile_fields: frozenset[str] | None = None, pending_connections: list[dict[str, Any]] | None = None, max_entries: int | None = None, summarize_batch: int | None = None) -> ConsolidationReport

Run the full 11-phase consolidation roster (v7.0 T12).

quick_nap runs the 3-phase subset on every-N-turns cadence; nightly_nap runs the full roster (Phase 1, 1.5, 2, 2.5, 3, 4, 5, 8, 9, 10, 11, 12) and is intended for a cron / celery- beat / APScheduler job firing once per quiet window per tenant.

Accepts per-call overrides for the Phase 10 knobs so the scheduler can tune each tenant independently without rebuilding the consolidator.

soul_schema (deprecated) and profile_fields are forwarded unchanged to :meth:run -- see its docstring for the resolution rule and deprecation shape.

Returns:

Type	Description
`ConsolidationReport`	ConsolidationReport with every phase that actually fired
`ConsolidationReport`	populated in `phases_run`.

Source code in symfonic/core/learning/consolidation.py

async def nightly_nap(
    self,
    scope: TenantScope,
    *,
    soul_schema: dict[str, Any] | None = None,
    profile_fields: frozenset[str] | None = None,
    pending_connections: list[dict[str, Any]] | None = None,
    max_entries: int | None = None,
    summarize_batch: int | None = None,
) -> ConsolidationReport:
    """Run the full 11-phase consolidation roster (v7.0 T12).

    ``quick_nap`` runs the 3-phase subset on every-N-turns cadence;
    ``nightly_nap`` runs the full roster (Phase 1, 1.5, 2, 2.5, 3,
    4, 5, 8, 9, 10, 11, 12) and is intended for a cron / celery-
    beat / APScheduler job firing once per quiet window per
    tenant.

    Accepts per-call overrides for the Phase 10 knobs so the
    scheduler can tune each tenant independently without rebuilding
    the consolidator.

    ``soul_schema`` (deprecated) and ``profile_fields`` are forwarded
    unchanged to :meth:`run` -- see its docstring for the resolution
    rule and deprecation shape.

    Returns:
        ConsolidationReport with every phase that actually fired
        populated in ``phases_run``.
    """
    # Apply per-call overrides by temporarily swapping instance
    # defaults so the existing ``run`` path picks them up without
    # duplicating phase orchestration here.
    original_max = self._episodic_summarization_max_entries
    original_batch = self._episodic_summarization_batch_size
    try:
        if max_entries is not None:
            self._episodic_summarization_max_entries = max_entries
        if summarize_batch is not None:
            self._episodic_summarization_batch_size = summarize_batch
        logger.info(
            "nightly_nap starting: tenant=%s",
            scope.tenant_id,
        )
        report = await self.run(
            scope,
            soul_schema=soul_schema,
            profile_fields=profile_fields,
            pending_connections=pending_connections,
        )
        logger.info(
            "nightly_nap complete: tenant=%s phases_run=%d errors=%d",
            scope.tenant_id,
            len(report.phases_run),
            len(report.errors),
        )
        return report
    finally:
        self._episodic_summarization_max_entries = original_max
        self._episodic_summarization_batch_size = original_batch

quick_nap `async` ¶

quick_nap(scope: TenantScope, *, soul_schema: dict[str, Any] | None = None, profile_fields: frozenset[str] | None = None, max_entries: int | None = None, summarize_batch: int | None = None) -> ConsolidationReport

Lightweight consolidation: Phase 1 (strengthen) + Phase 10 (episodic summary).

Runs in <1s vs full run(). Intended for every-N-turns auto-consolidation. Only operates on recently updated semantic nodes within the lookback window.

Parameters:

Name	Type	Description	Default
`scope`	`TenantScope`	Tenant scope for multi-tenant isolation.	required
`soul_schema`	`dict[str, Any] \| None`	Deprecated. Pass `profile_fields` instead. When supplied without `profile_fields`, derives `frozenset(soul_schema.keys())` and emits a `DeprecationWarning`. Read-only here -- never mutated.	`None`
`profile_fields`	`frozenset[str] \| None`	Field names that constitute the tenant's profile. When provided, Phase 5 runs so user corrections recorded in recent nodes are promoted onto the tenant's SOUL node. Auto-forwarded by `SymfonicAgent` from `frozenset(FrameworkConfig.domain.soul_schema)` so callers get Phase 5 without touching the auto- consolidation plumbing.	`None`
`max_entries`	`int \| None`	Optional Phase 10 threshold override (per-call). When `None`, the instance default set at construction is used. Forwarded by the engine from `FrameworkConfig.episodic_summarization_max_entries`.	`None`
`summarize_batch`	`int \| None`	Optional Phase 10 batch size override (per-call). When `None`, the instance default set at construction is used. Forwarded by the engine from `FrameworkConfig.episodic_summarization_batch_size`.	`None`

Returns:

Type	Description
`ConsolidationReport`	ConsolidationReport with nodes_strengthened, soul_updates,
`ConsolidationReport`	profile_updates, and episodes_summarized populated.

Source code in symfonic/core/learning/consolidation.py

async def quick_nap(
    self,
    scope: TenantScope,
    *,
    soul_schema: dict[str, Any] | None = None,
    profile_fields: frozenset[str] | None = None,
    max_entries: int | None = None,
    summarize_batch: int | None = None,
) -> ConsolidationReport:
    """Lightweight consolidation: Phase 1 (strengthen) + Phase 10 (episodic summary).

    Runs in <1s vs full run(). Intended for every-N-turns auto-consolidation.
    Only operates on recently updated semantic nodes within the lookback window.

    Args:
        scope: Tenant scope for multi-tenant isolation.
        soul_schema: Deprecated.  Pass ``profile_fields`` instead.  When
            supplied without ``profile_fields``, derives
            ``frozenset(soul_schema.keys())`` and emits a
            ``DeprecationWarning``.  Read-only here -- never mutated.
        profile_fields: Field names that constitute the tenant's
            profile.  When provided, Phase 5 runs so user corrections
            recorded in recent nodes are promoted onto the tenant's
            SOUL node.  Auto-forwarded by ``SymfonicAgent`` from
            ``frozenset(FrameworkConfig.domain.soul_schema)`` so
            callers get Phase 5 without touching the auto-
            consolidation plumbing.
        max_entries: Optional Phase 10 threshold override (per-call). When
            ``None``, the instance default set at construction is used.
            Forwarded by the engine from
            ``FrameworkConfig.episodic_summarization_max_entries``.
        summarize_batch: Optional Phase 10 batch size override (per-call).
            When ``None``, the instance default set at construction is
            used. Forwarded by the engine from
            ``FrameworkConfig.episodic_summarization_batch_size``.

    Returns:
        ConsolidationReport with nodes_strengthened, soul_updates,
        profile_updates, and episodes_summarized populated.
    """
    report = ConsolidationReport(tenant_id=scope.tenant_id)
    logger.debug("quick_nap starting: tenant=%s", scope.tenant_id)

    try:
        all_nodes = await self._graph.query_nodes(scope, layer=MemoryLayer.SEMANTIC)
    except Exception as exc:
        report.errors.append(f"quick_nap: failed to query nodes: {exc}")
        report.finished_at = datetime.now(UTC)
        return report

    cutoff = datetime.now(UTC) - timedelta(hours=self._lookback_hours)
    recent_nodes = [n for n in all_nodes if n.updated_at and n.updated_at >= cutoff]

    # Phase 1: Strengthen recurring nodes
    report.phases_run.append("strengthen")
    try:
        report.nodes_strengthened = await phases.strengthen(
            self._graph, scope, recent_nodes,
            spreading_weight=self._phase1_spreading_weight,
        )
    except Exception as exc:
        logger.warning("quick_nap phase 1 failed: %s", exc)
        report.errors.append(f"quick_nap phase 1 (strengthen): {exc}")

    # Phase 5: profile-fact promotion onto the tenant's SOUL node
    # (auto-forwarded by the engine from domain.soul_schema, see
    # ``profile_fields`` above). Idempotent on repeat calls:
    # ``promote_profile_corrections`` re-stamps its own writes with
    # ``PROMOTION_MARKER`` so a promoted node is never re-read as a
    # fresh user correction on the next pass (see phases_profile.py).
    effective_profile_fields = _resolve_profile_fields(
        profile_fields, soul_schema,
    )
    if effective_profile_fields is not None:
        report.phases_run.append("soul_corrections")
        try:
            count = await promote_profile_corrections(
                self._graph, scope, recent_nodes, effective_profile_fields,
            )
            report.soul_updates = count
            report.profile_updates = count
        except Exception as exc:
            logger.warning("quick_nap phase 5 failed: %s", exc)
            report.errors.append(f"quick_nap phase 5 (soul): {exc}")

    # Phase 10: Episodic summarization (only if episodic layer provided).
    # Per-call overrides fall back to instance defaults so call-sites that
    # do not thread FrameworkConfig still respect the ctor-time config.
    if self._episodic is not None:
        report.phases_run.append("episodic_summary")
        effective_max = (
            max_entries
            if max_entries is not None
            else self._episodic_summarization_max_entries
        )
        effective_batch = (
            summarize_batch
            if summarize_batch is not None
            else self._episodic_summarization_batch_size
        )
        try:
            _summary_text, deleted = await summarize_episodic(
                self._episodic,
                scope,
                llm_summarise=self._llm_summarise,
                max_entries=effective_max,
                summarize_batch=effective_batch,
            )
            report.episodes_summarized = deleted
        except Exception as exc:
            logger.warning("quick_nap phase 10 failed: %s", exc)
            report.errors.append(f"quick_nap phase 10 (episodic): {exc}")

    # Phase 13: LLM-judged semantic-duplicate merge (v8.17). Added to
    # the quick_nap roster because quick_nap is the ONLY cadence the
    # engine schedules in-process -- Phase 13 living solely in run()
    # made ``enable_semantic_merge`` dead config for every adopter who
    # did not own a scheduler (adopter-verified against 9.6.0).
    # Default-off gate keeps the pre-8.17 roster byte-identical; the
    # per-run pair budget bounds the LLM cost on this hot-ish path.
    if self._enable_semantic_merge and self._chat_model is not None:
        report.phases_run.append("semantic_merge")
        try:
            from symfonic.core.learning.phases_semantic_merge import (
                merge_semantic_duplicates,
            )

            report.semantic_merges = await merge_semantic_duplicates(
                self._graph,
                scope,
                chat_model=self._chat_model,
                embedding_provider=(
                    self._semantic_merge_embedding_provider
                ),
                embedding_threshold=(
                    self._semantic_merge_embedding_threshold
                ),
                lexical_threshold=(
                    self._semantic_merge_lexical_threshold
                ),
                max_pairs_per_run=(
                    self._semantic_merge_max_pairs_per_run
                ),
            )
        except Exception as exc:
            logger.warning(
                "quick_nap phase 13 (semantic_merge) failed: %s", exc
            )
            report.errors.append(
                f"quick_nap phase 13 (semantic_merge): {exc}"
            )

    report.finished_at = datetime.now(UTC)
    logger.debug(
        "quick_nap complete: tenant=%s strengthened=%d episodes_summarized=%d errors=%d",
        scope.tenant_id,
        report.nodes_strengthened,
        report.episodes_summarized,
        len(report.errors),
    )
    return report

run `async` ¶

run(scope: TenantScope, *, soul_schema: dict[str, Any] | None = None, profile_fields: frozenset[str] | None = None, pending_connections: list[dict[str, Any]] | None = None) -> ConsolidationReport

Execute the full consolidation pipeline.

Parameters:

Name	Type	Description	Default
`scope`	`TenantScope`	Tenant isolation scope.	required
`soul_schema`	`dict[str, Any] \| None`	Deprecated. Pass `profile_fields` instead. When supplied without `profile_fields`, this function derives `frozenset(soul_schema.keys())` and emits a `DeprecationWarning`. The dict is read-only here (only `.keys()` is used) -- never mutated.	`None`
`profile_fields`	`frozenset[str] \| None`	Field names that constitute the tenant's profile (e.g. `frozenset(domain.soul_schema.keys())`). Activates Phase 5 (profile-fact promotion onto the tenant's SOUL node); `None` (and no `soul_schema`) skips Phase 5 entirely.	`None`
`pending_connections`	`list[dict[str, Any]] \| None`	Inferred edges to materialise in Phase 2.5.	`None`

Source code in symfonic/core/learning/consolidation.py

async def run(
    self,
    scope: TenantScope,
    *,
    soul_schema: dict[str, Any] | None = None,
    profile_fields: frozenset[str] | None = None,
    pending_connections: list[dict[str, Any]] | None = None,
) -> ConsolidationReport:
    """Execute the full consolidation pipeline.

    Args:
        scope: Tenant isolation scope.
        soul_schema: Deprecated.  Pass ``profile_fields`` instead.
            When supplied without ``profile_fields``, this function
            derives ``frozenset(soul_schema.keys())`` and emits a
            ``DeprecationWarning``.  The dict is read-only here (only
            ``.keys()`` is used) -- never mutated.
        profile_fields: Field names that constitute the tenant's
            profile (e.g. ``frozenset(domain.soul_schema.keys())``).
            Activates Phase 5 (profile-fact promotion onto the
            tenant's SOUL node); ``None`` (and no ``soul_schema``)
            skips Phase 5 entirely.
        pending_connections: Inferred edges to materialise in Phase 2.5.
    """
    from symfonic.core.learning.durability_gate import DurabilityGate

    report = ConsolidationReport(tenant_id=scope.tenant_id)
    # v7.26.2: one gate instance shared across Phases 10/11/12 so the
    # per-run transient/expired rejection tally is aggregated.
    gate = DurabilityGate()
    cutoff = datetime.now(UTC) - timedelta(hours=self._lookback_hours)

    logger.info(
        "Consolidation starting: tenant=%s lookback_hours=%d pending_connections=%d",
        scope.tenant_id,
        self._lookback_hours,
        len(pending_connections) if pending_connections else 0,
    )

    try:
        all_nodes = await self._graph.query_nodes(
            scope, layer=MemoryLayer.SEMANTIC,
        )
    except Exception as exc:
        report.errors.append(f"Failed to query nodes: {exc}")
        report.finished_at = datetime.now(UTC)
        return report

    recent_nodes = [
        n for n in all_nodes
        if n.updated_at and n.updated_at >= cutoff
    ]
    logger.info(
        "Consolidation nodes: total=%d recent=%d",
        len(all_nodes), len(recent_nodes),
    )

    # Each phase appends its canonical name to ``report.phases_run``
    # BEFORE invocation so a half-way failure still records the fact
    # the phase started. Conditionally-skipped phases (2.5 / 5 / 10 /
    # 11 / 12) only append inside the branch that actually runs --
    # omission from the list proves short-circuit, presence with a
    # zero counter proves "ran and found nothing". Names match the
    # `phase_<n>_*` test file suffixes for easy correlation.

    # Phase 1: Strengthen recurring nodes
    report.phases_run.append("strengthen")
    try:
        report.nodes_strengthened = await phases.strengthen(
            self._graph, scope, recent_nodes,
            spreading_weight=self._phase1_spreading_weight,
        )
    except Exception as exc:
        logger.warning("phase 1 (strengthen) failed: %s", exc)
        report.errors.append(f"phase 1 (strengthen): {exc}")

    # Phase 1.5: Co-occurrence edges
    report.phases_run.append("cooccurrence")
    try:
        report.edges_created += await phases.create_cooccurrence_edges(
            self._graph, scope, recent_nodes,
        )
    except Exception as exc:
        logger.warning("phase 1.5 (cooccurrence) failed: %s", exc)
        report.errors.append(f"phase 1.5 (cooccurrence): {exc}")

    # Phase 2: Tag risk nodes
    report.phases_run.append("tag_risk")
    try:
        report.risk_nodes_tagged = await phases.tag_risk_nodes(
            self._graph, scope, recent_nodes,
        )
    except Exception as exc:
        logger.warning("phase 2 (tag_risk) failed: %s", exc)
        report.errors.append(f"phase 2 (tag_risk): {exc}")

    # Phase 2.5: Consolidate pending (inferred) edges
    if pending_connections:
        report.phases_run.append("pending_edges")
        try:
            report.edges_created += await phases.consolidate_pending_edges(
                self._graph, scope, all_nodes, pending_connections,
            )
        except Exception as exc:
            logger.warning("phase 2.5 (pending_edges) failed: %s", exc)
            report.errors.append(f"phase 2.5 (pending_edges): {exc}")

    # Phase 3: Prune orphans
    report.phases_run.append("prune_orphans")
    try:
        report.nodes_pruned = await phases.prune_orphans(
            self._graph, scope, all_nodes,
        )
    except Exception as exc:
        logger.warning("phase 3 (prune_orphans) failed: %s", exc)
        report.errors.append(f"phase 3 (prune_orphans): {exc}")

    # Phase 4: Generate meta-nodes from clusters
    report.phases_run.append("meta_nodes")
    try:
        report.meta_nodes_created = await phases.generate_meta_nodes(
            self._graph, scope, all_nodes, self._llm_summarise,
        )
    except Exception as exc:
        logger.warning("phase 4 (meta_nodes) failed: %s", exc)
        report.errors.append(f"phase 4 (meta_nodes): {exc}")

    # Phase 5: profile-fact promotion onto the tenant's SOUL node.
    # ``_resolve_profile_fields`` honours the deprecated soul_schema=
    # kwarg (read-only) without ever mutating the caller's dict.
    effective_profile_fields = _resolve_profile_fields(
        profile_fields, soul_schema,
    )
    if effective_profile_fields is not None:
        report.phases_run.append("soul_corrections")
        try:
            count = await promote_profile_corrections(
                self._graph, scope, recent_nodes, effective_profile_fields,
            )
            report.soul_updates = count
            report.profile_updates = count
        except Exception as exc:
            logger.warning("phase 5 (soul_corrections) failed: %s", exc)
            report.errors.append(f"phase 5 (soul_corrections): {exc}")

    # Phase 8: Working TTL cleanup
    report.phases_run.append("working_ttl")
    try:
        report.working_ttl_pruned = await cleanup_working_ttl(
            self._graph, scope,
        )
    except Exception as exc:
        logger.warning("phase 8 (working_ttl) failed: %s", exc)
        report.errors.append(f"phase 8 (working_ttl): {exc}")

    # Phase 9: Semantic importance decay
    report.phases_run.append("decay_importance")
    try:
        report.nodes_decayed = await decay_importance(
            self._graph, scope, all_nodes,
        )
    except Exception as exc:
        logger.warning("phase 9 (decay_importance) failed: %s", exc)
        report.errors.append(f"phase 9 (decay_importance): {exc}")

    # Phase 9.2 (v9.10, J1+J2): durability/validity-driven SEMANTIC
    # expiry. Soft-retracts nodes whose ``durability`` marker expired,
    # whose transient TTL elapsed, or whose ``valid_until`` passed --
    # see phases_semantic_expiry.py for the full design rationale
    # (§2.5a of the prompt-part-blocks architecture doc). Runs after
    # decay so a node's importance reflects its pre-expiry state in
    # the audit trail, and before prune_retracted so newly-stamped
    # retractions are visible to this run's report immediately (they
    # are only hard-deleted in a LATER run, past the grace window).
    report.phases_run.append("expire_semantic_durability")
    try:
        report.nodes_expired = await expire_semantic_durability(
            self._graph, scope, all_nodes,
        )
    except Exception as exc:
        logger.warning("phase 9.2 (expire_semantic_durability) failed: %s", exc)
        report.errors.append(f"phase 9.2 (expire_semantic_durability): {exc}")

    # Phase 9.5: Prune soft-retracted nodes past their grace window.
    # Soft-retracted (false-positive/corrected) nodes are already invisible
    # to reads via the query_nodes exclusion; this reclaims them once the
    # reversibility window has elapsed.
    report.phases_run.append("prune_retracted")
    try:
        report.retracted_nodes_pruned = await prune_retracted(
            self._graph, scope,
        )
    except Exception as exc:
        logger.warning("phase 9.5 (prune_retracted) failed: %s", exc)
        report.errors.append(f"phase 9.5 (prune_retracted): {exc}")

    # Phase 10: Episodic summarization
    if self._episodic is not None:
        report.phases_run.append("episodic_summary")
        try:
            summary_text, deleted = await summarize_episodic(
                self._episodic,
                scope,
                llm_summarise=self._llm_summarise,
                max_entries=self._episodic_summarization_max_entries,
                summarize_batch=self._episodic_summarization_batch_size,
                gate=gate,
            )
            report.episodes_summarized = deleted
            if summary_text:
                # Late import to avoid circular import via core.__init__
                from symfonic.memory.models.node import MemoryNode

                # Write summary as a semantic meta-node
                meta = MemoryNode(
                    layer=MemoryLayer.SEMANTIC,
                    tenant_id=scope.tenant_id,
                    label="META:episodic_summary",
                    properties={
                        "type": "episodic_summary",
                        "summary": summary_text,
                        "entries_compressed": deleted,
                    },
                    importance=6.0,
                )
                try:
                    await self._graph.add_node(scope, meta)
                    logger.info(
                        "Episodic summary meta-node created: %d entries compressed",
                        deleted,
                    )
                except Exception:
                    logger.debug(
                        "Failed to create episodic summary meta-node",
                        exc_info=True,
                    )
        except Exception as exc:
            logger.warning("phase 10 (episodic_summary) failed: %s", exc)
            report.errors.append({"phase": "episodic_summary", "error": str(exc)})

    # Phase 11: Synthetic linking (episodic -> semantic cross-pollination)
    if self._episodic is not None:
        report.phases_run.append("synthetic_links")
        try:
            from symfonic.core.learning.phases_synthetic import (
                build_synthetic_links,
            )

            report.synthetic_edges_created = await build_synthetic_links(
                self._graph, self._episodic, scope,
                min_co_count=self._synthetic_link_min_co_count,
                gate=gate,
            )
        except Exception as exc:
            logger.warning("phase 11 failed: %s", exc)
            report.errors.append(f"phase 11 (synthetic): {exc}")

    # Phase 12.5: EntityLinker (v7.2 Roadmap Item 12). Default-off via
    # ``enable_entity_linker``. Skipping the flag short-circuits the
    # whole block so ``"entity_links"`` does NOT appear in
    # ``report.phases_run`` (Decision 7.5 / acceptance #1).
    if self._enable_entity_linker and self._episodic is not None:
        report.phases_run.append("entity_links")
        try:
            from symfonic.core.learning.phases_entity import build_entity_links

            extractor = self._build_entity_extractor()
            nodes_added, edges_added = await build_entity_links(
                self._graph, self._episodic, scope,
                extractor=extractor,
                min_mention_count=self._entity_linker_min_mention_count,
                max_episodics_per_run=self._entity_linker_max_episodics_per_run,
                confidence_threshold=self._entity_linker_confidence_threshold,
            )
            report.entity_nodes_created = nodes_added
            report.mentions_edges_created = edges_added
        except Exception as exc:
            logger.warning("phase 12.5 (entity_links) failed: %s", exc)
            report.errors.append(f"phase_12.5 (entity_links): {exc}")

    # Phase 12: Episodic -> procedural promotion (draft skills)
    if self._episodic is not None and self._procedural is not None:
        report.phases_run.append("procedural_promotion")
        try:
            from symfonic.core.learning.phases_procedural_promotion import (
                promote_episodic_to_procedural,
            )

            report.draft_skills_promoted = await promote_episodic_to_procedural(
                self._episodic,
                self._procedural,
                scope,
                min_pattern_count=self._promotion_min_pattern_count,
                recency_days=self._promotion_recency_days,
                max_drafts_per_run=self._promotion_max_drafts_per_run,
                use_tool_calls_fallback=(
                    self._promotion_use_tool_calls_fallback
                ),
                # v7.6: pass the LLM extractor instance when opt-in;
                # ``None`` keeps the regex path running unchanged.
                llm_extractor=self._build_procedural_extractor(),
                llm_model_name=self._phase_12_llm_model,
                # v7.7.2: speaker filter for both extractor paths.
                promote_assistant_content=(
                    self._promotion_promote_assistant_content
                ),
                gate=gate,
            )
        except Exception as exc:
            logger.warning("phase 12 failed: %s", exc)
            report.errors.append(f"phase_12: {exc}")

    # Phase 13: LLM-judged semantic-duplicate merge (v8.16,
    # default-off via ``enable_semantic_merge``; also requires a
    # ``chat_model`` -- the judge IS the correctness gate).  Runs
    # last so it sees the nodes the earlier phases produced.  Newly
    # stamped supersede-retractions are reclaimed by phase 9.5
    # (prune_retracted) in a later run, after the grace window.
    if self._enable_semantic_merge and self._chat_model is not None:
        report.phases_run.append("semantic_merge")
        try:
            from symfonic.core.learning.phases_semantic_merge import (
                merge_semantic_duplicates,
            )

            report.semantic_merges = await merge_semantic_duplicates(
                self._graph,
                scope,
                chat_model=self._chat_model,
                embedding_provider=(
                    self._semantic_merge_embedding_provider
                ),
                embedding_threshold=(
                    self._semantic_merge_embedding_threshold
                ),
                lexical_threshold=(
                    self._semantic_merge_lexical_threshold
                ),
                max_pairs_per_run=(
                    self._semantic_merge_max_pairs_per_run
                ),
            )
        except Exception as exc:
            logger.warning("phase 13 (semantic_merge) failed: %s", exc)
            report.errors.append(f"phase_13 (semantic_merge): {exc}")

    # v7.26.2: surface the aggregate durability-gate rejection tally.
    report.transient_rejected = gate.rejected

    report.finished_at = datetime.now(UTC)
    logger.info(
        "Consolidation complete: tenant=%s strengthened=%d pruned=%d "
        "edges_created=%d meta_nodes=%d soul_updates=%d "
        "working_ttl_pruned=%d nodes_decayed=%d episodes_summarized=%d "
        "errors=%d",
        scope.tenant_id,
        report.nodes_strengthened,
        report.nodes_pruned,
        report.edges_created,
        report.meta_nodes_created,
        report.soul_updates,
        report.working_ttl_pruned,
        report.nodes_decayed,
        report.episodes_summarized,
        len(report.errors),
    )
    return report

build_entity_extractor ¶

build_entity_extractor(kind: str = 'regex', *, chat_model: Any = None) -> Any

Build the entity extractor selected by a deployment.

This is the public composition door used by workers. Returning a regex fallback when an optional integration is unavailable preserves Deep Sleep while making the degradation visible in its operational log.

Source code in symfonic/core/learning/__init__.py

def build_entity_extractor(kind: str = "regex", *, chat_model: Any = None) -> Any:
    """Build the entity extractor selected by a deployment.

    This is the public composition door used by workers.  Returning a regex
    fallback when an optional integration is unavailable preserves Deep Sleep
    while making the degradation visible in its operational log.
    """
    from symfonic.capabilities.memory.phases.entity import RegexHeuristicExtractor

    selected = (kind or "regex").strip().lower()
    if selected == "regex":
        return RegexHeuristicExtractor()
    try:
        if selected == "spacy":
            from symfonic.core.learning.entity_extractor_spacy import (
                SpacyEntityExtractor,
            )

            return SpacyEntityExtractor()
        if selected == "llm":
            if chat_model is None:
                raise RuntimeError("the llm entity extractor needs chat_model")
            from symfonic.core.learning.entity_extractor_llm import LlmEntityExtractor

            return LlmEntityExtractor(chat_model=chat_model)
    except Exception as exc:  # noqa: BLE001 - optional integration fallback
        logger.warning(
            "entity extractor %s unavailable (%s); falling back to regex",
            selected,
            exc,
        )
        return RegexHeuristicExtractor()
    raise ValueError(
        f"unknown entity extractor {kind!r}; expected regex, spacy, or llm"
    )

symfonic.core.learning¶

learning ¶

ConsolidationReport dataclass ¶

InsightExtractor ¶

extract async ¶

LearnedInsight dataclass ¶

LearningPolicy dataclass ¶

as_kwargs ¶

as_phase_kwargs ¶

from_settings classmethod ¶

SleepConsolidator ¶

bind_layers ¶

nightly_nap async ¶

quick_nap async ¶

run async ¶

build_entity_extractor ¶

`symfonic.core.learning`¶

ConsolidationReport `dataclass` ¶

extract `async` ¶

LearnedInsight `dataclass` ¶

LearningPolicy `dataclass` ¶

from_settings `classmethod` ¶

nightly_nap `async` ¶

quick_nap `async` ¶

run `async` ¶