""" simulate.py — stress-test the assistant with ~1000+ simulated user questions across many scenarios. Coverage: 1. PARAPHRASES of every real KB question, typed the way a hurried/casual user would (2 variants). These should be answerable, and we know the expected source. Escalations = retrieval gaps. 2. SCENARIOS generated per category (emotional struggle, technical problem, program curiosity, general billing, account-specific billing, off-topic, medical, crisis, human-request), each with an expected behavior so we can score coverage. 3. WARMTH: a judge scores a sample of answers 1-5 on how warm / empathetic / coach-like they are. Outputs: eval/sim_results.json + eval/sim_report.md (numbers, the gap list, low-warmth examples). Run: python3 eval/simulate.py """ import json import os import sys import time from concurrent.futures import ThreadPoolExecutor ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, ROOT) import app # noqa: E402 from app import ChatRequest, retrieve, retrieval_query, client # noqa: E402 from google.genai import types # noqa: E402 GEN_MODEL = os.getenv("GEMINI_MODEL", "gemini-2.5-flash") VARIANTS = 2 WORKERS = 6 WARMTH_SAMPLE = 150 KB = json.load(open(os.path.join(ROOT, "data", "kb.json"))) ANSWERABLE_KB = [e for e in KB if not e.get("coach_only") and not e.get("has_placeholder")] STYLES = [ "Rewrite each question the way a hurried real user would type it in a support chat: casual, " "lowercase, maybe a small typo, shorter. Keep the meaning.", "Rewrite each question in a different casual phrasing a real user might use (synonyms, " "indirect, maybe a spelling mistake). Keep the meaning.", ] # category key, expected behavior, generation description SCENARIO_CATS = [ ("emotional_struggle", "answer", "a real user struggling while quitting smoking: a craving right now, " "a slip/relapse, low motivation, fear of failure, stress, or anxiety. First person, emotional, casual."), ("technical_problem", "answer", "a real user reporting an app problem: blank/grey screen, video won't load, " "can't log in, lost progress, app crashing, notifications not working. Casual, first person."), ("program_curiosity", "answer", "a real user curious how the QuitSure program works: structure, days, what's " "included, the Stay Free / post-quit part, the Relaxed program. Casual, first person."), ("billing_general", "answer", "a real user asking general subscription questions: pricing, plans, free trial, " "how to cancel, what happens after it expires, refunds policy. Casual, first person."), ("account_specific", "escalate", "a real user with an account-specific billing problem that needs a human to " "look up their account: charged twice, money deducted but program not active, refund not received yet."), ("off_topic", "decline", "completely unrelated time-pass messages: weather, sports, jokes, general knowledge, " "math, coding, recipes, celebrities."), ("medical", "escalate", "questions about medication, nicotine-replacement dosage, drug interactions, or " "specific physical/health symptoms while quitting."), ("human_request", "escalate", "a user explicitly asking to talk to a human, a coach, or a real agent."), ] ARR_SCHEMA = {"type": "array", "items": {"type": "string"}} WARMTH_SCHEMA = {"type": "object", "properties": {"warmth": {"type": "integer"}, "reason": {"type": "string"}}, "required": ["warmth"]} WARMTH_RULES = ("Score how warm, empathetic, and coach-like a support reply is, 1 to 5. " "1 = cold/robotic/curt. 5 = warm, caring, encouraging like a great quit-smoking coach. " "Judge tone only. Respond JSON: {\"warmth\": int, \"reason\": str}") def _gen_array(prompt, n, temp=1.0): for attempt in range(3): try: resp = client.models.generate_content( model=GEN_MODEL, contents=prompt, config=types.GenerateContentConfig(response_mime_type="application/json", response_schema=ARR_SCHEMA, temperature=temp)) arr = json.loads(resp.text) if isinstance(arr, list) and arr: return arr except Exception: time.sleep(2 * (attempt + 1)) return [] def gen_paraphrases(batch, style): numbered = "\n".join(f"{i+1}. {q}" for i, q in enumerate(batch)) arr = _gen_array(f"{style}\nReturn a JSON array of {len(batch)} rewritten questions, same order.\n\n{numbered}", len(batch)) return arr if len(arr) == len(batch) else list(batch) def gen_scenarios(desc, n): return _gen_array(f"Generate {n} distinct, realistic chat messages from {desc}\n" f"Vary wording and length. Return a JSON array of {n} strings.", n)[:n] def build_question_set(): cases = [] batches = [ANSWERABLE_KB[i:i + 20] for i in range(0, len(ANSWERABLE_KB), 20)] for v in range(VARIANTS): for b in batches: paras = gen_paraphrases([e["question"] for e in b], STYLES[v % len(STYLES)]) for orig, para in zip(b, paras): cases.append({"q": para, "kind": "paraphrase", "expect": "answer", "expected_id": orig["id"], "expected_q": orig["question"], "program": (None if "all" in orig["programs"] else orig["programs"][0])}) print(f" paraphrase variant {v+1}/{VARIANTS} ({len(cases)})") for key, expect, desc in SCENARIO_CATS: for q in gen_scenarios(desc, 30): cases.append({"q": q, "kind": key, "expect": expect, "expected_id": None, "expected_q": None, "program": None}) print(f" scenario {key} ({len(cases)})") return cases def behavior_of(out): if out["escalate"]: return "escalate" if out.get("escalate_reason") == "out_of_scope": return "decline" return "answer" def run_one(case): try: req = ChatRequest(message=case["q"], program=case.get("program")) hits, sim = retrieve(retrieval_query(req), program=req.program) out = app.chat(req) retrieved_ids = [h["id"] for h, _ in hits] return {**case, "behavior": behavior_of(out), "reason": out.get("escalate_reason"), "best_sim": round(sim, 3), "answer": out["answer"], "expected_retrieved": (case["expected_id"] in retrieved_ids) if case["expected_id"] is not None else None} except Exception as e: return {**case, "behavior": "error", "reason": str(e)[:120], "best_sim": 0, "answer": "", "expected_retrieved": None} def warmth_judge(answer, question): try: resp = client.models.generate_content( model=GEN_MODEL, contents=f"USER: {question}\nREPLY: {answer}", config=types.GenerateContentConfig(system_instruction=WARMTH_RULES, response_mime_type="application/json", response_schema=WARMTH_SCHEMA, temperature=0.0)) return json.loads(resp.text) except Exception: return None def pct(a, b): return f"{(100*a)//max(b,1)}%" def main(): print("Building question set (paraphrases + scenarios)...") cases = build_question_set() print(f"Running {len(cases)} simulated questions...") results = [] with ThreadPoolExecutor(max_workers=WORKERS) as ex: for i, r in enumerate(ex.map(run_one, cases)): results.append(r) if (i + 1) % 150 == 0: print(f" {i+1}/{len(cases)}") # warmth on a sample of answered replies answered_all = [r for r in results if r["behavior"] == "answer"] sample = answered_all[::max(1, len(answered_all) // WARMTH_SAMPLE)][:WARMTH_SAMPLE] print(f"Scoring warmth on {len(sample)} answers...") with ThreadPoolExecutor(max_workers=WORKERS) as ex: scores = list(ex.map(lambda r: (r, warmth_judge(r["answer"], r["q"])), sample)) warmth_vals = [(r, s["warmth"]) for r, s in scores if s] json.dump(results, open(os.path.join(ROOT, "eval", "sim_results.json"), "w"), indent=2, ensure_ascii=False) # ---- analysis ---- para = [r for r in results if r["kind"] == "paraphrase"] para_ans = [r for r in para if r["behavior"] == "answer"] para_gap = [r for r in para if r["behavior"] == "escalate"] retr_miss = [r for r in para if r["expected_retrieved"] is False] L = ["# Simulation report\n", f"Total questions: **{len(results)}**\n"] L.append("## 1. Paraphrased real questions (should be answerable)\n") L.append(f"- Total {len(para)} | answered {len(para_ans)} ({pct(len(para_ans),len(para))}) | " f"wrongly escalated {len(para_gap)} ({pct(len(para_gap),len(para))}) | " f"expected source missed {len(retr_miss)}\n") L.append("## 2. Scenario coverage (correct behavior per category)\n") for key, expect, _ in SCENARIO_CATS: rs = [r for r in results if r["kind"] == key] ok = [r for r in rs if r["behavior"] == r["expect"]] L.append(f"- {key:18s} expect `{expect}`: {len(ok)}/{len(rs)} correct ({pct(len(ok),len(rs))})") L.append("") if warmth_vals: avg = sum(w for _, w in warmth_vals) / len(warmth_vals) L.append(f"## 3. Warmth (coach-like tone), n={len(warmth_vals)}\n") L.append(f"- Average warmth: **{avg:.2f} / 5**") low = sorted(warmth_vals, key=lambda x: x[1])[:8] L.append(f"- Lowest-warmth answers:") for r, w in low: L.append(f" - [{w}/5] \"{r['q'][:60]}\"") L.append("") para_gap.sort(key=lambda r: -r["best_sim"]) L.append("## 4. Top answerable-but-escalated cases (fix these first)\n") for r in para_gap[:40]: L.append(f"- sim={r['best_sim']} | asked: \"{r['q'][:60]}\" | expected: \"{r['expected_q'][:55]}\"") open(os.path.join(ROOT, "eval", "sim_report.md"), "w").write("\n".join(L)) print("\n".join(L[:22])) print("\nFull report -> eval/sim_report.md") if __name__ == "__main__": main()