symfonic.core.learning.procedural_extractor_llm

procedural_extractor_llm

LLM-backed procedural extractor (v7.6 — Jarvio Ask 9 Option A).

Single LLM call per tenant per nightly_nap. Reads recent episodic entries enriched with the v7.4.7 metadata['tool_calls'] trace producer, asks the model to identify durable action / preflight / recovery / guardrail rules, and emits structured :class:~symfonic.core.learning.procedural_extractor.ProceduralDraft records that the dispatch site in :mod:symfonic.core.learning.phases_procedural_promotion turns into draft :class:~symfonic.memory.models.entry.MemoryEntry writes via :func:~symfonic.memory.layers.procedural.layer.ProceduralLayer.store_skill.

Lift-and-shift from :mod:symfonic.core.learning.entity_extractor_llm (v7.2) — same best-effort JSON parsing contract: malformed responses, missing fields, and model-invocation exceptions all yield [] rather than raise.

Default-off ship: the extractor is reachable only when FrameworkConfig.phase_12_use_llm_extractor=True AND the consolidator has a bound chat model. Either-or with the regex extractor (the eval recommended this over a merged stream).

Design reference: .claude/docs/2026-05-28-jarvio-llm-procedural-extractor-eval.md §8.2 / §8.3.

LlmProceduralExtractor

LlmProceduralExtractor(
    chat_model: Any,
    *,
    max_drafts_per_run: int = 10,
    max_episodes_per_run: int = 100,
    observability_hook: ObservabilityHook | None = None,
)

LLM-backed Phase 12 procedural-draft extractor.

Implements :class:~symfonic.core.learning.procedural_extractor.ProceduralExtractor.

Construct the extractor.

Parameters:

Name	Type	Description	Default
chat_model	Any	A LangChain-compatible chat model exposing ainvoke (or invoke as a sync fallback for stubs). None is tolerated; extract then degrades to [] -- the consolidator factory only builds the extractor when a chat model is available but this guard keeps the contract robust for direct instantiation in tests.	required
max_drafts_per_run	int	Hard cap on drafts the extractor will return per call. Default 5 mirrors the regex extractor's DEFAULT_MAX_DRAFTS_PER_RUN so the review queue cannot flood when consumers flip to the LLM path without re-tuning.	10
max_episodes_per_run	int	Hard cap on episodes fed into the prompt. Excess episodes are dropped from the tail (older entries) so the freshest context wins.	100

Source code in src/symfonic/core/learning/procedural_extractor_llm.py

def __init__(
    self,
    chat_model: Any,
    *,
    max_drafts_per_run: int = 10,
    max_episodes_per_run: int = 100,
    observability_hook: ObservabilityHook | None = None,
) -> None:
    """Construct the extractor.

    Args:
        chat_model: A LangChain-compatible chat model exposing
            ``ainvoke`` (or ``invoke`` as a sync fallback for
            stubs).  ``None`` is tolerated; ``extract`` then
            degrades to ``[]`` -- the consolidator factory only
            builds the extractor when a chat model is available
            but this guard keeps the contract robust for direct
            instantiation in tests.
        max_drafts_per_run: Hard cap on drafts the extractor will
            return per call.  Default 5 mirrors the regex
            extractor's ``DEFAULT_MAX_DRAFTS_PER_RUN`` so the
            review queue cannot flood when consumers flip to the
            LLM path without re-tuning.
        max_episodes_per_run: Hard cap on episodes fed into the
            prompt.  Excess episodes are dropped from the tail
            (older entries) so the freshest context wins.
    """
    self._chat_model = chat_model
    self._max_drafts_per_run = max_drafts_per_run
    self._max_episodes_per_run = max_episodes_per_run
    # v8.3.2 observability-bug fix (G7): the procedural-extractor LLM
    # call emitted ONLY on the merge-dependent callback manager (same
    # bug class as the v8.3.1 critic).  When a hook is supplied it also
    # emits on the always-on ``ObservabilityHook``.  ``None`` resolves
    # to ``NoOpObservabilityHook`` so the default is byte-identical.
    # NOTE: the live caller (background ``SleepConsolidator`` ->
    # Phase 12 ``phases_procedural_promotion``) does NOT yet thread a
    # real hook -- see v8.3.2 report for the plumbing this would need.
    if observability_hook is None:
        from symfonic.core.observability.hooks import NoOpObservabilityHook

        observability_hook = NoOpObservabilityHook()
    self._observability_hook = observability_hook

extract async

extract(
    episodes: list[MemoryEntry],
    existing_skills: list[MemoryEntry],
    scope: TenantScope,
    *,
    callback_manager: CallbackManager | None = None,
    run_id: str = "",
    model_name: str = "procedural_extractor_llm",
) -> list[ProceduralDraft]

Extract procedural drafts from recent episodes.

Returns [] rather than raising on every failure path: empty input, None chat model, model exception, malformed JSON, missing fields. The dispatch site logs the empty result via the standard Phase 12 telemetry line.

Parameters:

Name	Type	Description	Default
episodes	list[MemoryEntry]	Recent episodic entries (already filtered to recency_days by the caller).	required
existing_skills	list[MemoryEntry]	Procedural skills already in storage so the LLM can avoid duplicating them.	required
scope	TenantScope	Tenant scope -- forwarded for future per-tenant prompt customisation; not used by the default prompt.	required
callback_manager	CallbackManager | None	Optional v7.4.3 callback manager. When provided, fires on_llm_end(node_name="procedural_extractor_llm") so the call shows up in token billing / OTel.	None
run_id	str	Engine run identifier for callback correlation.	''
model_name	str	Identifier stamped on the LLMEndEvent. Defaults to the module identifier; consumers that want to surface the actual SKU should pass the resolved model string.	'procedural_extractor_llm'

Returns:

Type	Description
list[ProceduralDraft]	A list of :class:ProceduralDraft records, length
list[ProceduralDraft]	<= max_drafts_per_run.

Source code in src/symfonic/core/learning/procedural_extractor_llm.py

async def extract(
    self,
    episodes: list[MemoryEntry],
    existing_skills: list[MemoryEntry],
    scope: TenantScope,
    *,
    callback_manager: CallbackManager | None = None,
    run_id: str = "",
    model_name: str = "procedural_extractor_llm",
) -> list[ProceduralDraft]:
    """Extract procedural drafts from recent episodes.

    Returns ``[]`` rather than raising on every failure path: empty
    input, ``None`` chat model, model exception, malformed JSON,
    missing fields.  The dispatch site logs the empty result via
    the standard Phase 12 telemetry line.

    Args:
        episodes: Recent episodic entries (already filtered to
            ``recency_days`` by the caller).
        existing_skills: Procedural skills already in storage so
            the LLM can avoid duplicating them.
        scope: Tenant scope -- forwarded for future per-tenant
            prompt customisation; not used by the default prompt.
        callback_manager: Optional v7.4.3 callback manager.  When
            provided, fires
            ``on_llm_end(node_name="procedural_extractor_llm")``
            so the call shows up in token billing / OTel.
        run_id: Engine run identifier for callback correlation.
        model_name: Identifier stamped on the LLMEndEvent.
            Defaults to the module identifier; consumers that want
            to surface the actual SKU should pass the resolved
            model string.

    Returns:
        A list of :class:`ProceduralDraft` records, length
        <= ``max_drafts_per_run``.
    """
    if not episodes or self._chat_model is None:
        return []

    capped_episodes = episodes[: self._max_episodes_per_run]
    prompt = _EXTRACTION_PROMPT.format(
        max_drafts=self._max_drafts_per_run,
        existing_skills=self._format_existing_skills(existing_skills),
        episodes=self._format_episodes(capped_episodes),
    )
    try:
        response = await self._invoke_model(
            prompt,
            callback_manager=callback_manager,
            run_id=run_id,
            model_name=model_name,
        )
    except Exception:
        logger.debug(
            "procedural_extractor_llm: model invocation failed",
            exc_info=True,
        )
        return []
    return self._parse_response(response)

draft_to_memory_entry

draft_to_memory_entry(
    draft: ProceduralDraft, scope: TenantScope
) -> MemoryEntry

Convert an :class:ProceduralDraft to a procedural MemoryEntry.

Used by the Phase 12 dispatch site in :func:promote_episodic_to_procedural so the LLM and regex paths share the same store_skill write contract. The draft's structured fields land in the entry metadata under stable keys so the v7.5 query_skills projection surfaces them:

• precondition -> properties['precondition'] (v7.5)
• source -> "phase_12_llm_extractor" (distinct from the regex extractor's "phase_12_promotion" so SRE telemetry can split adoption).
• kind / confidence / rationale / trigger / source_episode_ids are first-class properties for v7.7+ router enrichment.

The label field follows the v6.1 "Auto-learned: <name>" convention so downstream label-prefix grouping (e.g. SemanticMerge similarity, /procedures listing UI) keeps working unchanged.

Source code in src/symfonic/core/learning/procedural_extractor_llm.py

def draft_to_memory_entry(
    draft: ProceduralDraft,
    scope: TenantScope,
) -> MemoryEntry:
    """Convert an :class:`ProceduralDraft` to a procedural ``MemoryEntry``.

    Used by the Phase 12 dispatch site in
    :func:`promote_episodic_to_procedural` so the LLM and regex paths
    share the same ``store_skill`` write contract.  The draft's
    structured fields land in the entry metadata under stable keys so
    the v7.5 ``query_skills`` projection surfaces them:

    * ``precondition`` -> ``properties['precondition']`` (v7.5)
    * ``source``       -> ``"phase_12_llm_extractor"`` (distinct from
      the regex extractor's ``"phase_12_promotion"`` so SRE telemetry
      can split adoption).
    * ``kind`` / ``confidence`` / ``rationale`` / ``trigger`` /
      ``source_episode_ids`` are first-class properties for v7.7+
      router enrichment.

    The ``label`` field follows the v6.1 ``"Auto-learned: <name>"``
    convention so downstream label-prefix grouping (e.g. SemanticMerge
    similarity, /procedures listing UI) keeps working unchanged.
    """
    from symfonic.memory.types import MemoryLayer

    # Compose a content string that the SemanticMerge janitor and
    # human reviewers can scan at a glance.  The trigger is the
    # one-line gate; we add the action + rationale so the content
    # survives Phase 12 telemetry truncation in audit logs.
    content = (
        f"{draft.trigger}\nAction: {draft.action}\n"
        f"Rationale: {draft.rationale}"
    )
    label = f"Auto-learned ({draft.kind}): {draft.trigger[:80]}"[:100]
    return MemoryEntry(
        layer=MemoryLayer.PROCEDURAL,
        tenant_id=scope.tenant_id,
        content=content,
        importance=5.0,
        metadata={
            "label": label,
            "steps": [draft.action],
            "context": draft.rationale,
            "source": "phase_12_llm_extractor",
            "kind": draft.kind,
            "trigger": draft.trigger,
            "rationale": draft.rationale,
            "confidence": draft.confidence,
            "source_episode_ids": list(draft.source_episode_ids),
            **(
                {"precondition": draft.precondition}
                if draft.precondition else {}
            ),
        },
    )