symfonic.capabilities.memory.extraction

extraction

Deliverable 1: turning one finished turn into candidate memories.

The service does four things the shipped extractor spreads across three modules and one orchestrator method, and it reports on all four:

1. Ask. It builds the extraction prompt — the same wire format the shipped extractor uses, so a deployment that swaps implementations mid-migration gets the same answers out of the same model.
2. Read. The reply is read per provider family (:mod:.families) instead of stringified, so an Anthropic block list or an OpenAI envelope is parsed rather than mined.
3. Scrub. Credentials are removed before a record exists (:mod:.scrubbing), in both directions: out of the turn before the model sees it, and out of the extracted memory before the write port does.
4. Refuse, out loud. Every omission carries a reason (:mod:.operations).

The service holds no provider SDK. It takes an :class:ExtractionModelPort — one method, complete(prompt) -> reply — which a composition root binds. Unbound, the service is inert: no model is a configuration, not an error, which is the same property that makes a memoryless agent an ordinary deployment rather than a fork.

Failure never reaches the turn. A model that raises produces a degraded result. Two things are not caught: this capability's own contract errors, which mean the call was impossible rather than the model, and CancelledError, which is re-raised unswallowed (CXL-2) — a cancelled invocation is not a turn that failed to remember.

ExtractionModelPort

Bases: Protocol

The one model call this capability makes.

complete async

complete(prompt: str) -> Any

Return the model's reply, in whatever shape its provider emits.

Source code in src/symfonic/capabilities/memory/extraction.py

async def complete(self, prompt: str) -> Any:
    """Return the model's reply, in whatever shape its provider emits."""
    ...

MemoryExtractionService

MemoryExtractionService(model: ExtractionModelPort | None = None, *, scrubber: CredentialScrubber | None = None)

Extracts candidate memories from a finished turn.

Source code in src/symfonic/capabilities/memory/extraction.py

def __init__(
    self,
    model: ExtractionModelPort | None = None,
    *,
    scrubber: CredentialScrubber | None = None,
) -> None:
    self._model = model
    self._scrubber = scrubber if scrubber is not None else CredentialScrubber()

extract async

extract(request: ExtractionRequest) -> ExtractionResult

Ask the model, read the reply, and mint what survives the policy.

Source code in src/symfonic/capabilities/memory/extraction.py

async def extract(self, request: ExtractionRequest) -> ExtractionResult:
    """Ask the model, read the reply, and mint what survives the policy."""
    prompt, redactions = self.prompt(request)
    if self._model is None:
        return ExtractionResult(
            scope=request.scope,
            turn=request.turn,
            redactions=redactions,
            reason="no extraction model is bound; this deployment does not extract",
        )
    try:
        reply = read_reply(await self._model.complete(prompt))
    except MemoryCapabilityError:
        raise
    except Exception as exc:  # noqa: BLE001 - a failed extraction is not a failed turn
        return ExtractionResult(
            scope=request.scope,
            turn=request.turn,
            redactions=redactions,
            degraded=True,
            reason=f"extraction model failed: {type(exc).__name__}",
        )
    return self.parse(reply, request, redactions=redactions)

parse

parse(reply: ProviderReply, request: ExtractionRequest, *, redactions: tuple[str, ...] = ()) -> ExtractionResult

Turn a read reply into records. Pure, so a corpus can replay it.

Source code in src/symfonic/capabilities/memory/extraction.py

def parse(
    self,
    reply: ProviderReply,
    request: ExtractionRequest,
    *,
    redactions: tuple[str, ...] = (),
) -> ExtractionResult:
    """Turn a read reply into records. Pure, so a corpus can replay it."""
    if not reply.readable:
        return ExtractionResult(
            scope=request.scope,
            turn=request.turn,
            family=reply.family,
            redactions=redactions,
            reason=(
                "the model reply is unreadable: no known provider family "
                "matched its shape, and stringifying it would mine the transport"
            ),
        )
    payload = json_payload(reply.text)
    if payload is None:
        return ExtractionResult(
            scope=request.scope,
            turn=request.turn,
            family=reply.family,
            redactions=redactions,
            reason="the model reply carried no JSON object",
        )
    return self._mint(payload, request, reply.family, redactions)

prompt

prompt(request: ExtractionRequest) -> tuple[str, tuple[str, ...]]

The extraction prompt for request, with credentials removed.

Scrubbing here is not belt-and-braces for the store: a secret handed to a provider is disclosed whether or not anyone ever writes it down.

Source code in src/symfonic/capabilities/memory/extraction.py

def prompt(self, request: ExtractionRequest) -> tuple[str, tuple[str, ...]]:
    """The extraction prompt for ``request``, with credentials removed.

    Scrubbing here is not belt-and-braces for the store: a secret handed to
    a provider is disclosed whether or not anyone ever writes it down.
    """
    user = self._scrubber.scrub_text(request.user_message)
    assistant = self._scrubber.scrub_text(request.assistant_message)
    text = EXTRACTION_PROMPT.format(
        user_message=user.text, assistant_message=assistant.text
    )
    return text, user.redactions + assistant.redactions