symfonic.memory._internal.extraction.service

service

ExtractionService -- extracts structured MemoryOperations from conversation turns.

Owns the LLM prompt construction, invocation, and response parsing for memory extraction. Separated from the orchestrator to honour SRP.

ExtractionService

ExtractionService(
    *, observability_hook: ObservabilityHook | None = None
)

Extracts structured MemoryOperations from conversation turns via LLM.

Source code in src/symfonic/memory/_internal/extraction/service.py

def __init__(
    self, *, observability_hook: ObservabilityHook | None = None
) -> None:
    # v8.3.2 observability-bug fix (G4): the consolidation 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 now
    # also emits on the always-on ``ObservabilityHook``.  ``None``
    # resolves to ``NoOpObservabilityHook`` so the default is
    # byte-identical (additive only).  NOTE: the live in-tree caller
    # (``MemoryOrchestrator`` -> background ``SleepConsolidator``) does
    # NOT yet thread a real hook -- see v8.3.2 CHANGELOG / report; this
    # parameter lets direct callers wire one without invasive plumbing.
    if observability_hook is None:
        from symfonic.core.observability.hooks import NoOpObservabilityHook

        observability_hook = NoOpObservabilityHook()
    self._observability_hook = observability_hook

extract async

extract(
    scope: TenantScope,
    interaction: dict[str, Any],
    llm: Any,
    *,
    callback_manager: CallbackManager | None = None,
    run_id: str = "",
    model_name: str = "consolidation_extractor",
) -> list[MemoryOperation]

Analyse an interaction and return memory operations.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
interaction	dict[str, Any]	Dict with 'user_message' and 'assistant_response'.	required
llm	Any	LLM instance (BaseChatModel from langchain-core).	required
callback_manager	CallbackManager | None	v7.4.3 (Jarvio Ask 5) -- when provided, fires on_llm_end(node_name="consolidation_extractor") after the LLM call so the (opt-in) consolidation extractor is visible to token billing / OTel. None preserves pre-7.4.3 behaviour.	None
run_id	str	Engine run identifier for callback correlation.	''
model_name	str	Optional model identifier stamped on the LLMEndEvent. Defaults to "consolidation_extractor".	'consolidation_extractor'

Returns:

Type	Description
list[MemoryOperation]	List of MemoryOperations to be applied via commit_pending.

Source code in src/symfonic/memory/_internal/extraction/service.py

async def extract(
    self,
    scope: TenantScope,
    interaction: dict[str, Any],
    llm: Any,
    *,
    callback_manager: CallbackManager | None = None,
    run_id: str = "",
    model_name: str = "consolidation_extractor",
) -> list[MemoryOperation]:
    """Analyse an interaction and return memory operations.

    Args:
        scope: Tenant scope for isolation.
        interaction: Dict with 'user_message' and 'assistant_response'.
        llm: LLM instance (BaseChatModel from langchain-core).
        callback_manager: v7.4.3 (Jarvio Ask 5) -- when provided, fires
            ``on_llm_end(node_name="consolidation_extractor")`` after
            the LLM call so the (opt-in) consolidation extractor is
            visible to token billing / OTel.  ``None`` preserves
            pre-7.4.3 behaviour.
        run_id: Engine run identifier for callback correlation.
        model_name: Optional model identifier stamped on the
            ``LLMEndEvent``.  Defaults to ``"consolidation_extractor"``.

    Returns:
        List of MemoryOperations to be applied via commit_pending.
    """
    prompt = (
        "Analyze this interaction and extract memory operations.\n\n"
        f"User: {interaction.get('user_message', '')}\n"
        f"Assistant: {interaction.get('assistant_response', '')}\n\n"
        "Return a JSON object with an 'ops' list. Each op has: "
        "action (upsert_node/create_edge/store_skill/set_trigger/create_event), "
        "layer (semantic/episodic/procedural/prospective), "
        "label, properties, importance (1-10).\n"
        'Return {"ops": []} if no memories worth extracting.'
    )

    try:
        from symfonic.core.callbacks.emit import (
            _extract_usage,
            emit_llm_end_from_result,
            emit_llm_start,
            llm_timing,
            resolve_model_name,
        )

        _resolved_model = resolve_model_name(llm, model_name)

        # v8.3.2 (G4): emit on the always-on ObservabilityHook BEFORE
        # the call (mirrors react.py:1078 / metacognitive.py:412).
        try:
            await self._observability_hook.on_llm_start(
                _resolved_model, [prompt], run_id
            )
        except Exception:
            logger.exception("ObservabilityHook.on_llm_start failed")

        # v8.4.0 CALLBACK-PATH fix: open the OTel span on the merged
        # manager BEFORE the ainvoke so the existing END (:124) closes
        # it.  Same node_name; byte-identical no-op when no manager.
        if callback_manager is not None and not callback_manager.is_noop:
            await emit_llm_start(
                callback_manager,
                model=_resolved_model,
                messages=[prompt],
                run_id=run_id,
                node_name="consolidation_extractor",
            )

        # v7.15.0: capture timing so LLMEndEvent carries duration.
        async with llm_timing() as _timing:
            response = await llm.ainvoke(prompt)

        # v8.3.2 (G4): emit on the always-on ObservabilityHook AFTER
        # the call (mirrors react.py:1433 / metacognitive.py:447).
        try:
            await self._observability_hook.on_llm_end(
                _resolved_model,
                str(getattr(response, "content", response)),
                _extract_usage(response),
                run_id,
            )
        except Exception:
            logger.exception("ObservabilityHook.on_llm_end failed")

        # v7.4.3 Jarvio Ask 5: emit on_llm_end so this (opt-in)
        # consolidation extractor's tokens show up in billing / OTel.
        # v7.14.0: stamp the resolved SKU instead of the caller-passed
        # name (which defaults to the node-discriminator string and
        # missed MODEL_PRICING).  See callbacks/emit.py:resolve_model_name.
        await emit_llm_end_from_result(
            callback_manager,
            model=_resolved_model,
            result=response,
            run_id=run_id,
            node_name="consolidation_extractor",
            timing=_timing,
        )

        content = extract_llm_content(response)
        return _parse_extraction(content, scope)
    except Exception:
        logger.exception("Memory extraction failed for tenant %s", scope.tenant_id)
        return []