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)

parse_inline

parse_inline(operations: tuple[Mapping[str, Any], ...], request: ExtractionRequest) -> ExtractionResult

Mint operations emitted by the response model's directive.

The response adapter has already recognised and stripped the envelope, so this method receives only JSON objects. It deliberately shares the same minting path as a dedicated extraction model: inline operations get the same validation, credential scrubbing, write receipt and lifecycle publication rather than becoming an unreviewed side path.

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

def parse_inline(
    self, operations: tuple[Mapping[str, Any], ...], request: ExtractionRequest
) -> ExtractionResult:
    """Mint operations emitted by the response model's directive.

    The response adapter has already recognised and stripped the envelope,
    so this method receives only JSON objects.  It deliberately shares the
    same minting path as a dedicated extraction model: inline operations
    get the same validation, credential scrubbing, write receipt and
    lifecycle publication rather than becoming an unreviewed side path.
    """
    legacy = tuple(
        (f"op[{index}]", "the tagged operation uses the legacy type/payload schema")
        for index, operation in enumerate(operations)
        if "type" in operation and "action" not in operation
    )
    if legacy:
        return ExtractionResult(
            scope=request.scope,
            turn=request.turn,
            dropped=legacy,
            family=ProviderFamily.TEXT,
            reason="the inline extraction schema is not supported by this service",
        )
    return self._mint({"ops": list(operations)}, request, ProviderFamily.TEXT, ())

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