| """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() |
|
|
