symfonic.agent.prompts

prompts

symfonic.agent.prompts -- PromptSection implementations.

Public API::

from symfonic.agent.prompts import (
    BRAIN_PROMPT_VERSION,
    DBPromptSection,
    HMSSystemPromptSection,
    MemoryExtractionSection,
)

DBPromptSection dataclass

DBPromptSection(
    name: str,
    required: bool = False,
    priority: int = 50,
    cache_breakpoint: bool = False,
    fallback_content: str = "",
    section_key: str = "",
    cache_ttl_seconds: int = 300,
)

PromptSection that loads versioned Jinja2 templates from PostgreSQL.

The state dict is passed as Jinja2 template context, enabling: - {{ name }} variable substitution - {% if user_is_premium %}...{% endif %} conditionals - {% for item in items %}...{% endfor %} loops - {{ name | upper }} filters

Gracefully degrades when Redis or DB are unavailable, falling back to fallback_content rendered through the same Jinja2 pipeline.

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.
fallback_content	str	Jinja2 template string used when DB/cache are unavailable.
section_key	str	DB lookup key; defaults to name if empty.
cache_ttl_seconds	int	Redis TTL for cached template content (default 300s).

configure

configure(
    redis_client: Any = None, db_session_factory: Any = None
) -> DBPromptSection

Inject optional Redis and DB dependencies. Returns self for chaining.

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

def configure(
    self,
    redis_client: Any = None,
    db_session_factory: Any = None,
) -> DBPromptSection:
    """Inject optional Redis and DB dependencies. Returns self for chaining."""
    self._redis_client = redis_client
    self._db_session_factory = db_session_factory
    return self

render async

render(state: PromptState) -> str

Render the section: cache -> DB -> fallback, all through Jinja2.

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

async def render(self, state: PromptState) -> str:
    """Render the section: cache -> DB -> fallback, all through Jinja2."""
    key = self.section_key or self.name
    raw_template: str | None = None

    # 1. Try Redis cache
    raw_template = await self._try_cache(key)

    # 2. Try DB
    if raw_template is None:
        raw_template = await self._try_db(key)
        if raw_template is not None:
            await self._set_cache(key, raw_template)

    # 3. Fallback
    if raw_template is None:
        raw_template = self.fallback_content

    if not raw_template:
        return ""

    return self._render_jinja2(raw_template, dict(state))

HMSSystemPromptSection dataclass

HMSSystemPromptSection(
    name: str = "HMS System Prompt",
    required: bool = False,
    priority: int = 5,
    cache_breakpoint: bool = True,
    template_path: Path | None = None,
)

PromptSection that renders the HMS-aware system prompt template.

Substitutes all {{VARIABLE}} placeholders from the PromptState context. Raises ValueError if any placeholder remains unresolved after rendering.

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 (file or directory) for the hms_system.txt template. Defaults to the bundled template.

estimate_tokens staticmethod

estimate_tokens(text: str) -> int

Estimate token count using a conservative 3-chars-per-token heuristic.

The len // 4 heuristic undercounts by 20-30 % on structured / markdown text. len // 3 is closer to observed tokeniser output for prompts that contain headings, bullet lists, and JSON blocks.

Parameters:

Name	Type	Description	Default
text	str	The text to estimate token count for.	required

Returns:

Type	Description
int	Estimated number of tokens (conservative upper bound).

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

@staticmethod
def estimate_tokens(text: str) -> int:
    """Estimate token count using a conservative 3-chars-per-token heuristic.

    The ``len // 4`` heuristic undercounts by 20-30 % on structured /
    markdown text.  ``len // 3`` is closer to observed tokeniser output
    for prompts that contain headings, bullet lists, and JSON blocks.

    Args:
        text: The text to estimate token count for.

    Returns:
        Estimated number of tokens (conservative upper bound).
    """
    return len(text) // 3

render async

render(state: PromptState) -> str

Render the HMS system prompt by substituting placeholders.

Parameters:

Name	Type	Description	Default
state	PromptState	Mapping that must contain values for every {{VARIABLE}} placeholder in the template.	required

Returns:

Type	Description
str	The fully substituted prompt text.

Raises:

Type	Description
SecurityScopeError	If TENANT_ID is missing or empty.
ValueError	If any {{VARIABLE}} placeholder remains after substitution (i.e. the state is missing required keys).

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

async def render(self, state: PromptState) -> str:
    """Render the HMS system prompt by substituting placeholders.

    Args:
        state: Mapping that must contain values for every
            ``{{VARIABLE}}`` placeholder in the template.

    Returns:
        The fully substituted prompt text.

    Raises:
        SecurityScopeError: If ``TENANT_ID`` is missing or empty.
        ValueError: If any ``{{VARIABLE}}`` placeholder remains after
            substitution (i.e. the state is missing required keys).
    """
    tenant_id: Any = state.get("TENANT_ID") or state.get("tenant_id")
    if not tenant_id or not str(tenant_id).strip():
        raise SecurityScopeError(
            "tenant_id is required for HMS prompt rendering"
        )

    template = self._get_template()
    rendered = _substitute(template, state)
    remaining = _PLACEHOLDER_RE.findall(rendered)
    if remaining:
        raise ValueError(
            f"Unresolved HMS system prompt placeholders: {remaining}"
        )
    return rendered

validate_budget

validate_budget(
    rendered: str, max_tokens: int = 1000
) -> tuple[bool, int]

Check whether rendered fits within the token budget.

Does NOT raise — callers decide how to handle over-budget results.

Parameters:

Name	Type	Description	Default
rendered	str	The already-rendered prompt text.	required
max_tokens	int	Maximum allowed token count.	1000

Returns:

Type	Description
tuple[bool, int]	Tuple of (within_budget, estimated_tokens).

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

def validate_budget(
    self, rendered: str, max_tokens: int = 1000
) -> tuple[bool, int]:
    """Check whether *rendered* fits within the token budget.

    Does NOT raise — callers decide how to handle over-budget results.

    Args:
        rendered: The already-rendered prompt text.
        max_tokens: Maximum allowed token count.

    Returns:
        Tuple of ``(within_budget, estimated_tokens)``.
    """
    estimated = self.estimate_tokens(rendered)
    return estimated <= max_tokens, estimated

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