Retrieval-Augmented Answering RAG Project

"""A small extractive RAG-style pipeline with source citations."""

from __future__ import annotations

from dataclasses import dataclass
import re
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity


PASSAGES = [
    ("handbook-attendance", "Learners qualify for the attendance requirement when they attend at least 80 percent of scheduled sessions."),
    ("handbook-projects", "Final projects should include source code, setup instructions, tests, and a short explanation of limitations."),
    ("handbook-certificates", "Certificate eligibility is reviewed after attendance and required project work are completed."),
    ("handbook-support", "Technical questions should include the error message, environment details, and steps to reproduce the issue."),
]


@dataclass(frozen=True)
class RagAnswer:
    answer: str
    sources: tuple[str, ...]
    scores: tuple[float, ...]


def answer(question: str, top_k: int = 2, threshold: float = 0.1) -> RagAnswer:
    texts = [text for _, text in PASSAGES]
    vectorizer = TfidfVectorizer(stop_words="english", ngram_range=(1, 2))
    matrix = vectorizer.fit_transform(texts + [question])
    scores = cosine_similarity(matrix[-1], matrix[:-1])[0]
    selected = [int(i) for i in scores.argsort()[::-1][:top_k] if scores[i] >= threshold]
    if not selected:
        return RagAnswer("The supplied knowledge base does not contain a confident answer.", (), ())
    sentences = []
    for index in selected:
        sentences.extend(part.strip() for part in re.split(r"(?<=[.!?])\s+", texts[index]) if part.strip())
    return RagAnswer(" ".join(sentences), tuple(PASSAGES[i][0] for i in selected), tuple(float(scores[i]) for i in selected))


def main() -> None:
    result = answer("What should the final project contain?")
    print(result.answer)
    print("Sources:", ", ".join(result.sources))
    print("This extractive demo returns source text and does not invent an answer.")


if __name__ == "__main__":
    main()