symfonic.agent.prompts.extraction_prompt

extraction_prompt

MemoryExtractionSection -- post-response extraction directive prompt.

Renders the bundled extraction.txt template, substituting the {{TENANT_ID}} placeholder. Conforms to the PromptSection protocol.

MemoryExtractionSection dataclass

MemoryExtractionSection(
    name: str = "Memory Extraction Directives",
    required: bool = False,
    priority: int = 90,
    cache_breakpoint: bool = False,
    template_path: Path | None = None,
    template_name: str = _EXTRACTION_TEMPLATE,
)

PromptSection that renders the memory extraction directives.

Substitutes {{TENANT_ID}} from the PromptState context. Used as a post-response instruction to guide structured memory mutations.

Attributes:

Name	Type	Description
name	str	Section display name used by PromptBuilder.
required	bool	Whether a render failure should abort the build.
priority	int	Lower numbers appear earlier in the assembled prompt.
template_path	Path | None	Optional override path for extraction.txt.
template_name	str	Bundled template to load when template_path is unset (v9.2.0 #25.4c): "extraction" (default, Anthropic delimiter) or "extraction_openai" (explicit directive for OpenAI-family models). Ignored when template_path points at a file.

render async

render(state: PromptState) -> str

Render the extraction directives with state substitution.

Parameters:

Name	Type	Description	Default
state	PromptState	Must contain tenant_id or TENANT_ID. May also contain DOMAIN_NAME, REQUIRED_LABELS, ONBOARDING_CHECKLIST, SOUL_SCHEMA, and known_entities (a list of {id, label, type} dicts) from the domain template / hydration pipeline.	required

Returns:

Type	Description
str	Rendered extraction directive string, or empty string if no
str	tenant is present in state. Byte-for-byte compatible with
str	pre-7.4 behaviour: the KNOWN ENTITIES block is interleaved
str	at the {{KNOWN_ENTITIES_SECTION}} position in the
str	template, exactly where pre-split callers expected it.

Source code in src/symfonic/agent/prompts/extraction_prompt.py

async def render(self, state: PromptState) -> str:
    """Render the extraction directives with state substitution.

    Args:
        state: Must contain ``tenant_id`` or ``TENANT_ID``.
            May also contain ``DOMAIN_NAME``, ``REQUIRED_LABELS``,
            ``ONBOARDING_CHECKLIST``, ``SOUL_SCHEMA``, and
            ``known_entities`` (a list of ``{id, label, type}``
            dicts) from the domain template / hydration pipeline.

    Returns:
        Rendered extraction directive string, or empty string if no
        tenant is present in state.  Byte-for-byte compatible with
        pre-7.4 behaviour: the KNOWN ENTITIES block is interleaved
        at the ``{{KNOWN_ENTITIES_SECTION}}`` position in the
        template, exactly where pre-split callers expected it.
    """
    stable_part, volatile_part = await self.render_split(
        state, _legacy=True,
    )
    if not stable_part and not volatile_part:
        return ""
    # Legacy mode: the volatile block was interleaved at the
    # ``{{KNOWN_ENTITIES_SECTION}}`` placeholder position. The
    # ``_legacy=True`` path above performs that interleaving inline
    # and returns the joined string in ``stable_part``, leaving
    # ``volatile_part`` empty -- so the legacy public output is a
    # single byte-for-byte identical string.
    return stable_part

render_split async

render_split(
    state: PromptState, *, _legacy: bool = False
) -> tuple[str, str]

Render the extraction directives as a stable/volatile pair.

v7.4 Item 15: the extraction protocol's stable surface (schema, label rules, few-shot example, domain directives) lives in L1 (cached prefix) while the per-turn volatile KNOWN ENTITIES block lives in L2. This method returns the two strings separately so the engine can route them to different cache layers.

Parameters:

Name	Type	Description	Default
state	PromptState	Same contract as :meth:render.	required
_legacy	bool	INTERNAL. When True, the volatile KNOWN ENTITIES block is interleaved into the stable string at the {{KNOWN_ENTITIES_SECTION}} placeholder position, exactly matching pre-split byte output for legacy callers. volatile_part is then returned as "". Public callers should NOT set this -- use the two-tuple shape and route each part independently.	False

Returns:

Type	Description
str	(stable_part, volatile_part). Either or both may be
str	empty. When no tenant is in state, both are empty.

Source code in src/symfonic/agent/prompts/extraction_prompt.py

async def render_split(
    self,
    state: PromptState,
    *,
    _legacy: bool = False,
) -> tuple[str, str]:
    """Render the extraction directives as a stable/volatile pair.

    v7.4 Item 15: the extraction protocol's stable surface (schema,
    label rules, few-shot example, domain directives) lives in L1
    (cached prefix) while the per-turn volatile KNOWN ENTITIES
    block lives in L2.  This method returns the two strings
    separately so the engine can route them to different cache
    layers.

    Args:
        state: Same contract as :meth:`render`.
        _legacy: INTERNAL.  When ``True``, the volatile KNOWN
            ENTITIES block is interleaved into the stable string at
            the ``{{KNOWN_ENTITIES_SECTION}}`` placeholder
            position, exactly matching pre-split byte output for
            legacy callers.  ``volatile_part`` is then returned as
            ``""``.  Public callers should NOT set this -- use the
            two-tuple shape and route each part independently.

    Returns:
        ``(stable_part, volatile_part)``.  Either or both may be
        empty.  When no tenant is in state, both are empty.
    """
    tenant_id: str | None = state.get("TENANT_ID") or state.get("tenant_id")  # type: ignore[assignment]
    if not tenant_id:
        return "", ""

    template = self._get_template()
    rendered = template.replace("{{TENANT_ID}}", str(tenant_id))

    # Render the optional KNOWN ENTITIES directive.  An empty list
    # collapses to "" so we don't emit a header with no body.
    known_entities = state.get("known_entities") or state.get("KNOWN_ENTITIES")  # type: ignore[assignment]
    known_section = _render_known_entities_section(
        known_entities if isinstance(known_entities, list) else None,
    )

    if _legacy:
        # Interleave at the placeholder position so legacy callers
        # see the pre-7.4 byte layout.
        rendered = rendered.replace(
            "{{KNOWN_ENTITIES_SECTION}}", known_section,
        )
        volatile_part = ""
    else:
        # Stable part: drop the placeholder entirely so the L1 body
        # contains zero per-turn content.  Volatile part: the
        # rendered KNOWN ENTITIES directive (with its leading
        # newline preserved so it can be appended cleanly after
        # other L2 content).
        rendered = rendered.replace("{{KNOWN_ENTITIES_SECTION}}", "")
        volatile_part = known_section.lstrip("\n") if known_section else ""

    # Substitute domain-specific placeholders with defaults.
    _domain_defaults: dict[str, str] = {
        "DOMAIN_NAME": "default",
        "REQUIRED_LABELS": "(none)",
        "ONBOARDING_CHECKLIST": "  (none)",
        "SOUL_SCHEMA": "  (none)",
    }
    for key, default in _domain_defaults.items():
        value = str(state.get(key, default))  # type: ignore[arg-type]
        rendered = rendered.replace(f"{{{{{key}}}}}", value)

    # Validate no other placeholders remain in the stable body.
    remaining = _PLACEHOLDER_RE.findall(rendered)
    if remaining:
        raise ValueError(
            f"Unresolved extraction prompt placeholders: {remaining}"
        )
    return rendered, volatile_part