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, )