symfonic.capabilities.memory.phases.procedural_llm

procedural_llm

Phase 12's other extractor: ask a model which procedures repeat.

Split from :mod:.procedural because that module crossed its 300-line budget on the way into this package, and this is the split the code already draws -- one extractor reads patterns with a regex, the other asks a model, and they share only the draft-writing tail.

v7.6 adopter Ask 9 Option A: either-or dispatch. When an extractor is supplied it runs instead of the regex one; the eval rejected merged streams.

run_llm_extractor async

run_llm_extractor(llm_extractor: Any, procedural_layer: ProceduralLayer, scope: TenantScope, *, recent_episodes: list[MemoryEntry], max_drafts_per_run: int, model_name: str, promote_assistant_content: bool = False, gate: DurabilityGate | None = None) -> int

Run the LLM-backed procedural extractor and write its drafts.

Mirrors the regex extractor's tail (existing-skill snapshot -> store_skill loop -> dedup-collision count -> summary log line) so SRE telemetry can compare the two paths apples-to-apples and consumers can flip phase_12_use_llm_extractor without changing their log parsers.

The Phase 12 LLM summary log line carries:

Phase 12 LLM: tenant=<id> drafts_emitted=K drafts_kept=K' \
  drafts_rejected_by_dedup=K'' max_drafts=N model=<name> \
  cost_usd=<calc-or-0>

cost_usd is best-effort: when the extractor returns 0 drafts on a failed model call the helper still emits a summary line so consumers can distinguish "no signal" from "extractor crashed silently" (matches the v7.4.8 rationale on the regex extractor).

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

async def run_llm_extractor(
    llm_extractor: Any,
    procedural_layer: ProceduralLayer,
    scope: TenantScope,
    *,
    recent_episodes: list[MemoryEntry],
    max_drafts_per_run: int,
    model_name: str,
    promote_assistant_content: bool = False,
    gate: DurabilityGate | None = None,
) -> int:
    """Run the LLM-backed procedural extractor and write its drafts.

    Mirrors the regex extractor's tail (existing-skill snapshot ->
    ``store_skill`` loop -> dedup-collision count -> summary log line)
    so SRE telemetry can compare the two paths apples-to-apples and
    consumers can flip ``phase_12_use_llm_extractor`` without changing
    their log parsers.

    The Phase 12 LLM summary log line carries:

        Phase 12 LLM: tenant=<id> drafts_emitted=K drafts_kept=K' \\
          drafts_rejected_by_dedup=K'' max_drafts=N model=<name> \\
          cost_usd=<calc-or-0>

    ``cost_usd`` is best-effort: when the extractor returns 0 drafts on
    a failed model call the helper still emits a summary line so
    consumers can distinguish "no signal" from "extractor crashed
    silently" (matches the v7.4.8 rationale on the regex extractor).
    """
    try:
        existing_entries = await procedural_layer.query_skills(
            scope, "", top_k=1000, include_drafts=True,
        )
        existing_ids = {e.node_id for e in existing_entries if e.node_id}
    except Exception:
        existing_entries = []
        existing_ids = set()

    # v7.7.2: speaker filter on the LLM extractor input.  The default
    # policy (``promote_assistant_content=False``) removes assistant-
    # narrated rows AND legacy mixed-speaker rows from the prompt the
    # model sees, so it cannot identify procedural skills from the
    # agent's own talking.  Action grounding from
    # ``metadata['tool_calls']`` is preserved on the user-row turn
    # neighbours and reaches the extractor's prompt via the user-row
    # itself.
    _gate = gate if gate is not None else DurabilityGate()
    if promote_assistant_content:
        filtered_episodes = [
            e for e in recent_episodes if not _gate.is_gated(e, phase="12")
        ]
    else:
        filtered_episodes = [
            e for e in recent_episodes
            if not _gate.is_gated(e, phase="12")
            and not (
                _is_assistant_narrated(e)
                or _has_legacy_mixed_speaker_content(e)
            )
        ]

    try:
        drafts = await llm_extractor.extract(
            filtered_episodes, existing_entries, scope,
        )
    except Exception:
        logger.debug("Phase 12 LLM: extract() raised", exc_info=True)
        drafts = []

    # Honour the consolidator's hard cap even if the extractor returned
    # more drafts (defensive -- the LLM extractor already caps to its
    # own ``max_drafts_per_run`` so this is a second belt).
    drafts = list(drafts)[:max_drafts_per_run]

    created = 0
    drafts_rejected_by_dedup = 0
    if drafts:
        for draft in drafts:
            skill_entry = draft_to_memory_entry(draft, scope)
            try:
                node = await procedural_layer.store_skill(
                    scope, skill_entry, status="draft",
                )
            except Exception:
                logger.debug(
                    "Phase 12 LLM: store_skill failed for draft %r",
                    draft.trigger,
                    exc_info=True,
                )
                continue
            if node.id in existing_ids:
                drafts_rejected_by_dedup += 1
                continue
            existing_ids.add(node.id)
            created += 1

    logger.info(
        "Phase 12 LLM: tenant=%s drafts_emitted=%d drafts_kept=%d "
        "drafts_rejected_by_dedup=%d max_drafts=%d model=%s",
        scope.tenant_id, len(drafts), created,
        drafts_rejected_by_dedup, max_drafts_per_run,
        model_name or "(unspecified)",
    )
    return created