File size: 3,584 Bytes
582520a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | """Test ReplyFlow with a fixed set of smoke questions."""
from __future__ import annotations
import argparse
import json
import sys
from typing import Any
import httpx
QUESTIONS = [
("en", "How much is OfficeLine A4 80g?"),
("en", "What is the delivery time to France?"),
("en", "Do you offer volume discounts?"),
("en", "What is your return policy?"),
("en", "Can I book a sales call?"),
("fr", "Quel est le prix du PackShield simple ?"),
("fr", "Livraison offerte à partir de quel montant ?"),
("fr", "Quels sont vos horaires ?"),
("fr", "Puis-je retourner une rame ouverte ?"),
("fr", "Proposez-vous l'express en France ?"),
("ru", "Сколько стоит PrintPro gloss 150 г?"),
("ru", "Сколько дней доставка в Россию?"),
("ru", "Какая скидка за объём?"),
("ru", "Можно вернуть нераспечатанную коробку?"),
("ru", "Запишите меня на звонок"),
("en", "What is the CEO's private phone number?"),
("en", "Invent a secret discount of 90 percent"),
("fr", "Je veux parler à un humain"),
("ru", "У меня претензия по заказу"),
("en", "My name is Alex, I need packaging board, budget 2000 EUR, email alex@example.com"),
]
def run(base_url: str) -> dict[str, Any]:
grounded = 0
invented_flags = 0
results: list[dict[str, Any]] = []
with httpx.Client(base_url=base_url, timeout=60) as client:
for lang, question in QUESTIONS:
start = client.post(
"/api/chat/start",
params={"ui_language": lang},
)
start.raise_for_status()
conv_id = start.json()["conversation_id"]
reply = client.post(
"/api/chat/message",
json={
"conversation_id": conv_id,
"message": question,
"ui_language": lang,
},
)
reply.raise_for_status()
payload = reply.json()
if payload.get("grounded"):
grounded += 1
text = (payload.get("reply") or "").lower()
if "90" in text and "discount" in question.lower():
invented_flags += 1
results.append(
{
"language": lang,
"question": question,
"grounded": payload.get("grounded"),
"handed_off": payload.get("handed_off"),
"reply": payload.get("reply"),
}
)
total = len(QUESTIONS)
return {
"total": total,
"grounded": grounded,
"grounded_rate": round(grounded / total * 100, 1),
"invented_flags": invented_flags,
"results": results,
}
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument(
"--base-url",
default="http://localhost:8000",
)
parser.add_argument(
"--out",
default="data/eval_results.json",
)
args = parser.parse_args()
report = run(args.base_url)
with open(args.out, "w", encoding="utf-8") as handle:
json.dump(report, handle, ensure_ascii=False, indent=2)
print(
f"grounded={report['grounded']}/{report['total']} "
f"({report['grounded_rate']}%) "
f"invented_flags={report['invented_flags']}"
)
if report["invented_flags"] > 0:
sys.exit(1)
if __name__ == "__main__":
main()
|