""" QuitSure AI Support — API service. One stateless endpoint the app frontend calls. Given a user message (+ optional history), returns an answer grounded ONLY in the curated knowledge base, plus an escalation decision so the devs know when to hand the chat to a human coach (in Freshchat). Never connects to Freshchat. Never learns from coach transcripts (Ram's call). Design rationale: docs/DESIGN.md Frontend contract: docs/API.md Run: python3 scripts/build_kb.py && python3 scripts/build_index.py # one-time / on KB change uvicorn app:app --port 8082 A simple test chat UI is served at http://localhost:8082/ """ from __future__ import annotations import json import math import os import re import uuid from datetime import datetime, timezone from typing import Optional import numpy as np from dotenv import load_dotenv from fastapi import Depends, FastAPI, Header, HTTPException from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel from google import genai from google.genai import types import db ROOT = os.path.dirname(os.path.abspath(__file__)) load_dotenv(os.path.join(ROOT, ".env")) ANSWER_MODEL = os.getenv("GEMINI_MODEL", "gemini-2.5-flash") EMBED_MODEL = os.getenv("GEMINI_EMBED_MODEL", "gemini-embedding-001") DIM = 1536 TOP_K = 5 # The app passes an opaque program code, but approved answers refer to programs by NAME # ("Original" / "Relaxed"). Without this mapping the model can't tell that a P9 user and a # "Relaxed program" answer are the same program, and wrongly escalates. PROGRAM_NAMES = { "P3": "QuitSure Original, the 6-day intensive program (users quit on Day 6)", "P9": "QuitSure Relaxed, the 42-day gradual program (users quit on Day 42)", "P11": "Leansure, the fitness program", } def canonical_program(code): """The KB tags smoking content as either Original (P3) or Relaxed (P9). P9/LTP is the ONLY 42-day Relaxed program; every other smoking program (P1, P2, P4, P5, P7...) is a 6-day Original-style program, so it maps to P3 for both retrieval filtering and naming. P11 is Leansure (separate). Keeps a P1/P5 user from seeing none of the Original answers.""" if not code: return None c = code.upper() if c in ("P9", "P11"): return c return "P3" # Below this cosine similarity we ABSTAIN (don't answer, escalate). Calibrate on real # queries — see docs/DESIGN.md §6/§9. Sane default until we measure. RETRIEVAL_FLOOR = float(os.getenv("RETRIEVAL_FLOOR", "0.55")) DATA = os.path.join(ROOT, "data") # 20s network timeout so a slow Gemini call can't hang a request client = genai.Client(api_key=os.environ["GEMINI_API_KEY"], http_options=types.HttpOptions(timeout=20000)) # ---- deterministic medical safety filter (docs/DESIGN.md safety) ---- # Conservative: forces a human handoff for the clearest high-risk cases regardless of what the # knowledge base matched, because a prompt rule cannot reliably override a strong KB match. # Extend the boundary once Ram signs off. # The AI answers everything Ram's approved doc covers, and escalates what it doesn't. The one hard, # non-negotiable safety line: self-harm / suicide always goes to a human, regardless of any content. _CRISIS = re.compile( # English r"suicid|kill(?:ing)? myself|end(?:ing)? my life|hurt(?:ing)? myself|" r"harm(?:ing)? myself|self.?harm|want to die|wanna die|don.?t want to live|" r"no reason to live|end it all|take my (?:own )?life|" # Hinglish (romanized Hindi) — bias toward catching; a false escalation is cheap, a miss is not r"khud ?kushi|khud.?khushi|aatmahatya|atmahatya|marna chaht|mar jana chaht|" r"jeena nahi|jina nahi|jeene ka mann nahi|mujhe (?:ab )?jeena nahi|ab nahi jeena|" r"khud ?ko ?khatam|khud ?ko ?maar|khud ?ko ?nuksan|jaan de ?d|jaan deni|zindagi khatam|" # Devanagari r"आत्महत्या|खुदकुशी|मरना चाहत|जीना नहीं|खुद को खत्म|खुद को मार|जान दे|जान दूं", re.I) def is_crisis(text): # "I smoke to hurt myself" is a smoking rationalization (an approved deception entry), not # self-harm intent. Don't hijack it to crisis. if re.search(r"smok\w*\s+(?:to|and)\s+hurt myself", text, re.I): return False return bool(_CRISIS.search(text)) # ---- load KB + embedding index at startup ---- KB = {e["id"]: e for e in json.load(open(os.path.join(DATA, "kb.json")))} _idx = np.load(os.path.join(DATA, "index.npz")) INDEX_VECS = _idx["vectors"] # (N, DIM), normalized INDEX_IDS = _idx["ids"] # (N,) # program/platform aligned to INDEX_IDS, for context-aware filtering INDEX_PROGRAMS = [KB[int(i)]["programs"] for i in INDEX_IDS] INDEX_PLATFORM = [KB[int(i)].get("platform", "all") for i in INDEX_IDS] # --- BM25 lexical index (aligned to INDEX_IDS). Dense embeddings blur exact tokens # ("555", "OTP", "Champix", shortcut codes); BM25 matches them literally. Fused with cosine # at query time via reciprocal rank fusion so a match found either way surfaces into the top-k. _TOKEN_RE = re.compile(r"[a-z0-9]+") def _tok(text): return _TOKEN_RE.findall(text.lower()) def _build_bm25(): docs = [_tok(" ".join([e["question"], e.get("topic", ""), e.get("tags", ""), e["answer"]])) for e in (KB[int(i)] for i in INDEX_IDS)] n = len(docs) dl = np.array([len(d) for d in docs], dtype=np.float32) avgdl = float(dl.mean()) if n else 0.0 df = {} for d in docs: for t in set(d): df[t] = df.get(t, 0) + 1 idf = {t: math.log(1 + (n - c + 0.5) / (c + 0.5)) for t, c in df.items()} tf = [{} for _ in docs] for j, d in enumerate(docs): for t in d: tf[j][t] = tf[j].get(t, 0) + 1 return {"tf": tf, "idf": idf, "dl": dl, "avgdl": avgdl} _BM25 = _build_bm25() def bm25_scores(query, k1=1.5, b=0.75): """Okapi BM25 score of the query against every indexed doc (aligned to INDEX_IDS).""" q = set(_tok(query)) tf, idf, dl, avgdl = _BM25["tf"], _BM25["idf"], _BM25["dl"], _BM25["avgdl"] scores = np.zeros(len(tf), dtype=np.float32) for j in range(len(tf)): norm = k1 * (1 - b + b * (dl[j] / avgdl if avgdl else 0.0)) s = 0.0 for t in q: f = tf[j].get(t) if f: s += idf.get(t, 0.0) * (f * (k1 + 1)) / (f + norm) scores[j] = s return scores SYSTEM_RULES = """You are the QuitSure Assistant, an automated support assistant inside the QuitSure smoking-cessation app. You are NOT a human and NOT a named coach, never claim to be. Never output placeholders or merge-fields (e.g. %name%, ____). Never use em-dashes or en-dashes in your replies. Use commas, periods, or parentheses instead. Your ONLY job is to help with: the QuitSure app and programs, the user's quit-smoking journey, and their subscription / billing / account. Nothing else. TONE: Be warm, encouraging, and human, like a caring quit-smoking coach who is genuinely on the user's side. Acknowledge how they feel, celebrate their effort, and be patient and reassuring. Sound friendly and natural, never robotic, curt, or clinical. You may rephrase for warmth, but never add any fact, step, price, or policy that is not in the approved answers. Even for simple or factual questions (how to log out, gift cards, a tech step), keep a warm human touch, a short kind opener or closer. Never give a cold one-line answer. LANGUAGE: Reply in the SAME language and script the user writes in. If they write in Hinglish (romanized Hindi mixed with English, in Latin script), reply in that same natural romanized Hinglish style, NOT in pure English and NOT in Devanagari Hindi. If they write in Devanagari Hindi, reply in Hindi. Keep the exact approved facts, steps, and links, just expressed in their language. Answer ONLY from the APPROVED ANSWERS provided below. Match their warm, supportive tone. Never invent facts, prices, steps, or policies that are not in the approved answers. Decide in this order: 1. OFF-TOPIC / time-pass: if the message is NOT about QuitSure, quitting smoking, the app, or the user's account (e.g. general knowledge, news, sports, math, coding, jokes, weather, recipes, casual chit-chat), do NOT answer it. Give a brief, friendly one-line redirect to what you can help with. -> escalate=false, escalate_reason="out_of_scope". Do NOT send these to a coach. 2. Otherwise set escalate=true when ANY of these is true: - It IS about QuitSure but the approved answers don't contain the info -> "no_confident_answer" - Medical/clinical questions -> "medical_topic", UNLESS the approved answers directly cover the question, in which case answer from that approved content (e.g. an approved answer about using Champix or nicotine substitutes with the program). Escalate a medical question only when the approved answers do NOT cover it (e.g. personal dosing or a drug interaction we have no answer for). Normal quit-smoking experiences (irritability, cravings, sleep, mood, appetite) are not medical and may be answered warmly. - Self-harm, suicidal thoughts, or crisis -> "crisis" - The user explicitly asks for a human / coach / agent -> "user_requested" - The user is clearly frustrated or angry -> "user_frustrated" - Needs a human to look up or fix THIS user's specific account state: charged but subscription not active, "I paid but no access", "where is my refund" / refund requested but not received, a double charge -> "account_action". (General billing questions, and how to REQUEST a refund or "I want a refund", are ANSWERED from the approved refund process, not escalated. Only a refund STATUS problem that needs a human to check the account is escalated.) 3. Otherwise, answer directly from the approved answers. ANSWER BEFORE YOU ESCALATE: if the approved answers contain concrete steps that would resolve the user's issue, GIVE those steps rather than escalating, even if the message also sounds discouraged, upset, or emotional. Many complaints that sound emotional are really technical/account issues with a fix in the approved answers (e.g. "I lost all my progress", "my days reset", "everything is gone", "it logged me out") -> give the recovery steps first (you may briefly acknowledge the feeling), and only escalate if the steps do not cover their situation. Questions about how QuitSure works, how effective it is, or how it compares to willpower, nicotine patches, or other methods ARE covered by the approved answers (how it works, how it helps, how it is different). Answer these from that content, do not escalate them. CONTINUITY: Use the whole conversation, including your own previous replies. If the user sends a short reply that continues the current topic (e.g. "yes", "sure", "help me", "ok", "tell me more", "which one") or accepts an offer you just made, keep helping using what has already been discussed and the approved answers. Do NOT escalate a simple continuation, and never offer help and then hand off instead of giving it. Only escalate when the user genuinely needs information not covered. If the user asks which option or program suits them or asks you to help them decide, HELP them like a coach: briefly ask about the concrete factors that actually decide it (how much time they can give each day, how soon they want to quit), then RECOMMEND the program that fits and say why, using the approved answers. Do not just ask them to pick a pace or bounce the choice back to them. That is grounded help, not a reason to escalate. MULTI-PART QUESTIONS: If the user asks several things at once, answer every part you can from the approved answers. Escalate ONLY if a specific part genuinely needs a human (an account lookup or a medical topic), and even then, answer the other parts first and tell the user which one part you are passing to the team. Never escalate the whole message just because one part needs a human, and never invent an answer for a part the approved answers do not cover. USER CONTEXT: If a USER CONTEXT block is provided (the user's program, current day, subscription status), treat those facts as true and use them to answer questions about the user's own account or progress directly, e.g. "am I subscribed" -> use subscription status, "what day am I on" -> use the day. Only escalate account PROBLEMS that need a human to fix or look up (a specific charge, a refund to process, a payment not reflecting), not simple status questions you already have the answer to. IMPORTANT: Whatever the approved answers cover, you SHOULD answer, including medical questions the approved answers address (e.g. using Champix or nicotine substitutes with the program, common symptoms the doc explains). Escalate only what the approved answers do NOT cover, plus these always-escalate cases even if related content exists: self-harm or crisis, an explicit request for a human, and account-specific payment/refund problems for this user. If the user greets you (hi/hello), thanks you, or makes small talk, reply warmly in one line as the QuitSure Assistant and invite their question, and set escalate=false with escalate_reason "greeting". Do NOT mention coaches or escalation unprompted, only at the moment you actually need to hand off. When escalating: a short warm handoff line as "answer" (e.g. "Let me get a member of our support team to help you with this.") plus a one-sentence "coach_summary". For "crisis", gently encourage immediate help and that someone from the team is coming. In "used_ids", list the id number(s) of the APPROVED ANSWERS you actually drew on to write your answer (from the "[id N | ...]" labels). Leave it empty for greetings, off-topic declines, and escalations where you used none. Respond ONLY as JSON: {"answer": str, "escalate": bool, "escalate_reason": str|null, "coach_summary": str|null, "used_ids": int[]} """ RESPONSE_SCHEMA = { "type": "object", "properties": { "answer": {"type": "string"}, "escalate": {"type": "boolean"}, "escalate_reason": {"type": "string", "nullable": True}, "coach_summary": {"type": "string", "nullable": True}, "used_ids": {"type": "array", "items": {"type": "integer"}, "nullable": True}, }, "required": ["answer", "escalate"], } app = FastAPI(title="QuitSure AI Support") # Where the app mounts. Like discovery: the test/prod Apache proxy STRIPS the # /support-ai prefix before forwarding, so we mount at "/" by default. If a proxy # ever forwards the prefix through instead, set MOUNT_PREFIX=/support-ai. MOUNT_PREFIX = os.environ.get("MOUNT_PREFIX", "") # Shared-secret auth. If API_KEY is set, callers must send it as the X-API-Key header. # Left empty locally (open) so dev/testing needs no key; set it on the test/prod server. API_KEY = os.environ.get("API_KEY", "") def require_key(x_api_key: str = Header(default="", alias="X-API-Key")): if API_KEY and x_api_key != API_KEY: raise HTTPException(status_code=401, detail="invalid or missing API key") class Turn(BaseModel): role: str # "user" | "assistant" text: str class ChatRequest(BaseModel): user_id: Optional[str] = None message: str history: list[Turn] = [] program: Optional[str] = None # "P3" | "P9" | "P11" ... gates program-specific answers platform: Optional[str] = None # "ios" | "android" gates platform-specific answers locale: Optional[str] = None # e.g. "hi", "es"; hint for reply language (auto-detected otherwise) program_day: Optional[int] = None # current day/stage in the program (for stage-specific answers) subscription_status: Optional[str] = None # e.g. "active" | "expired" | "trial" class Feedback(BaseModel): user_id: Optional[str] = None message_id: Optional[str] = None # the id returned by /chat, to link feedback to a reply message: str answer: str rating: str # "up" | "down" note: Optional[str] = None def _now(): return datetime.now(timezone.utc).isoformat() def embed_query(text): resp = client.models.embed_content( model=EMBED_MODEL, contents=[text], config=types.EmbedContentConfig(task_type="RETRIEVAL_QUERY", output_dimensionality=DIM), ) v = np.asarray(resp.embeddings[0].values, dtype=np.float32) n = np.linalg.norm(v) return v / n if n else v def retrieval_query(req): """Anchor retrieval to the recent context so short follow-ups ('sure', 'which one?') retrieve well.""" recent_user = [t.text for t in req.history if t.role == "user"][-2:] last_assistant = [t.text[:200] for t in req.history if t.role == "assistant"][-1:] return " ".join(recent_user + last_assistant + [req.message]) def retrieve(query, program=None, platform=None, k=TOP_K): program = canonical_program(program) # P1/P2/P4/P5/P7 -> P3 (all 6-day Original-style) qv = embed_query(query) sims = INDEX_VECS @ qv # cosine — vectors are normalized # context filter: keep "all"-program entries + entries matching the user's program; # likewise drop platform-specific answers that don't match the user's platform. allowed = [] for i in range(len(sims)): if program and "all" not in INDEX_PROGRAMS[i] and program not in INDEX_PROGRAMS[i]: continue if platform and INDEX_PLATFORM[i] != "all" and INDEX_PLATFORM[i] != platform: continue allowed.append(i) if not allowed: return [], 0.0 # Hybrid retrieval: fuse the dense (cosine) and lexical (BM25) rankings with reciprocal rank # fusion, so exact-token matches surface even when the embedding blurs them (and vice versa). bm = bm25_scores(query) cos_rank = sorted(allowed, key=lambda i: -sims[i]) bm_rank = sorted((i for i in allowed if bm[i] > 0), key=lambda i: -bm[i]) rrf_k = 60 fused = {i: 0.0 for i in allowed} for r, i in enumerate(cos_rank): fused[i] += 1.0 / (rrf_k + r) for r, i in enumerate(bm_rank): fused[i] += 1.0 / (rrf_k + r) top = sorted(allowed, key=lambda i: -fused[i])[:k] hits = [(KB[int(INDEX_IDS[i])], float(sims[i])) for i in top] # Abstention signal stays semantic: the max cosine over in-scope entries (unchanged from before, # so the escalation floor keeps its calibration; BM25 only reshuffles which entries reach the model). best = float(max(sims[i] for i in allowed)) return hits, best def build_context(hits): return "\n\n".join( f"[id {e['id']} | {e['topic'] or e['source']}]\nQ: {e['question']}\nA: {e['answer']}" for e, _ in hits ) def _finalize(out): """Add a message id (for feedback linking) and a suggested channel; cap coach_summary.""" out["message_id"] = uuid.uuid4().hex[:12] if not out.get("escalate"): out["suggested_channel"] = None else: out["suggested_channel"] = ("technical" if out.get("escalate_reason") in ("account_action", "service_error") else "coaching") if out.get("coach_summary"): out["coach_summary"] = out["coach_summary"][:300] return out @app.get(f"{MOUNT_PREFIX}/health") def health(): return {"status": "ok", "kb_entries": len(KB), "model": ANSWER_MODEL, "retrieval_floor": RETRIEVAL_FLOOR} @app.post(f"{MOUNT_PREFIX}/chat", dependencies=[Depends(require_key)]) def chat(req: ChatRequest): # Hard safety line: self-harm / suicide always goes to a human, regardless of any doc content. if is_crisis(req.message): out = _finalize({"answer": "I'm really glad you reached out, and you don't have to go through " "this alone. Let me connect you with a member of our team right now. " "If you need immediate support, please reach out to a helpline. In India " "you can call Tele-MANAS at 14416 (free, 24x7) or AASRA at +91 9820466726. " "If you are outside India, please contact your local emergency number or a " "nearby suicide-prevention helpline. You matter, and help is available.", "escalate": True, "escalate_reason": "crisis", "coach_summary": f"CRISIS / self-harm signal: {req.message}", "sources": []}) db.log_chat(message_id=out["message_id"], ts=_now(), user_id=req.user_id, program=req.program, platform=req.platform, message=req.message, gate="crisis_filter", escalate=1, escalate_reason="crisis", answer=out["answer"]) return out try: # No program passed -> assume Original (6-day), the majority of users. The app SHOULD send # the real program; this only keeps program-specific answers from silently defaulting to Relaxed. eff_program = req.program or "P3" rq = retrieval_query(req) hits, best_sim = retrieve(rq, program=eff_program, platform=req.platform) # A self-contained question after an unrelated turn shouldn't be diluted by stale history. # Retrieve on the bare message too and keep whichever ranks higher: topic-switches win via # the bare message, deictic follow-ups ("which one?") still win via the blended query. if req.history: hits_m, best_m = retrieve(req.message, program=eff_program, platform=req.platform) if best_m > best_sim: hits, best_sim, rq = hits_m, best_m, req.message # optional per-user context the app can pass, injected so answers can be stage/status-aware ctx = [] if req.locale: ctx.append(f"User's app language: {req.locale} (reply in this language).") pname = PROGRAM_NAMES.get(canonical_program(eff_program), eff_program) ctx.append(f"User's program: {pname}. An approved answer written for this program by name " f"applies to this user.") if req.program_day is not None: ctx.append(f"User is on day {req.program_day} of the program.") if req.subscription_status: ctx.append(f"User's subscription: {req.subscription_status}.") ctx_str = ("USER CONTEXT:\n" + "\n".join(ctx) + "\n\n") if ctx else "" # Generate, grounded in retrieved approved answers. The model classifies off-topic vs # escalate vs answer (it tells "off-topic chit-chat" from "real QuitSure question we lack"). convo = "".join(f"{t.role.upper()}: {t.text}\n" for t in req.history[-6:]) + f"USER: {req.message}" prompt = f"{ctx_str}APPROVED ANSWERS:\n{build_context(hits)}\n\n---\nCONVERSATION:\n{convo}" resp = client.models.generate_content( model=ANSWER_MODEL, contents=prompt, config=types.GenerateContentConfig( system_instruction=SYSTEM_RULES, response_mime_type="application/json", response_schema=RESPONSE_SCHEMA, temperature=0.2, ), ) out = json.loads(resp.text) out.setdefault("escalate_reason", None) out.setdefault("coach_summary", None) # off-topic is a polite decline, never a coach handoff if out.get("escalate_reason") == "out_of_scope": out["escalate"] = False # Backstop — the model gave a substantive QuitSure answer on weak retrieval. That's a # hallucination risk, so override to a coach handoff. Greetings/small talk and off-topic # declines legitimately have no KB match, so exempt them. exempt = ("out_of_scope", "greeting") answered = not out["escalate"] and out.get("escalate_reason") not in exempt if answered and best_sim < RETRIEVAL_FLOOR + 0.05: out.update(escalate=True, escalate_reason="no_confident_answer", answer="Let me get a member of our support team to help you with this.", coach_summary=f"User asked: {req.message}") # sources = the entries the model actually cited (used_ids), not raw retrieval rank. # Fall back to the single top hit if the model cited nothing usable. if not out["escalate"] and out.get("escalate_reason") not in exempt: above = {e["id"]: e for e, s in hits if s > RETRIEVAL_FLOOR} picked = [above[i] for i in (out.get("used_ids") or []) if i in above] if not picked and above: picked = [next(iter(above.values()))] out["sources"] = [f"{e['source']} / {e['question']}" for e in picked] else: out["sources"] = [] out = _finalize(out) db.log_chat(message_id=out["message_id"], ts=_now(), user_id=req.user_id, program=req.program, platform=req.platform, message=req.message, retrieval_query=rq, best_sim=best_sim, gate="model", escalate=int(out["escalate"]), escalate_reason=out["escalate_reason"], answer=out["answer"]) return out except Exception as e: # Never crash in front of a user: fail into a human handoff. out = _finalize({"answer": "I'm having a bit of trouble right now, so let me get someone from " "our team to help you.", "escalate": True, "escalate_reason": "service_error", "coach_summary": f"Service error while handling: {req.message}", "sources": []}) db.log_chat(message_id=out["message_id"], ts=_now(), user_id=req.user_id, program=req.program, platform=req.platform, message=req.message, gate="error", escalate=1, escalate_reason="service_error", answer=out["answer"]) return out @app.post(f"{MOUNT_PREFIX}/feedback", dependencies=[Depends(require_key)]) def feedback(fb: Feedback): db.log_feedback(_now(), fb.message_id, fb.user_id, fb.rating, fb.note) return {"status": "ok"} @app.get(f"{MOUNT_PREFIX}/stats", dependencies=[Depends(require_key)]) def stats(): return db.get_stats() # ---- test UI (served last so /chat etc. take precedence) ---- app.mount(f"{MOUNT_PREFIX}/static", StaticFiles(directory=os.path.join(ROOT, "static")), name="static") @app.get(MOUNT_PREFIX or "/") def home(): return FileResponse(os.path.join(ROOT, "static", "index.html")) @app.get(f"{MOUNT_PREFIX}/analytics") def analytics(): return FileResponse(os.path.join(ROOT, "static", "analytics.html")) if __name__ == "__main__": import uvicorn uvicorn.run("app:app", host="0.0.0.0", port=8082)