#!/usr/bin/env python3 """ The live demo: speech in -> retrieve -> read -> speech out, in 14 languages. python src/serve.py --langs hi,ta,bn --port 8000 python src/serve.py --no-voice # text only, no model downloads DESIGN NOTES THAT MATTER FOR THE SUBMISSION * EVERY STAGE IS TIMED SEPARATELY and returned in the response. A single end-to-end figure hides which part of the budget the voice layer spends, and ASR/TTS dominate: reporting one number would make the retrieval work look slow when it is the cheap part (search is ~0.1-0.2 ms/query, measured). * THE READER USES prior_weight=1.0. Measured on real bge-m3 cosine over 606 answerable queries: reader F1 0.1835 -> 0.2146 (+17%), top-1 passage accuracy 26.4% -> 49.3%. The oracle-reranker ceiling on the same subset is 0.302, so 0.088 F1 of reranking headroom is still on the table and is stated as such. * NO CONFORMAL ABSTENTION IS WIRED IN. The calibration produced tau = 0.9496 accepting 1 of 28,000 queries, i.e. no threshold reaches alpha at any usable coverage. Shipping it would mean abstaining on everything. The machinery is in router.py and switches on the moment a scorer earns it; claiming a guarantee we cannot honour would be worse than having none. * THE PAGE IS ONE SELF-CONTAINED FILE with no build step and no CDN, so the demo works on a pod with no outbound network. """ from __future__ import annotations import argparse import json import sys import time from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from src.reader import LexicalSpanReader # noqa: E402 from src.router import Passage # noqa: E402 from src.schema_utils import LANG_NAMES, default_root, pick_device # noqa: E402 from src.guardrails import GuardrailConfig, Guardrails # noqa: E402 from src.harness import AskRequest, Harness # noqa: E402 from src.voice import Timing, VoiceStack, read_wav # noqa: E402 # MODULE-LEVEL ON PURPOSE. `from __future__ import annotations` stringifies every # annotation, and FastAPI resolves `req: Request` by looking the name up in the # MODULE globals. Importing Request inside build_app() leaves it a local, the # name never resolves, FastAPI falls back to treating `req` as a query # parameter, and every POST returns 422 "Field required" while GET endpoints # keep working -- which is exactly the failure that is easiest to miss by hand # and is what tests/test_serve.py caught. try: from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse, JSONResponse except ImportError: # text-only / no fastapi FastAPI = Request = HTMLResponse = JSONResponse = None # type: ignore STATE: dict = {} # ------------------------------------------------------------------ retrieval class Index: """Per-(language, strategy) dense index held in RAM. Flat cosine over float16 -- at a few thousand chunks this is well under a millisecond and avoids a FAISS dependency the pod may not have. Several strategies are held at once so /api/compare can run the SAME query through each of them live. That is what turns the chunking study from a table into something a judge can poke at.""" def __init__(self, root: Path, langs: list[str], model: str, max_len: int, strategies: list[str] | None = None): import numpy as np import torch from src.evaluate_retrieval import Embedder self.np = np idx_dir = root / "index" man = json.loads((idx_dir / "manifest.json").read_text()) self.manifest = man self.default_strategy = man.get("default_strategy", "FW") self.device = pick_device() self.emb = Embedder(model or man["model"], self.device, 32, max_len or man["max_len"]) self.store: dict = {} self.langs: dict = {} for key, info in man.get("indices", {}).items(): lg, st = info["lang"], info["strategy"] if lg not in langs: continue if strategies and st not in strategies: continue vp, cp = idx_dir / f"{key}.vecs.npy", idx_dir / f"{key}.chunks.json" if not (vp.exists() and cp.exists()): print(f" !! missing files for {key}") continue payload = json.loads(cp.read_text()) self.store[(lg, st)] = {"vecs": np.load(vp).astype("float32"), "texts": payload["texts"], "meta": payload["meta"], "info": info} self.langs.setdefault(lg, []).append(st) print(f" {lg} {st:5s}: {len(payload['texts']):,} chunks " f"({info['chunks_per_doc']}/doc, {info['mean_chunk_words']} words)") for lg in self.langs: self.langs[lg].sort() def strategies_for(self, lang: str) -> list[str]: return self.langs.get(lang, []) def default_for(self, lang: str) -> str: """The strategy that SERVES this language. strategies_for() is sorted for stable display, so taking [0] would make the alphabetically-first strategy the server default -- adding DFC to the index for the comparison tab would silently switch what /api/ask uses. The manifest's declared default wins whenever it is actually loaded. """ avail = self.langs.get(lang, []) if self.default_strategy in avail: return self.default_strategy return avail[0] if avail else self.default_strategy def embed_query(self, query: str): return self.emb.encode([query]).cpu().numpy().astype("float32")[0] def search(self, query: str, lang: str, k: int = 5, strategy: str | None = None, qv=None): st = strategy or self.default_for(lang) d = self.store.get((lang, st)) if not d: return [] if qv is None: qv = self.embed_query(query) sims = d["vecs"] @ qv k = min(k, len(sims)) top = self.np.argpartition(-sims, k - 1)[:k] top = top[self.np.argsort(-sims[top])] return [Passage(d["meta"][i]["chunk_id"], d["texts"][i], float(sims[i]), lang) for i in top] def info(self, lang: str, strategy: str) -> dict: d = self.store.get((lang, strategy)) return dict(d["info"]) if d else {} # ------------------------------------------------------------------ pipeline def answer(query: str, lang: str, k: int = 5, want_audio: bool = True, strategy: str | None = None, audio_wav_b64: str = ""): """Thin adapter over the Harness. All orchestration -- retries, timeouts, guardrails, error recovery -- lives in harness.py (requirement 5), so there is exactly one code path and the demo cannot drift from what was measured.""" req = AskRequest(query=query, lang=lang, k=k, strategy=strategy, want_audio=want_audio, audio_wav_b64=audio_wav_b64) out = STATE["harness"].run(req).as_dict() out["abstention"] = ("guardrails active — see /guardrails; conformal " "abstention NOT enabled, see /about") return out def compare(query: str, lang: str, k: int = 5, budget_words: int = 400): """Run ONE query through every chunking strategy we hold for this language. Two views, because the whole finding is that they disagree: top-k the naive protocol. Confounded -- a strategy that emits more, smaller chunks gets more shots at the same passage, and corr(chunks_per_doc, nDCG@5) = -0.820 across our 7 strategies. equal budget as many chunks as fit in `budget_words`. Correcting for the confound drops that correlation to -0.152 AND INVERTS the ranking: FW is 1st on top-k and 5th on hit@400w. Both are shown side by side so the judge sees the correction happen on their own query rather than taking the table on trust. """ idx, reader = STATE["index"], STATE["reader"] strategies = idx.strategies_for(lang) if not strategies: return {"error": f"no indices for {lang}", "rows": []} t0 = time.perf_counter() qv = idx.embed_query(query) # embed ONCE, reuse for every strategy embed_ms = (time.perf_counter() - t0) * 1000 rows = [] for st in strategies: t0 = time.perf_counter() # Pull enough to fill the word budget even for large-chunk strategies. psgs = idx.search(query, lang, max(k, 12), st, qv=qv) search_ms = (time.perf_counter() - t0) * 1000 used, words = [], 0 for p in psgs: w = len(p.text.split()) if words + w > budget_words and used: break used.append(p) words += w t0 = time.perf_counter() span = reader.read(query, psgs[:k]) if psgs else None read_ms = (time.perf_counter() - t0) * 1000 info = idx.info(lang, st) rows.append({ "strategy": st, "n_chunks": info.get("n_chunks"), "chunks_per_doc": info.get("chunks_per_doc"), "mean_chunk_words": info.get("mean_chunk_words"), "index_mb": info.get("index_mb"), "top_score": round(psgs[0].score, 4) if psgs else 0.0, "answer": span.text[:300] if span else "", "confidence": round(float(span.score), 4) if span else 0.0, "search_ms": round(search_ms, 2), "read_ms": round(read_ms, 1), "topk": [{"score": round(p.score, 4), "words": len(p.text.split()), "text": p.text[:260]} for p in psgs[:k]], "budget": {"chunks_used": len(used), "words_used": words, "mean_score": round(sum(p.score for p in used)/max(1, len(used)), 4)}, }) by_topk = sorted(rows, key=lambda r: -r["top_score"]) by_budget = sorted(rows, key=lambda r: -r["budget"]["mean_score"]) return { "query": query, "lang": lang, "budget_words": budget_words, "embed_ms": round(embed_ms, 1), "rows": rows, "rank_topk": [r["strategy"] for r in by_topk], "rank_budget": [r["strategy"] for r in by_budget], "ranking_changed": [r["strategy"] for r in by_topk] != [r["strategy"] for r in by_budget], "measured_aggregate": MEASURED, } # Aggregate numbers from the full evaluation (hi/ta/bn, 400 queries/lang), shown # next to the live single-query result so nobody mistakes one query for evidence. MEASURED = { "note": "aggregate over 400 queries x 3 languages — the live panel above is ONE query", "top_k_protocol": {"FW": 0.7052, "FCC": 0.6834, "LCTS": 0.6705, "RC": 0.6677, "DFC": 0.6490, "PGC": 0.6253, "SGC": 0.6092}, "equal_budget_ndcg": {"FCC": 0.6686, "DFC": 0.6582, "LCTS": 0.6532, "FW": 0.6434, "SGC": 0.6245, "RC": 0.6192, "PGC": 0.5890}, "equal_budget_hit": {"DFC": 0.905, "SGC": 0.875, "FCC": 0.8675, "LCTS": 0.8258, "RC": 0.7658, "FW": 0.795, "PGC": 0.7583}, "confound": {"corr_chunks_vs_ndcg_topk": -0.820, "corr_chunks_vs_ndcg_budget": -0.152}, } # ------------------------------------------------------------------ http def build_app(cors: str = ""): if FastAPI is None: raise SystemExit("fastapi is not installed: pip install fastapi uvicorn") app = FastAPI(title="Voice RAG · MSMARCO-XI", docs_url="/docs") # The page is served from Vercel but the API lives here, so the browser # treats every /api/* call as cross-origin and blocks it without this. origins = [o.strip() for o in cors.split(",") if o.strip()] if origins: from fastapi.middleware.cors import CORSMiddleware app.add_middleware(CORSMiddleware, allow_origins=origins, allow_methods=["*"], allow_headers=["*"]) @app.get("/", response_class=HTMLResponse) def home(): return PAGE.replace("__LANGS__", json.dumps( [{"code": c, "name": LANG_NAMES.get(c, c)} for c in sorted(STATE["index"].langs)])) @app.get("/health") def health(): v, idx = STATE["voice"], STATE["index"] # VoiceStack is loaded tts_only=True, so v.asr_ok is False by # construction and reporting it says "not configured" next to a # configured key. The ASR provider hangs off the harness; ask the # thing that would actually do the transcribing. h = STATE.get("harness") asr = bool(getattr(h, "asr", None) and h.asr.ok) return {"ok": True, "languages": sorted(idx.langs), "strategies": {lg: idx.strategies_for(lg) for lg in sorted(idx.langs)}, "asr": asr, "asr_provider": getattr(getattr(h, "asr", None), "provider", ""), "tts": v.tts_ok, "notes": v.notes} @app.get("/about") def about(): return { "reader": "lexical span reader, prior_weight=1.0", "measured": { "retrieval_ndcg@5": 0.7052, "retrieval_hit@5": 0.9008, "reader_f1_no_prior": 0.1835, "reader_f1_with_prior": 0.2146, "reader_f1_oracle_rerank": 0.3023, "extraction_ceiling_f1": 0.7147, "answer_extractable_verbatim_pct": 11.1, }, "abstention": "NOT enabled. Conformal calibration gave tau=0.9496, " "which accepts 1 of 28,000 queries — no threshold " "reaches alpha=0.10 at any usable coverage.", "known_limits": [ "Answers average 19.2 words and are human-written sentences; " "only 11.1% exist verbatim in the passage, so an extractive " "reader is capped at F1 0.715 before any implementation error.", "0.088 F1 of reranking headroom remains unclaimed.", ], } @app.post("/api/ask") async def ask(req: Request): b = await req.json() q, lang = (b.get("query") or "").strip(), b.get("lang") or "hi" if not q: return JSONResponse({"error": "empty query"}, status_code=400) return answer(q, lang, int(b.get("k", 5)), bool(b.get("audio", True)), b.get("strategy")) @app.get("/guardrails") def guardrails_ep(): """Requirement 6, documented and inspectable.""" return STATE["guards"].describe() @app.get("/harness") def harness_ep(): """Requirement 5: the declared stage graph, with each stage's timeout, retry policy and failure behaviour.""" h = STATE["harness"] return {"budget_ms": h.budget_ms, "asr_provider": getattr(h.asr, "provider", None), "asr_ready": bool(h.asr and h.asr.ok), "stages": [{"name": st.name, "timeout_s": st.timeout_s, "retries": st.retries, "on_error": st.on_error.value} for st in h.stages()]} @app.post("/api/compare") async def compare_ep(req: Request): b = await req.json() q, lang = (b.get("query") or "").strip(), b.get("lang") or "hi" if not q: return JSONResponse({"error": "empty query"}, status_code=400) return compare(q, lang, int(b.get("k", 5)), int(b.get("budget_words", 400))) @app.post("/api/normalise") async def normalise_ep(req: Request): """Show the Indic normaliser bug on text the judge types. A table saying "we fixed a Unicode bug" is a claim. Watching your own sentence shatter into consonant fragments under the standard SQuAD normaliser, and stay intact under ours, is evidence.""" import re import unicodedata from src.textnorm import normalise as fixed b = await req.json() text = (b.get("text") or "").strip() if not text: return JSONResponse({"error": "empty text"}, status_code=400) # The standard formulation, verbatim, as used across the QA literature. broken = re.sub(r"\s+", " ", re.sub(r"[^\w\s]", " ", unicodedata.normalize("NFKC", text).lower())).strip() good = fixed(text, True) marks = sum(1 for c in text if unicodedata.category(c)[0] == "M") return { "input": text, "input_words": len(text.split()), "standard_normaliser": {"regex": r"[^\w\s]", "output": broken, "tokens": len(broken.split())}, "ours": {"method": "Unicode category: P*/S* -> space, Cf -> deleted, " "L*/N*/M* kept", "output": good, "tokens": len(good.split())}, "combining_marks_in_input": marks, "inflation": round(len(broken.split()) / max(1, len(text.split())), 2), "why": "Python re \\w is str.isalnum(), which is False for Unicode " "Mn/Mc -- the combining marks carrying the vowels in every " "Brahmic script. They are deleted AND replaced with a space, " "so each word also splits at every mark.", } @app.post("/api/voice") async def voice_ep(req: Request): """Speech in. Transcription is a harness stage (Sarvam/ElevenLabs, with retries), so a provider hiccup degrades exactly like any other stage.""" b = await req.json() if not b.get("audio_wav_b64"): return JSONResponse({"error": "no audio"}, status_code=400) return answer("", b.get("lang") or "hi", int(b.get("k", 5)), True, b.get("strategy"), b["audio_wav_b64"]) return app PAGE = r""" Voice RAG · 14 Indic languages

Voice RAG · 14 Indic languages

MSMARCO-XI · bge-m3 dense retrieval · extractive reader · measured numbers & known limits
Ask
Chunking, live
The Indic bug
The same query, run through every chunking strategy right now. Top-k is the protocol everyone uses; it is confounded, because a strategy emitting more, smaller chunks gets more shots at the same passage (corr with chunks/doc = −0.820). Equal budget gives every strategy the same 400 words of context — correlation drops to −0.152 and the ranking inverts.
Every token-level metric in QA runs text through re.sub(r"[^\w\s]", " ", s). In Python, \w is str.isalnum(), which is False for Unicode Mn/Mc — the combining marks carrying the vowels in every Brahmic script. Paste any Indic sentence and watch it shatter.
""" def parser() -> argparse.ArgumentParser: ap = argparse.ArgumentParser() ap.add_argument("--root", type=Path, default=None) ap.add_argument("--langs", default=None) ap.add_argument("--host", default="0.0.0.0") ap.add_argument("--port", type=int, default=8000) ap.add_argument("--k", type=int, default=5) ap.add_argument("--model", default=None) ap.add_argument("--max-len", type=int, default=0) ap.add_argument("--answer-mode", default="sentence", choices=["span", "sentence"], help="sentence = speakable (default for serving)") ap.add_argument("--no-voice", action="store_true") ap.add_argument("--asr-provider", default=None, choices=[None, "sarvam", "elevenlabs"], help="requirement 1; defaults to $VOICERAG_ASR") ap.add_argument("--prior-weight", type=float, default=1.0, help="measured best on real bge-m3 scores; 0 disables") ap.add_argument("--cors", default="", help="origins allowed to call /api/* — the Vercel page. " "Comma-separated. Empty = same-origin only.") ap.add_argument("--prewarm-tts", default="", help="langs to warm, or 'all'. MMS voices are lazy AND " "downloaded on first use; a cold request once cost " "23,511 ms inside speak alone.") return ap def bootstrap(args): """Everything main() does except bind a port. Hugging Face's Docker SDK is a paid feature, so the Space runs the Gradio base image and owns its own uvicorn. app.py calls this and mounts the result; nothing below the HTTP layer knows the difference. """ root = args.root.expanduser().resolve() if args.root else default_root() man = root / "index" / "manifest.json" if not man.exists(): raise SystemExit(f"no index at {root/'index'} — run src/index_build.py first") # index_build.py writes indices/model/strategies -- never a "languages" # key. Reading one raised KeyError for every caller that omitted --langs. manifest = json.loads(man.read_text()) langs = args.langs.split(",") if args.langs else sorted( {v["lang"] for v in manifest.get("indices", {}).values()}) print(f"==> loading index from {root/'index'}") STATE["index"] = Index(root, langs, args.model, args.max_len) if not STATE["index"].langs: raise SystemExit("no languages loaded — build the index first") # answer_mode="sentence" for SERVING: the answer is spoken aloud, and a # trimmed span starts mid-clause and stops mid-number. Still fully # extractive, so guardrail gate 4 (verbatim substring) is unaffected. STATE["reader"] = LexicalSpanReader(prior_weight=args.prior_weight, answer_mode=args.answer_mode) print(f"==> reader prior_weight={args.prior_weight} answer_mode={args.answer_mode}") # REQUIREMENT 1: speech-to-text must be Sarvam or ElevenLabs. No local # fallback -- a stand-in that quietly works would be a compliance failure # dressed up as a success, so this reports NOT READY and the voice path # returns a clear error instead. from src.asr_api import CloudASR asr = None if args.no_voice else CloudASR(args.asr_provider) if asr: print(f"==> ASR: {asr.why}") if not asr.ok: print(" !! voice input will return an error until the key is set") print("==> loading TTS (MMS — the brief specifies the speech-to-TEXT provider only)") STATE["voice"] = VoiceStack.load(enable=not args.no_voice, tts_only=True) for n in STATE["voice"].notes: print(f" {n}") if args.prewarm_tts and STATE["voice"].tts_ok: warm = (sorted(STATE["index"].langs) if args.prewarm_tts == "all" else [x.strip() for x in args.prewarm_tts.split(",") if x.strip()]) for lg in warm: t0 = time.time() try: STATE["voice"].tts.speak("नमस्ते", lg) print(f" warmed {lg} in {(time.time() - t0) * 1000:.0f} ms") except Exception as e: print(f" !! {lg} has no MMS voice or failed: {str(e)[:80]}") # GuardrailConfig.load is the single source for gate 3: it reads the signal # and its threshold from the same file in the same call, and refuses a # threshold written on the other signal's scale. tp = root / "results" / "guardrail_calibration.json" cfg = GuardrailConfig.load(tp) depth = f" over top {cfg.gate_depth}" if cfg.topic_signal == "spread" else "" if cfg.topic_calibrated: print(f"==> guardrails: gate 3 = {cfg.topic_signal} < {cfg.tau_topic}{depth}") print(f" {cfg.topic_provenance}") else: print(f"==> guardrails: gate 3 = {cfg.topic_signal} < {cfg.tau_topic}{depth}") print(f" !! PLACEHOLDER — no {tp.name}; rebuild the index to calibrate") STATE["guards"] = Guardrails(cfg) STATE["harness"] = Harness(STATE["index"], STATE["reader"], STATE["voice"], asr=asr, guards=STATE["guards"], budget_ms=200.0) return build_app(args.cors) def main() -> int: # Imported after parsing so --help works without the serving deps # installed, which is how this file gets checked on a laptop. args = parser().parse_args() import uvicorn app = bootstrap(args) print(f"\n==> http://{args.host}:{args.port} " f"({len(STATE['index'].langs)} languages)") uvicorn.run(app, host=args.host, port=args.port, log_level="warning") return 0 if __name__ == "__main__": raise SystemExit(main())