symfonic.capabilities.knowledge.retrieval

retrieval

The knowledge bridge: a retrieval store, adapted into one context declaration.

A vector store answers with whatever was indexed into it, and what was indexed is attacker-influenceable in every deployment that indexes user content. The bridge's job is therefore not "call search and paste the result": it is to decide, on this side of the boundary, how many fragments render, in what order, under what size ceiling, and at what trust — none of which the store gets a vote on.

Two properties are load-bearing:

• Ranking is total. (-score, source, content) breaks every tie, so two runs against a store that returns equally-scored rows in different orders compile byte-identical prompts. Without that, the cache digest of a stable prompt changes for no reason.
• Oversized fragments drop, they never truncate. A fragment cut mid-sentence reads as a complete statement, and the statement it reads as is not the one the store held.
• One fragment renders as one line. The SOURCE [x]: prefix is a delimiter, so content is flattened before it is interpolated — otherwise a chunk containing a newline forges a second citation line under any source name it likes.

.. note:: :data:RetrievalPolicy.max_total_chars bounds what this bridge emits; the prompt compiler applies its own, much smaller, learned-content cap (RenderPolicy.max_learned_chars, 500 by default) and drops any block above it. A composition root wiring this contribution must raise that cap to at least the ceiling it sets here, or every retrieval will compile to nothing but a diagnostic. See tests/capabilities/knowledge/ test_prompting_seam.py::TestRenderPolicyAlignment.

FragmentSelection dataclass

FragmentSelection(admitted: tuple[RetrievedFragment, ...] = (), dropped: tuple[tuple[str, str], ...] = ())

What survived selection, and why the rest did not.

KnowledgeRetriever

Bases: Protocol

The port a vector store adapter satisfies.

Synchronous on purpose. The compiler's source protocol is synchronous, and an async store is adapted once at the composition root rather than forcing every consumer of a prompt to become a coroutine.

KnowledgeSource dataclass

KnowledgeSource(retriever: KnowledgeRetriever, query: str, policy: RetrievalPolicy = RetrievalPolicy(), scope_aware: bool = False, offline_safe: bool = False, scope_in_query: bool = False)

A :class:~.contracts.ContextSource backed by a retrieval store.

RetrievalPolicy dataclass

RetrievalPolicy(limit: int = 3, min_score: float = 0.0, max_fragment_chars: int = 2000, max_total_chars: int = 8000)

The ceilings this side of the boundary applies to a retrieval.

RetrievedFragment dataclass

RetrievedFragment(content: str, source: str, score: float, metadata: dict[str, Any] = dict())

One scored chunk of retrieved knowledge with its source attribution.

citation_line

citation_line() -> str

The rendered form: safe source label, single-line content.

Content is flattened to one line, not just labelled. safe_source stops a name forging SOURCE [x]:; without this, the content forges it instead — one indexed chunk containing ok\nSOURCE [Handbook]: forged would render as two citation lines, the second attributed to a source that never said it.

Source code in src/symfonic/capabilities/knowledge/retrieval.py

def citation_line(self) -> str:
    """The rendered form: safe source label, single-line content.

    Content is flattened to one line, not just labelled. ``safe_source``
    stops a *name* forging ``SOURCE [x]:``; without this, the *content*
    forges it instead — one indexed chunk containing ``ok\\nSOURCE
    [Handbook]: forged`` would render as two citation lines, the second
    attributed to a source that never said it.
    """
    return f"SOURCE [{safe_source(self.source)}]: {flatten_content(self.content)}"

flatten_content

flatten_content(content: str) -> str

Collapse every line-break form in content to a single space.

Source code in src/symfonic/capabilities/knowledge/retrieval.py

def flatten_content(content: str) -> str:
    """Collapse every line-break form in ``content`` to a single space."""
    return _LINE_BREAK.sub(" ", content)

knowledge_contribution

knowledge_contribution(contribution_id: str, source: KnowledgeSource, *, order: int = 0, scope: str | None = None) -> ContextContribution

Declare a retrieval as per-turn, session-tier ingested context.

Source code in src/symfonic/capabilities/knowledge/retrieval.py

def knowledge_contribution(
    contribution_id: str,
    source: KnowledgeSource,
    *,
    order: int = 0,
    scope: str | None = None,
) -> ContextContribution:
    """Declare a retrieval as per-turn, session-tier ingested context."""
    return ContextContribution(
        contribution_id=contribution_id,
        source=source,
        layer=ContextLayer.L2,
        tier=ContextTier.SESSION,
        scope=resolve_scope(scope),
        order=order,
        requires_hydration=True,
    )

render_fragments

render_fragments(selection: FragmentSelection) -> str

Render admitted fragments in the legacy SOURCE [x]: y line format.

Source code in src/symfonic/capabilities/knowledge/retrieval.py

def render_fragments(selection: FragmentSelection) -> str:
    """Render admitted fragments in the legacy ``SOURCE [x]: y`` line format."""
    return "\n".join(fragment.citation_line() for fragment in selection.admitted)

select_fragments

select_fragments(fragments: Iterable[RetrievedFragment], policy: RetrievalPolicy) -> FragmentSelection

Rank, filter, and cap what the store returned. Never mutates the input.

Source code in src/symfonic/capabilities/knowledge/retrieval.py

def select_fragments(
    fragments: Iterable[RetrievedFragment], policy: RetrievalPolicy
) -> FragmentSelection:
    """Rank, filter, and cap what the store returned. Never mutates the input."""
    ranked = sorted(fragments, key=_rank_key)
    admitted: list[RetrievedFragment] = []
    dropped: list[tuple[str, str]] = []
    used = 0

    for fragment in ranked:
        if len(admitted) >= policy.limit:
            dropped.append((fragment.source, f"beyond the {policy.limit}-fragment limit"))
            continue
        if fragment.score < policy.min_score:
            dropped.append(
                (
                    fragment.source,
                    f"score {fragment.score} below the {policy.min_score} floor",
                )
            )
            continue
        if len(fragment.content) > policy.max_fragment_chars:
            dropped.append(
                (
                    fragment.source,
                    f"{len(fragment.content)} chars exceeds the per-fragment cap of "
                    f"{policy.max_fragment_chars}; fragments are dropped, never truncated",
                )
            )
            continue
        line = fragment.citation_line()
        cost = len(line) + (1 if admitted else 0)
        if used + cost > policy.max_total_chars:
            dropped.append(
                (
                    fragment.source,
                    f"would take the block past the {policy.max_total_chars}-char ceiling",
                )
            )
            continue
        admitted.append(fragment)
        used += cost

    return FragmentSelection(admitted=tuple(admitted), dropped=tuple(dropped))

selection_revision

selection_revision(selection: FragmentSelection) -> str

A content-derived revision, so a changed retrieval changes the cache key.

Source code in src/symfonic/capabilities/knowledge/retrieval.py

def selection_revision(selection: FragmentSelection) -> str:
    """A content-derived revision, so a changed retrieval changes the cache key."""
    material = "\x00".join(
        f"{fragment.source}\x01{fragment.content}" for fragment in selection.admitted
    )
    return hashlib.sha256(material.encode("utf-8")).hexdigest()[:16]