The citation and output bridge: what the framework lets out.
Two independent filters, deliberately not merged. They answer different
questions and a caller may want one without the other.
:class:CitationTagFilter removes the internal layer annotations
([semantic], [episodic], …) a model inlines when narrating which memory
layer a fact came from. Those drive analytics and UI badges; a reader must
never see them. A per-chunk regex cannot do this because a tag straddles chunk
boundaries, so the filter holds back only the tail that could still become a
tag and emits everything else immediately.
:func:drop_ungrounded_citations removes source references the model invented.
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.
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/citations.py
| @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
The reference list, one citation per line.
Source code in src/symfonic/capabilities/knowledge/citations.py
| 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/citations.py
| def sources(self) -> frozenset[str]:
"""The grounded source labels a reference may name."""
return frozenset(citation.source for citation in self.citations)
|
CitationTagFilter
CitationTagFilter(layer_names: Sequence[str] = DEFAULT_LAYER_NAMES)
Streaming filter that strips internal layer tags across chunk boundaries.
Feed chunks in order and concatenate what comes back; call :meth:flush
once at stream end. A partial prefix that never completes is dropped at
flush rather than emitted: it was only held back because it matched a known
layer name, so emitting it would leak a fragment of an internal annotation.
Source code in src/symfonic/capabilities/knowledge/citations.py
| def __init__(self, layer_names: Sequence[str] = DEFAULT_LAYER_NAMES) -> None:
if not layer_names:
raise ValueError(
"a citation filter needs at least one layer name; an empty vocabulary "
"is a filter that strips nothing, which is better written as no filter."
)
self._tags = tuple(f"[{name.lower()}]" for name in layer_names)
# T3.2.4 finding 3: horizontal whitespace only, never ``\s*``.
#
# ``\s*`` is greedy over newlines too, so a tag on its own indented line
# took the preceding line break with it and welded two lines together:
# ``"code:\n [working] indented"`` became ``"code: indented"``.
# That was the most destructive of the three answers one input could
# produce, and it is a *whole-buffer* defect — under chunked delivery
# the newline had already been emitted and survived, which is why it is
# a different finding from the chunk-invariance one and not the same one
# seen twice.
self._tag_re = re.compile(
r"[ \t]*\[(?:" + "|".join(re.escape(name) for name in layer_names) + r")\]",
re.IGNORECASE,
)
self._buffer = ""
self._last_emitted_was_space = False
self._pending_seam = False
|
feed
Append text and return the prefix that is safe to emit.
Source code in src/symfonic/capabilities/knowledge/citations.py
| def feed(self, text: str) -> str:
"""Append ``text`` and return the prefix that is safe to emit."""
if not text:
return ""
self._buffer += text
return self._drain(final=False)
|
flush
Drain the buffer at stream end, dropping any unfinished tag prefix.
Source code in src/symfonic/capabilities/knowledge/citations.py
| def flush(self) -> str:
"""Drain the buffer at stream end, dropping any unfinished tag prefix."""
return self._drain(final=True)
|
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/citations.py
| 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)
|