File size: 2,075 Bytes
d787a09 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 | """Turn inline [n] markers in an answer into Citation objects.
We only emit citations for markers the model actually used AND that map to a real
retrieved chunk. Markers that point past the context are dropped (defensive
against a model inventing [9] when only 4 passages exist).
"""
from __future__ import annotations
import re
from typing import List
from app.schemas import Citation, RetrievedChunk
_MARKER_RE = re.compile(r"\[(\d+)\]")
def _snippet(text: str, limit: int = 320) -> str:
text = " ".join(text.split())
return text if len(text) <= limit else text[: limit - 1].rstrip() + "…"
def used_marker_indices(answer: str) -> List[int]:
"""Distinct 1-based marker numbers used in the answer, in first-seen order."""
seen = []
for m in _MARKER_RE.finditer(answer):
n = int(m.group(1))
if n not in seen:
seen.append(n)
return seen
def build_citations(answer: str, context: List[RetrievedChunk]) -> List[Citation]:
"""Map used [n] markers to Citation objects (n is 1-based into context)."""
citations: List[Citation] = []
for n in used_marker_indices(answer):
if 1 <= n <= len(context):
c = context[n - 1]
citations.append(
Citation(
marker=str(n),
source_id=c.source_id,
chunk_id=c.chunk_id,
title=c.title,
url=c.url,
publisher=c.publisher,
retrieved=c.retrieved,
snippet=_snippet(c.text),
)
)
return citations
def strip_unsupported_markers(answer: str, context_len: int) -> str:
"""Remove any [n] where n is out of range, so the UI never shows a dead cite."""
def repl(m: "re.Match") -> str:
n = int(m.group(1))
return m.group(0) if 1 <= n <= context_len else ""
cleaned = _MARKER_RE.sub(repl, answer)
# Collapse any double spaces left behind by a removed marker.
return re.sub(r"[ \t]{2,}", " ", cleaned).strip()
|