NuklearRabbit
b511ba2dbc
KnowledgeProvider protocol with a deterministic TF-IDF-weighted extractive demo provider (never generative, always cites real excerpts) and a RAGcore HTTP adapter that degrades cleanly to unavailable. Knowledge nav + chat-style Q&A UI with source cards and honest grounded/insufficient/unavailable states. 57 backend tests passing, ruff clean. Fixed a real relevance bug (generic terms like "vehicle" crowding out distinctive matches) via IDF weighting, found by testing the actual S6 scenario. Verified end-to-end in the browser: grounded damage question cites both expected procedures; unrelated question honestly returns insufficient evidence with no fabrication.
222 lines
7.6 KiB
Python
222 lines
7.6 KiB
Python
from __future__ import annotations
|
|
|
|
import math
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
from app.core.config import get_settings
|
|
from app.services.knowledge import GroundedAnswer, KnowledgeHealth, SourceCard
|
|
|
|
STOPWORDS = {
|
|
"a", "an", "the", "is", "are", "was", "were", "be", "been", "being",
|
|
"to", "of", "in", "on", "at", "for", "and", "or", "but", "if", "then",
|
|
"do", "does", "did", "must", "may", "can", "could", "should", "would",
|
|
"i", "you", "it", "we", "they", "my", "your", "what", "when", "how",
|
|
"with", "without", "this", "that", "these", "those", "not", "no",
|
|
}
|
|
|
|
_WORD_RE = re.compile(r"[a-z0-9]+")
|
|
|
|
|
|
def _stem(word: str) -> str:
|
|
# Deterministic, intentionally crude suffix stripping — good enough to match "returns"
|
|
# with "return" or "damaged" with "damage" without pulling in a stemming dependency.
|
|
for suffix in ("ing", "edly", "ed", "es", "s"):
|
|
if len(word) > len(suffix) + 2 and word.endswith(suffix):
|
|
return word[: -len(suffix)]
|
|
return word
|
|
|
|
|
|
def _tokenize(text: str) -> set[str]:
|
|
words = _WORD_RE.findall(text.lower())
|
|
return {_stem(w) for w in words if w not in STOPWORDS and len(w) > 2}
|
|
|
|
|
|
@dataclass
|
|
class Document:
|
|
document_id: str
|
|
title: str
|
|
version: str
|
|
title_tokens: set[str] = field(default_factory=set)
|
|
|
|
|
|
@dataclass
|
|
class ScoredSection:
|
|
document: Document
|
|
heading: str
|
|
text: str
|
|
heading_tokens: set[str]
|
|
body_tokens: set[str]
|
|
|
|
|
|
def _parse_frontmatter(raw: str) -> tuple[dict[str, str], str]:
|
|
if not raw.startswith("---"):
|
|
return {}, raw
|
|
end = raw.find("\n---", 3)
|
|
if end == -1:
|
|
return {}, raw
|
|
block = raw[3:end].strip()
|
|
body = raw[end + 4 :].lstrip("\n")
|
|
meta: dict[str, str] = {}
|
|
for line in block.splitlines():
|
|
if ":" not in line:
|
|
continue
|
|
key, _, value = line.partition(":")
|
|
meta[key.strip()] = value.strip().strip('"')
|
|
return meta, body
|
|
|
|
|
|
def _split_sections(body: str) -> list[tuple[str, str]]:
|
|
sections: list[tuple[str, str]] = []
|
|
current_heading = "Overview"
|
|
current_lines: list[str] = []
|
|
for line in body.splitlines():
|
|
if line.startswith("## "):
|
|
if current_lines:
|
|
sections.append((current_heading, "\n".join(current_lines).strip()))
|
|
current_heading = line[3:].strip()
|
|
current_lines = []
|
|
elif line.startswith("# "):
|
|
continue
|
|
else:
|
|
current_lines.append(line)
|
|
if current_lines:
|
|
sections.append((current_heading, "\n".join(current_lines).strip()))
|
|
return sections
|
|
|
|
|
|
def _load_sections(procedures_dir: Path) -> list[ScoredSection]:
|
|
sections: list[ScoredSection] = []
|
|
for path in sorted(procedures_dir.glob("*.md")):
|
|
raw = path.read_text(encoding="utf-8")
|
|
meta, body = _parse_frontmatter(raw)
|
|
title = meta.get("title", path.stem)
|
|
doc = Document(
|
|
document_id=meta.get("document_id", path.stem),
|
|
title=title,
|
|
version=meta.get("version", "1.0"),
|
|
title_tokens=_tokenize(title),
|
|
)
|
|
for heading, text in _split_sections(body):
|
|
sections.append(
|
|
ScoredSection(
|
|
document=doc,
|
|
heading=heading,
|
|
text=text,
|
|
heading_tokens=_tokenize(heading),
|
|
body_tokens=_tokenize(text),
|
|
)
|
|
)
|
|
return sections
|
|
|
|
|
|
class DemoKnowledgeProvider:
|
|
"""Deterministic extractive retrieval over the local procedure Markdown files.
|
|
|
|
Not a generative model: it scores sections with TF-IDF-weighted keyword overlap
|
|
(downweighting terms common across the whole corpus, like "vehicle", in favor of
|
|
distinctive ones, like "damage") and returns real excerpts, never invented text.
|
|
"""
|
|
|
|
name = "demo"
|
|
|
|
def __init__(self) -> None:
|
|
settings = get_settings()
|
|
self._settings = settings
|
|
self._procedures_dir = Path(settings.knowledge_dir)
|
|
self._sections = _load_sections(self._procedures_dir)
|
|
self._document_count = len({s.document.document_id for s in self._sections})
|
|
self._idf = self._build_idf(self._sections)
|
|
|
|
@staticmethod
|
|
def _build_idf(sections: list[ScoredSection]) -> dict[str, float]:
|
|
n = len(sections) or 1
|
|
doc_freq: dict[str, int] = {}
|
|
for section in sections:
|
|
doc = section.document
|
|
all_tokens = doc.title_tokens | section.heading_tokens | section.body_tokens
|
|
for token in all_tokens:
|
|
doc_freq[token] = doc_freq.get(token, 0) + 1
|
|
return {token: math.log((n + 1) / (df + 1)) + 1 for token, df in doc_freq.items()}
|
|
|
|
def health(self) -> KnowledgeHealth:
|
|
return KnowledgeHealth(
|
|
provider=self.name,
|
|
available=True,
|
|
detail="Deterministic keyword-matching demo provider; no external service.",
|
|
tenant=self._settings.ragcore_tenant,
|
|
workspace=self._settings.ragcore_workspace,
|
|
collection=self._settings.ragcore_collection,
|
|
document_count=self._document_count,
|
|
)
|
|
|
|
def _score(self, query_tokens: set[str], section: ScoredSection) -> float:
|
|
score = 0.0
|
|
for token in query_tokens:
|
|
idf = self._idf.get(token, 0.0)
|
|
if idf == 0.0:
|
|
continue
|
|
if token in section.heading_tokens:
|
|
score += 3 * idf
|
|
elif token in section.document.title_tokens:
|
|
score += 2 * idf
|
|
elif token in section.body_tokens:
|
|
score += idf
|
|
return score
|
|
|
|
def ask(self, question: str, correlation_id: str) -> GroundedAnswer:
|
|
query_tokens = _tokenize(question)
|
|
scored = [
|
|
(self._score(query_tokens, section), section)
|
|
for section in self._sections
|
|
]
|
|
scored = [(score, section) for score, section in scored if score > 0]
|
|
scored.sort(key=lambda item: item[0], reverse=True)
|
|
top = scored[:3]
|
|
|
|
if not top:
|
|
return GroundedAnswer(
|
|
answer="No matching procedure was found for this question.",
|
|
evidence_state="insufficient",
|
|
sources=[],
|
|
provider=self.name,
|
|
correlation_id=correlation_id,
|
|
)
|
|
|
|
sources = [
|
|
SourceCard(
|
|
document_id=section.document.document_id,
|
|
title=section.document.title,
|
|
version=section.document.version,
|
|
section=section.heading,
|
|
excerpt=(section.text[:400] + "…") if len(section.text) > 400 else section.text,
|
|
)
|
|
for _, section in top
|
|
]
|
|
|
|
if top[0][0] < 3:
|
|
return GroundedAnswer(
|
|
answer=(
|
|
"The available procedures do not clearly answer this question. "
|
|
"The closest matches are included below for review."
|
|
),
|
|
evidence_state="insufficient",
|
|
sources=sources,
|
|
provider=self.name,
|
|
correlation_id=correlation_id,
|
|
)
|
|
|
|
lead_section = top[0][1]
|
|
answer = (
|
|
f'Per "{lead_section.document.title}" (v{lead_section.document.version}), '
|
|
f'section "{lead_section.heading}": {lead_section.text.splitlines()[0][:300]}'
|
|
)
|
|
return GroundedAnswer(
|
|
answer=answer,
|
|
evidence_state="grounded",
|
|
sources=sources,
|
|
provider=self.name,
|
|
correlation_id=correlation_id,
|
|
)
|