feat: add rag post-reindex acceptance

scripts/eval_rag_live_acceptance.py

@@ -0,0 +1,292 @@
+#!/usr/bin/env python3
+"""Live acceptance runner for post-reindex RAG retrieval checks.
+
+This script calls the running Spring Boot retrieval endpoint. It is intentionally
+separate from the offline fixture baseline because it depends on live service and
+Milvus/Zilliz state.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+
+DEFAULT_BASE_URL = "http://127.0.0.1:9900"
+DEFAULT_JSON_REPORT = Path("eval/rag-retrieval/reports/live-post-reindex.json")
+DEFAULT_MD_REPORT = Path("eval/rag-retrieval/reports/live-post-reindex.md")
+
+
+DEFAULT_CASES: list[dict[str, Any]] = [
+    {
+        "caseId": "breadcrumb-rag-chunk-context",
+        "query": "If a long RAG section is split into multiple chunks, how do we keep retrieval context?",
+        "topK": 5,
+        "purpose": "Breadcrumb-sensitive RAG chunk context retrieval.",
+    },
+    {
+        "caseId": "breadcrumb-diagnosis-flow",
+        "query": "What is the standard troubleshooting flow for an application incident?",
+        "topK": 5,
+        "purpose": "Process-style retrieval where section path matters.",
+    },
+    {
+        "caseId": "core-err-timeout",
+        "query": "ERR_TIMEOUT",
+        "topK": 3,
+        "purpose": "Exact error-code retrieval should remain stable.",
+    },
+    {
+        "caseId": "core-mysql-connection-pool",
+        "query": "MySQL connection pool is exhausted. How should I diagnose it?",
+        "topK": 3,
+        "purpose": "Core infrastructure troubleshooting retrieval.",
+    },
+    {
+        "caseId": "aiops-payment-latency",
+        "query": "Alert HighLatency on payment-service with p95 latency above threshold",
+        "topK": 3,
+        "purpose": "AIOps alert-style retrieval.",
+    },
+]
+
+
+@dataclass
+class LiveCase:
+    case_id: str
+    query: str
+    top_k: int
+    purpose: str
+    category: str | None = None
+
+    @classmethod
+    def from_json(cls, raw: dict[str, Any]) -> "LiveCase":
+        return cls(
+            case_id=str(raw["caseId"]),
+            query=str(raw["query"]),
+            top_k=int(raw.get("topK") or 3),
+            purpose=str(raw.get("purpose") or raw.get("notes") or ""),
+            category=(
+                str(raw.get("category"))
+                if raw.get("category") not in (None, "")
+                else None
+            ),
+        )
+
+
+def load_cases(path: Path | None) -> list[LiveCase]:
+    if path is None:
+        return [LiveCase.from_json(item) for item in DEFAULT_CASES]
+    with path.open("r", encoding="utf-8") as handle:
+        payload = json.load(handle)
+    raw_cases = payload.get("cases", payload)
+    return [LiveCase.from_json(item) for item in raw_cases]
+
+
+def write_json(path: Path, payload: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", encoding="utf-8", newline="\n") as handle:
+        json.dump(payload, handle, ensure_ascii=False, indent=2)
+        handle.write("\n")
+
+
+def write_text(path: Path, content: str) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", encoding="utf-8", newline="\n") as handle:
+        handle.write(content)
+
+
+def request_case(base_url: str, case: LiveCase, timeout_seconds: float) -> dict[str, Any]:
+    endpoint = base_url.rstrip("/") + "/api/search/similar"
+    params: dict[str, str] = {
+        "query": case.query,
+        "topK": str(case.top_k),
+    }
+    if case.category:
+        params["category"] = case.category
+    url = endpoint + "?" + urllib.parse.urlencode(params)
+
+    started_at = datetime.now(timezone.utc)
+    try:
+        with urllib.request.urlopen(url, timeout=timeout_seconds) as response:
+            body = response.read().decode("utf-8")
+            payload = json.loads(body)
+            status = int(getattr(response, "status", 200))
+    except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
+        return {
+            "caseId": case.case_id,
+            "query": case.query,
+            "topK": case.top_k,
+            "category": case.category,
+            "purpose": case.purpose,
+            "url": url,
+            "ok": False,
+            "error": str(exc),
+            "resultCount": 0,
+            "topCandidates": [],
+            "rawResponse": None,
+            "startedAt": started_at.isoformat(),
+        }
+
+    data = payload.get("data") if isinstance(payload, dict) else None
+    if not isinstance(data, list):
+        data = []
+
+    ok = status == 200 and payload.get("code") == 200
+    return {
+        "caseId": case.case_id,
+        "query": case.query,
+        "topK": case.top_k,
+        "category": case.category,
+        "purpose": case.purpose,
+        "url": url,
+        "ok": ok,
+        "httpStatus": status,
+        "responseCode": payload.get("code"),
+        "responseMessage": payload.get("message"),
+        "resultCount": len(data),
+        "topCandidates": [summarize_candidate(item, index + 1) for index, item in enumerate(data)],
+        "rawResponse": payload,
+        "startedAt": started_at.isoformat(),
+    }
+
+
+def summarize_candidate(raw: dict[str, Any], rank: int) -> dict[str, Any]:
+    metadata = parse_metadata(raw.get("metadata"))
+    return {
+        "rank": rank,
+        "id": raw.get("id"),
+        "title": metadata.get("title"),
+        "breadcrumb": metadata.get("breadcrumb"),
+        "category": metadata.get("category"),
+        "source": metadata.get("_source") or metadata.get("source"),
+        "score": raw.get("score"),
+        "rawScore": raw.get("rawScore"),
+        "scoreLabel": raw.get("scoreLabel"),
+        "contentPreview": preview(raw.get("content")),
+    }
+
+
+def parse_metadata(value: Any) -> dict[str, Any]:
+    if isinstance(value, dict):
+        return value
+    if isinstance(value, str) and value.strip():
+        try:
+            parsed = json.loads(value)
+            return parsed if isinstance(parsed, dict) else {}
+        except json.JSONDecodeError:
+            return {}
+    return {}
+
+
+def preview(value: Any, limit: int = 180) -> str:
+    text = " ".join(str(value or "").split())
+    if len(text) <= limit:
+        return text
+    return text[: limit - 3] + "..."
+
+
+def render_markdown(report: dict[str, Any]) -> str:
+    lines = [
+        "# RAG Live Post-Reindex Acceptance",
+        "",
+        f"Generated at: `{report['generatedAt']}`",
+        f"Base URL: `{report['baseUrl']}`",
+        "",
+        "> Reindex prerequisite: this report only reflects breadcrumb-aware embedding if the knowledge base was reindexed after the embedding-text change.",
+        "",
+        "## Summary",
+        "",
+        "| Metric | Value |",
+        "|---|---:|",
+        f"| Cases | {report['caseCount']} |",
+        f"| Successful calls | {report['successfulCalls']} |",
+        f"| Empty result cases | {report['emptyResultCases']} |",
+        "",
+        "## Cases",
+        "",
+        "| Case | Purpose | Results | Top Candidates |",
+        "|---|---|---:|---|",
+    ]
+    for item in report["results"]:
+        top = "<br>".join(format_candidate(candidate) for candidate in item["topCandidates"])
+        if not top and item.get("error"):
+            top = "ERROR: " + str(item["error"])
+        lines.append(
+            "| {case} | {purpose} | {count} | {top} |".format(
+                case=item["caseId"],
+                purpose=item.get("purpose") or "",
+                count=item["resultCount"],
+                top=top,
+            )
+        )
+    lines.append("")
+    return "\n".join(lines)
+
+
+def format_candidate(candidate: dict[str, Any]) -> str:
+    label = candidate.get("title") or candidate.get("source") or candidate.get("id") or ""
+    breadcrumb = candidate.get("breadcrumb") or ""
+    score_label = candidate.get("scoreLabel") or ""
+    score = candidate.get("score")
+    raw_score = candidate.get("rawScore")
+    details = f"score={score}"
+    if raw_score is not None:
+        details += f", raw={raw_score}"
+    if score_label:
+        details += f", label={score_label}"
+    if breadcrumb:
+        return f"{candidate['rank']}. {label} ({breadcrumb}; {details})"
+    return f"{candidate['rank']}. {label} ({details})"
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--base-url", default=DEFAULT_BASE_URL)
+    parser.add_argument("--cases", type=Path, default=None)
+    parser.add_argument("--json-report", type=Path, default=DEFAULT_JSON_REPORT)
+    parser.add_argument("--markdown-report", type=Path, default=DEFAULT_MD_REPORT)
+    parser.add_argument("--timeout-seconds", type=float, default=10.0)
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    cases = load_cases(args.cases)
+    results = [
+        request_case(args.base_url, case, args.timeout_seconds)
+        for case in cases
+    ]
+    successful = [item for item in results if item["ok"]]
+    empty = [item for item in results if item["ok"] and item["resultCount"] == 0]
+    report = {
+        "generatedAt": datetime.now(timezone.utc).isoformat(),
+        "baseUrl": args.base_url,
+        "caseCount": len(results),
+        "successfulCalls": len(successful),
+        "emptyResultCases": len(empty),
+        "reindexPrerequisite": "Run or trigger knowledge-base reindex before treating this as breadcrumb-aware embedding evidence.",
+        "results": results,
+    }
+    write_json(args.json_report, report)
+    write_text(args.markdown_report, render_markdown(report))
+    print(
+        "Ran {total} live cases: successful={successful}, empty={empty}".format(
+            total=len(results),
+            successful=len(successful),
+            empty=len(empty),
+        )
+    )
+    return 1 if len(successful) != len(results) else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())