Skip to content

symfonic.capabilities.knowledge.grounding

grounding

Grounded source references: what the framework lets out as evidence.

A citation is a claim that a specific retrieved document says something; a citation naming a document that was never retrieved is a fabrication wearing the costume of evidence, and shipping it is worse than shipping no citation at all. :func:drop_ungrounded_citations removes those references while grounded ones pass through byte-identical.

Split out of :mod:.citations: the tag filter and the grounding drop answer different questions -- one strips internal layer annotations by vocabulary, the other checks model-emitted references against a retrieval -- and each fits alone in a module with room to grow.

Citation dataclass 

Citation(marker: str, source: str)

One numbered, display-safe reference to a retrieved source.

CitationIndex dataclass 

CitationIndex(citations: tuple[Citation, ...] = ())

The grounded set: every source a retrieval actually returned.

from_fragments classmethod 

from_fragments(fragments: Iterable[RetrievedFragment]) -> CitationIndex

Number the distinct sources of fragments in ranked order.

Ranked, not received: the numbering has to be a function of the content so two runs over the same retrieval produce the same reference list.

Source code in src/symfonic/capabilities/knowledge/grounding.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@classmethod
def from_fragments(cls, fragments: Iterable[RetrievedFragment]) -> CitationIndex:
    """Number the distinct sources of ``fragments`` in ranked order.

    Ranked, not received: the numbering has to be a function of the content
    so two runs over the same retrieval produce the same reference list.
    """
    ranked = sorted(fragments, key=lambda f: (-f.score, f.source, f.content))
    seen: list[str] = []
    for fragment in ranked:
        label = safe_source(fragment.source)
        if label and label not in seen:
            seen.append(label)
    return cls(
        tuple(Citation(marker=str(i), source=s) for i, s in enumerate(seen, start=1))
    )

render 

render() -> str

The reference list, one citation per line.

Source code in src/symfonic/capabilities/knowledge/grounding.py 
72
73
74
def render(self) -> str:
    """The reference list, one citation per line."""
    return "\n".join(f"[{c.marker}] {c.source}" for c in self.citations)

sources 

sources() -> frozenset[str]

The grounded source labels a reference may name.

Source code in src/symfonic/capabilities/knowledge/grounding.py 
68
69
70
def sources(self) -> frozenset[str]:
    """The grounded source labels a reference may name."""
    return frozenset(citation.source for citation in self.citations)

drop_ungrounded_citations 

drop_ungrounded_citations(text: str, index: CitationIndex) -> tuple[str, tuple[str, ...]]

Remove [[source:X]] references whose X was never retrieved.

Returns the rewritten text and the ungrounded names, each reported once in first-appearance order. When nothing was ungrounded the text is returned byte-identical — a filter that reflowed whitespace on every clean response would make its own no-op observable.

Source code in src/symfonic/capabilities/knowledge/grounding.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def drop_ungrounded_citations(text: str, index: CitationIndex) -> tuple[str, tuple[str, ...]]:
    """Remove ``[[source:X]]`` references whose ``X`` was never retrieved.

    Returns the rewritten text and the ungrounded names, each reported once in
    first-appearance order. When nothing was ungrounded the text is returned
    byte-identical — a filter that reflowed whitespace on every clean response
    would make its own no-op observable.
    """
    grounded = index.sources()
    dropped: list[str] = []

    def replace(match: re.Match[str]) -> str:
        name = match.group(1).strip()
        if safe_source(name) in grounded:
            return match.group(0)
        if name not in dropped:
            dropped.append(name)
        return ""

    rewritten = SOURCE_REFERENCE.sub(replace, text)
    if not dropped:
        return text, ()
    return _SPACE_RUN.sub(" ", rewritten).strip(), tuple(dropped)