scripts/evaluate.py

"""
ByteAstra — Evaluation & Grounding QA Script.

Runs a set of predefined test queries against the tutoring engine and
reports on:
  - Whether grounded answers are returned for in-syllabus questions
  - Whether out-of-syllabus responses are returned for off-topic questions
  - Whether domain isolation is preserved (cross-domain queries should not leak)
  - Citation validity

Usage:
  python scripts/evaluate.py --domain ayurveda

Add more test cases in the TESTS list below as the knowledge base grows.
"""
from __future__ import annotations
import asyncio
import json
import logging
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Literal

sys.path.insert(0, str(Path(__file__).parent.parent))
sys.stdout.reconfigure(encoding='utf-8')

from app.services import engine

logging.basicConfig(level=logging.WARNING, format="%(levelname)s | %(message)s")
logger = logging.getLogger(__name__)


# ── Test Case Definition ───────────────────────────────────────────────────────

@dataclass
class TestCase:
    id: str
    domain: str
    query: str
    expected_grounded: bool  # True = expect in-syllabus grounded answer
    description: str
    history: list[dict] = field(default_factory=list)


# ── Test Suite ─────────────────────────────────────────────────────────────────

TEST_CASES: list[TestCase] = [
    # --- In-syllabus: expect grounded answers ---
    TestCase(
        id="ayurveda_001",
        domain="ayurveda",
        query="What are the three Doshas in Ayurveda and their qualities?",
        expected_grounded=True,
        description="Core Tridosha concept — should always be in the KB.",
    ),
    TestCase(
        id="ayurveda_002",
        domain="ayurveda",
        query="Explain the concept of Agni in Ayurveda and its role in digestion.",
        expected_grounded=True,
        description="Agni concept — fundamental to BAMS syllabus.",
    ),
    TestCase(
        id="ayurveda_003",
        domain="ayurveda",
        query="What are the seven Dhatus according to Ayurvedic texts?",
        expected_grounded=True,
        description="Sapta Dhatu — core anatomy concept.",
    ),
    # --- Out-of-syllabus: expect fallback ---
    TestCase(
        id="oos_001",
        domain="ayurveda",
        query="What is the mechanism of action of ibuprofen?",
        expected_grounded=False,
        description="Allopathic pharmacology — should NOT be in Ayurveda KB.",
    ),
    TestCase(
        id="oos_002",
        domain="ayurveda",
        query="Explain Section 420 of the Indian Penal Code.",
        expected_grounded=False,
        description="Law question — completely off-domain.",
    ),
    # --- Domain isolation (if multiple domains indexed) ---
    TestCase(
        id="isolation_001",
        domain="ayurveda",
        query="What are the anatomical layers of the skin according to modern anatomy?",
        expected_grounded=False,
        description="MBBS-style question posed to Ayurveda domain — should not bleed.",
    ),
]


# ── Runner ─────────────────────────────────────────────────────────────────────

@dataclass
class TestResult:
    case: TestCase
    actual_grounded: bool
    citations_count: int
    answer_preview: str
    passed: bool
    failure_reason: str = ""


async def run_test(case: TestCase) -> TestResult:
    full_content: list[str] = []
    actual_grounded = False
    citations_count = 0

    try:
        async for event in engine.stream_answer(case.domain, case.query, case.history):
            if event["type"] == "citations":
                actual_grounded = event.get("is_grounded", False)
                citations_count = len(event.get("citations", []))
            elif event["type"] == "delta":
                full_content.append(event.get("content", ""))
            elif event["type"] == "error":
                return TestResult(
                    case=case,
                    actual_grounded=False,
                    citations_count=0,
                    answer_preview="ERROR: " + event.get("error", "unknown"),
                    passed=False,
                    failure_reason="Engine returned error",
                )
    except Exception as e:
        return TestResult(
            case=case,
            actual_grounded=False,
            citations_count=0,
            answer_preview="",
            passed=False,
            failure_reason=f"Exception: {e}",
        )

    answer = "".join(full_content).strip()
    grounded_match = actual_grounded == case.expected_grounded
    passed = grounded_match

    failure_reason = ""
    if not grounded_match:
        if case.expected_grounded:
            failure_reason = f"Expected grounded answer but got out-of-syllabus response."
        else:
            failure_reason = f"Expected out-of-syllabus response but got grounded answer with {citations_count} citation(s)."

    return TestResult(
        case=case,
        actual_grounded=actual_grounded,
        citations_count=citations_count,
        answer_preview=answer[:200],
        passed=passed,
        failure_reason=failure_reason,
    )


async def run_suite(domain_filter: str | None = None) -> None:
    cases = TEST_CASES
    if domain_filter:
        cases = [c for c in cases if c.domain == domain_filter]

    print(f"\n{'='*65}")
    print(f"  ByteAstra Evaluation Suite — {len(cases)} test(s)")
    print(f"{'='*65}\n")

    results: list[TestResult] = []
    for case in cases:
        print(f"  ▶ [{case.id}] {case.description}")
        result = await run_test(case)
        results.append(result)
        status = "✓ PASS" if result.passed else "✗ FAIL"
        grounded_label = "grounded" if result.actual_grounded else "out-of-syllabus"
        print(f"    {status} | {grounded_label} | {result.citations_count} citation(s)")
        if not result.passed:
            print(f"    ⚠  {result.failure_reason}")
        if result.answer_preview:
            print(f"    Preview: {result.answer_preview[:120]}...")
        print()

    passed = sum(1 for r in results if r.passed)
    failed = len(results) - passed
    print(f"{'='*65}")
    print(f"  Results: {passed}/{len(results)} passed | {failed} failed")
    print(f"{'='*65}\n")

    if failed:
        sys.exit(1)


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--domain", default=None, help="Filter tests to a specific domain")
    args = parser.parse_args()
    asyncio.run(run_suite(domain_filter=args.domain))