293 lines
9.7 KiB
Python
293 lines
9.7 KiB
Python
#!/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())
|