Step 8: Test and Iterate

# test_pipeline.py
# ==========================================
# Evaluate the RAG pipeline with test cases
# ==========================================
 
import time
from src.query import ask
 
def run_tests():
    """Run test questions and evaluate the results."""
    test_cases = [
        {
            "question": "What is the refund policy for digital products?",
            "expected_keywords": ["5 business days", "digital"],
            "expected_source": "company-policy.txt",
        },
        {
            "question": "What happens on day 1 of onboarding?",
            "expected_keywords": ["HR orientation", "laptop", "badge"],
            "expected_source": "onboarding-guide.txt",
        },
        {
            "question": "How do I get a store credit?",
            "expected_keywords": ["30 days", "store credit"],
            "expected_source": "company-policy.txt",
        },
        {
            "question": "What tools do new employees need?",
            "expected_keywords": ["Slack", "GitHub", "Jira"],
            "expected_source": "onboarding-guide.txt",
        },
        {
            # This question has no answer in our docs
            "question": "What is the company's vacation policy?",
            "expected_keywords": [],
            "expected_source": None,
        },
    ]
 
    results = []
    for i, tc in enumerate(test_cases):
        print(f"\n{'=' * 60}")
        print(f"Test {i + 1}: {tc['question']}")
        print(f"{'=' * 60}")
 
        start = time.time()
        result = ask(tc["question"])
        elapsed = time.time() - start
 
        answer = result["answer"]
        source_files = [s["file"] for s in result["sources"]]
 
        # Check if expected keywords appear in the answer
        keyword_hits = sum(
            1 for kw in tc["expected_keywords"]
            if kw.lower() in answer.lower()
        )
        keyword_total = len(tc["expected_keywords"])
 
        # Check if the expected source was retrieved
        source_match = (
            tc["expected_source"] in source_files
            if tc["expected_source"]
            else True
        )
 
        passed = keyword_hits == keyword_total and source_match
 
        print(f"Answer: {answer[:300]}...")
        print(f"Sources: {source_files}")
        print(f"Keywords: {keyword_hits}/{keyword_total}")
        print(f"Source match: {'yes' if source_match else 'NO'}")
        print(f"Latency: {elapsed:.2f}s")
        print(f"Result: {'PASS' if passed else 'FAIL'}")
 
        results.append(passed)
 
    # Summary
    print(f"\n{'=' * 60}")
    print(f"Results: {sum(results)}/{len(results)} passed")
    print(f"{'=' * 60}")
 
 
if __name__ == "__main__":
    run_tests()

python test_pipeline.py

python -m src.ingest

# For strict factual answers (no speculation):
system_prompt = (
    "Answer ONLY based on the provided document excerpts. "
    "If the answer is not in the documents, say: "
    "'I could not find this information in the available documents.' "
    "Never speculate or use outside knowledge."
)
 
# For more conversational answers:
system_prompt = (
    "You are a helpful company assistant. Answer questions based on "
    "the provided documents. Be conversational but accurate. "
    "If you are not sure, say so."
)