symfonic.core.prompt.builder

builder

PromptBuilder -- ordered section pipeline for system prompt construction.

PromptBuilder

PromptBuilder()

Ordered pipeline that assembles system prompt from sections.

Source code in src/symfonic/core/prompt/builder.py

def __init__(self) -> None:
    self._sections: list[PromptSection] = []

add_section

add_section(section: PromptSection) -> PromptBuilder

Add a section. Returns self for chaining.

Source code in src/symfonic/core/prompt/builder.py

def add_section(self, section: PromptSection) -> PromptBuilder:
    """Add a section. Returns self for chaining."""
    self._sections.append(section)
    return self

build async

build(state: PromptState) -> BuildResult

Render all sections in priority order, collecting warnings.

Source code in src/symfonic/core/prompt/builder.py

async def build(self, state: PromptState) -> BuildResult:
    """Render all sections in priority order, collecting warnings."""
    sorted_sections = sorted(self._sections, key=lambda s: s.priority)
    parts: list[str] = []
    warnings: list[BuildWarning] = []
    degraded: list[str] = []

    for section in sorted_sections:
        try:
            content = await section.render(state)
            if content:
                parts.append(f"## {section.name}\n{content}")
        except Exception as e:
            warning = BuildWarning(
                section=section.name,
                type="critical_error" if section.required else "degraded",
                message=str(e),
            )
            warnings.append(warning)
            degraded.append(section.name)
            if section.required:
                return BuildResult(
                    content="",
                    status="failed",
                    warnings=tuple(warnings),
                    degraded_sections=tuple(degraded),
                )

    status = "partial" if degraded else "success"
    return BuildResult(
        content="\n\n".join(parts),
        status=status,
        warnings=tuple(warnings),
        degraded_sections=tuple(degraded),
    )

build_cached async

build_cached(
    state: PromptState, *, force_last_cache: bool = True
) -> BuildResult

Build prompt with Anthropic cache_control annotations.

Sections with cache_breakpoint=True get {"type": "ephemeral"} cache_control. The first rendered section is always marked for caching. By default the last rendered section is also marked for caching, matching the v6.x behaviour callers depend on.

Pass force_last_cache=False when the trailing section is per-turn user content (e.g. the L2 task layer in the v7.x stratigraphic stack: hydrated memory, prospective tasks, query context). Marking that block cached every turn forces a cache_creation write on every request, which is priced at 1.25x input cost on Anthropic — strictly worse than not caching when the content has cardinality of one per turn. With force_last_cache=False only the FIRST section and any section that explicitly declares cache_breakpoint=True carry a cache_control annotation; the last block is left bare so Anthropic stops the cache prefix at the previous breakpoint.

v7.24.0 cache_ttl interaction (load-bearing for adopters): force_last_cache=True auto-promotes the trailing block to a cache breakpoint. When the trailing section ALSO carries cache_ttl="1h" (via the v7.24.0 §2 plugin knob or §5 stratigraphic propagation), the cold-write premium on every uncached suffix is 2x input rate -- vs the 1.25x premium at the 5m default. Adopters who flip force_last_cache=True on a per-turn-mutating section while opting into 1h pay the 2x premium on every turn for content that doesn't survive the cache anyway. Usually unintended -- prefer force_last_cache=False for per-turn content, OR keep cache_ttl=None (5m default) on sections that mutate frequently. See docs/concepts/cache-tier-tradeoffs.md for the full trade-off analysis.

The BuildResult content field contains a JSON-encoded list of Anthropic content blocks so downstream callers can forward it directly to the API.

Existing build() behaviour is completely unchanged.

Parameters:

Name	Type	Description	Default
state	PromptState	Render-state mapping forwarded to each section.	required
force_last_cache	bool	When True (default) the last rendered section is annotated cached regardless of its cache_breakpoint attribute. When False the last section is annotated cached ONLY if it explicitly declares cache_breakpoint=True. The first section is always cached either way.	True

Source code in src/symfonic/core/prompt/builder.py

async def build_cached(
    self,
    state: PromptState,
    *,
    force_last_cache: bool = True,
) -> BuildResult:
    """Build prompt with Anthropic cache_control annotations.

    Sections with cache_breakpoint=True get ``{"type": "ephemeral"}``
    cache_control.  The first rendered section is always marked for
    caching.  By default the last rendered section is also marked for
    caching, matching the v6.x behaviour callers depend on.

    Pass ``force_last_cache=False`` when the trailing section is
    per-turn user content (e.g. the L2 task layer in the v7.x
    stratigraphic stack: hydrated memory, prospective tasks, query
    context).  Marking that block cached every turn forces a
    ``cache_creation`` write on every request, which is priced at
    1.25x input cost on Anthropic — strictly worse than not caching
    when the content has cardinality of one per turn.  With
    ``force_last_cache=False`` only the FIRST section and any
    section that explicitly declares ``cache_breakpoint=True`` carry
    a ``cache_control`` annotation; the last block is left bare so
    Anthropic stops the cache prefix at the previous breakpoint.

    v7.24.0 cache_ttl interaction (load-bearing for adopters):
    ``force_last_cache=True`` auto-promotes the trailing block to
    a cache breakpoint.  When the trailing section ALSO carries
    ``cache_ttl="1h"`` (via the v7.24.0 §2 plugin knob or §5
    stratigraphic propagation), the cold-write premium on every
    uncached suffix is 2x input rate -- vs the 1.25x premium at
    the 5m default.  Adopters who flip ``force_last_cache=True``
    on a per-turn-mutating section while opting into 1h pay the
    2x premium on every turn for content that doesn't survive
    the cache anyway.  Usually unintended -- prefer
    ``force_last_cache=False`` for per-turn content, OR keep
    ``cache_ttl=None`` (5m default) on sections that mutate
    frequently.  See ``docs/concepts/cache-tier-tradeoffs.md``
    for the full trade-off analysis.

    The BuildResult content field contains a JSON-encoded list of
    Anthropic content blocks so downstream callers can forward it
    directly to the API.

    Existing build() behaviour is completely unchanged.

    Args:
        state: Render-state mapping forwarded to each section.
        force_last_cache: When True (default) the last rendered
            section is annotated cached regardless of its
            ``cache_breakpoint`` attribute.  When False the last
            section is annotated cached ONLY if it explicitly
            declares ``cache_breakpoint=True``.  The first section
            is always cached either way.
    """
    sorted_sections = sorted(self._sections, key=lambda s: s.priority)
    blocks: list[CacheBlock] = []
    warnings: list[BuildWarning] = []
    degraded: list[str] = []

    # Pre-collect rendered (index, text) so we can mark first/last
    rendered: list[tuple[int, str]] = []
    section_map: dict[int, PromptSection] = {}

    for i, section in enumerate(sorted_sections):
        try:
            content = await section.render(state)
            if content:
                rendered.append((i, f"## {section.name}\n{content}"))
                section_map[i] = section
        except Exception as e:
            warning = BuildWarning(
                section=section.name,
                type="critical_error" if section.required else "degraded",
                message=str(e),
            )
            warnings.append(warning)
            degraded.append(section.name)
            if section.required:
                return BuildResult(
                    content="",
                    status="failed",
                    warnings=tuple(warnings),
                    degraded_sections=tuple(degraded),
                )

    last_pos = len(rendered) - 1
    for pos, (orig_idx, section_text) in enumerate(rendered):
        section = section_map[orig_idx]
        should_cache = getattr(section, "cache_breakpoint", False)
        if pos == 0 or (pos == last_pos and force_last_cache):
            should_cache = True
        cache_control = {"type": "ephemeral"} if should_cache else None
        # v7.24.0: per-block cache TTL hint.  Sections expose a
        # ``cache_ttl`` attribute (None default = legacy 5m wire).
        # Adapter sections (e.g. ``PluginPromptSection``) may also
        # expose ``resolve_cache_ttl()`` to bubble up contribution-
        # level hints; that wins over a static ``cache_ttl=None``
        # but is shadowed by an explicit non-None ``cache_ttl`` on
        # the section.  Hint is ignored for uncached blocks.
        cache_ttl = getattr(section, "cache_ttl", None)
        if cache_ttl is None:
            resolver = getattr(section, "resolve_cache_ttl", None)
            if callable(resolver):
                try:
                    cache_ttl = resolver()
                except Exception:
                    cache_ttl = None
        blocks.append(
            CacheBlock(
                text=section_text,
                cache_control=cache_control,
                cache_ttl=cache_ttl if should_cache else None,
            ),
        )

    # v7.24.1 HOTFIX: Anthropic enforces non-increasing TTL across the
    # cumulative prompt prefix.  The auto-promoted first block (above)
    # carries ``cache_ttl=None`` (5m wire) but a downstream cached
    # block may declare ``cache_ttl="1h"`` (e.g. v7.24.0
    # stratigraphic L1 with ``system_prefix_cache_ttl="1h"``).  Left
    # unnormalised this emits ``5m -> 1h`` ASCENDING and Anthropic
    # 400s every brain turn.  Right-to-left propagation promotes
    # earlier cached blocks to the highest tier seen on their right,
    # producing a non-increasing ladder.  No-op (byte-identical to
    # v7.23) when every cached block stays at the 5m default.
    normalised = _normalise_descending_ttl(tuple(blocks))
    cacheable = CacheablePrompt(blocks=normalised)
    content = json.dumps(cacheable.to_anthropic_content())

    status = "partial" if degraded else "success"
    return BuildResult(
        content=content,
        status=status,
        warnings=tuple(warnings),
        degraded_sections=tuple(degraded),
    )

PromptSection

Bases: Protocol

A named section that renders content from state.