120 lines
5.1 KiB
Python
120 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
"""اجرای مجموعه طلایی روی سرویس چت‌بات و ساخت گزارش مارک‌داون.
|
|
|
|
استفاده (داخل کانتینر یا هر جا که به سرویس دسترسی هست):
|
|
python eval_run.py --base http://localhost:8000
|
|
python eval_run.py --out eval-report.md
|
|
|
|
هر پرسش از /chat/eval سرویس می‌شود؛ پاسخ با زیررشته‌های انتظار
|
|
(پس از نرمال‌سازی فارسی) مقایسه و رتبه‌بندی می‌شود:
|
|
pass — همه/یکی از انتظارها برآورده شد
|
|
miss — انتظار برآورده نشد (پاسخ احتمالاً نادرست یا بی‌ربط)
|
|
manual — انتظار عددی تعریف نشده؛ بازبینی دستی
|
|
fallback-expected / unexpected-fallback — پیام جایگزین
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import time
|
|
import urllib.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
FALLBACK_MESSAGE = (
|
|
"متأسفانه اطلاعات دقیقی در این باره در سایت ثبت نشده است. "
|
|
"برای راهنمایی بیشتر می‌توانید از صفحه «تماس با ما» با ما در ارتباط باشید."
|
|
)
|
|
|
|
|
|
def ask(base: str, question: str, timeout: int = 180) -> dict:
|
|
url = f"{base}/chat/eval?{urllib.parse.urlencode({'message': question})}"
|
|
with urllib.request.urlopen(url, timeout=timeout) as r:
|
|
return json.load(r)
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser(description="اجرای مجموعه طلایی ارزیابی چت‌بات")
|
|
ap.add_argument("--base", default="http://localhost:8000", help="نشانی سرویس چت‌بات")
|
|
ap.add_argument("--set", dest="eval_set",
|
|
default=str(Path(__file__).with_name("eval_set.json")))
|
|
ap.add_argument("--out", default="-", help="فایل خروجی گزارش (- = stdout)")
|
|
ap.add_argument("--sleep", type=float, default=0.5, help="مکث بین پرسش‌ها (ثانیه)")
|
|
args = ap.parse_args()
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from chatbot.persian import normalize
|
|
|
|
data = json.loads(Path(args.eval_set).read_text(encoding="utf-8"))
|
|
rows: list[dict] = []
|
|
|
|
for item in data["questions"]:
|
|
q = item["q"]
|
|
t0 = time.time()
|
|
try:
|
|
res = ask(args.base, q)
|
|
except Exception as e:
|
|
rows.append({**item, "answer": f"ERROR: {e}", "sources": [], "seconds": 0.0,
|
|
"status": "error", "fallback": False})
|
|
print(f" error: {q[:50]}", file=sys.stderr)
|
|
continue
|
|
seconds = time.time() - t0
|
|
answer = str(res.get("answer", ""))
|
|
contexts = res.get("contexts") or []
|
|
is_fallback = answer.strip() == FALLBACK_MESSAGE or not contexts
|
|
|
|
if item.get("expect_fallback"):
|
|
status = "fallback-expected" if is_fallback else "unexpected-answer"
|
|
elif is_fallback:
|
|
status = "unexpected-fallback"
|
|
else:
|
|
needles = [normalize(s) for s in item.get("expect_any", [])]
|
|
if not needles:
|
|
status = "manual"
|
|
elif any(n in normalize(answer) for n in needles):
|
|
status = "pass"
|
|
else:
|
|
status = "miss"
|
|
|
|
rows.append({**item, "answer": answer, "sources": contexts, "seconds": seconds,
|
|
"status": status, "fallback": is_fallback})
|
|
print(f" [{status}] {q[:50]}", file=sys.stderr)
|
|
|
|
time.sleep(args.sleep)
|
|
|
|
# --- گزارش ---
|
|
lines = [f"# گزارش ارزیابی چت‌بات — {time.strftime('%Y-%m-%d %H:%M')}", ""]
|
|
counts: dict[str, int] = {}
|
|
for r in rows:
|
|
counts[r["status"]] = counts.get(r["status"], 0) + 1
|
|
graded = [r for r in rows if r["status"] in ("pass", "miss", "unexpected-fallback", "unexpected-answer")]
|
|
ok = sum(1 for r in rows if r["status"] in ("pass", "fallback-expected"))
|
|
lines += [
|
|
f"- مجموع پرسش‌ها: {len(rows)} — درست: {ok} — خطا: {counts.get('miss', 0) + counts.get('unexpected-fallback', 0) + counts.get('unexpected-answer', 0)}",
|
|
f"- بازبینی دستی: {counts.get('manual', 0)} — خطای فنی: {counts.get('error', 0)}",
|
|
f"- میانگین زمان پاسخ: {sum(r['seconds'] for r in rows) / max(len(rows), 1):.1f}s",
|
|
"",
|
|
]
|
|
for r in rows:
|
|
srcs = [c.rsplit("(منبع:", 1)[-1].strip(" ()\n") for c in r["sources"]] if r["sources"] else []
|
|
lines += [
|
|
f"## [{r['status']}] {r['q']}",
|
|
f"*موضوع: {r.get('topic', '?')} — {r['seconds']:.1f}s — منابع: {'، '.join(srcs) or '—'}*",
|
|
"",
|
|
r["answer"],
|
|
"",
|
|
]
|
|
|
|
report = "\n".join(lines)
|
|
if args.out == "-":
|
|
print(report)
|
|
else:
|
|
Path(args.out).write_text(report, encoding="utf-8")
|
|
print(f"Report written to {args.out}", file=sys.stderr)
|
|
|
|
return 1 if any(r["status"] in ("miss", "unexpected-answer") for r in rows) else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|