Skip to content

symfonic.capabilities.knowledge.citations

citations

Internal layer tags: what the framework strips before a reader sees them.

: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.

Grounded source references live next door in :mod:.grounding: the two filters answer different questions, so they live in different modules.

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 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
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 

feed(text: str) -> str

Append text and return the prefix that is safe to emit.

Source code in src/symfonic/capabilities/knowledge/citations.py 
74
75
76
77
78
79
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 

flush() -> str

Drain the buffer at stream end, dropping any unfinished tag prefix.

Source code in src/symfonic/capabilities/knowledge/citations.py 
81
82
83
def flush(self) -> str:
    """Drain the buffer at stream end, dropping any unfinished tag prefix."""
    return self._drain(final=True)