""" PolicyQA — Document QA Agent Upload a policy document or report (PDF or plain text). Ask questions in natural language. Get answers cited to exact sections. Pipeline: Upload → PDF extract → section-aware chunking → BM25 + BGE-base dense + RRF fusion → BGE-reranker-large (cross-encoder) → flan-t5-base generates answer (250MB) → flan-t5-large judges (score >= 6) (770MB — heavier than generator) → Answer + "Section X · Para Y" citations """ # ── Patch gradio_client BEFORE importing gradio ──────────────────────────────── # gradio-client 1.3.0 has a bug where schema can be a bool and # `if "X" in schema:` raises TypeError. Fix all occurrences first. import pathlib, re as _re, importlib _gc_path = None try: import gradio_client.utils as _gcu _gc_path = pathlib.Path(_gcu.__file__) _src = _gc_path.read_text() _fixed = _re.sub( r'\bif\s+"(\w+)"\s+in\s+schema\s*:', r'if isinstance(schema, dict) and "\1" in schema:', _src ) if _fixed != _src: _gc_path.write_text(_fixed) importlib.reload(_gcu) print(f"Patched {_gc_path.name} ✓") else: print("gradio_client already clean or pattern not found") except Exception as _e: print(f"gradio_client patch error: {_e}") # ────────────────────────────────────────────────────────────────────────────── import re import io import fitz # pymupdf — PDF text extraction import faiss import numpy as np import gradio as gr from dataclasses import dataclass, field from typing import Optional from rank_bm25 import BM25Okapi from sentence_transformers import SentenceTransformer, CrossEncoder from transformers import AutoTokenizer, AutoModelForSeq2SeqLM import torch # ── Models ───────────────────────────────────────────────────────────────────── EMBED_MODEL = "BAAI/bge-base-en-v1.5" RERANKER_MODEL = "BAAI/bge-reranker-large" GEN_MODEL = "google/flan-t5-base" # generator ~250MB JUDGE_MODEL = "google/flan-t5-large" # judge ~770MB (heavier than generator) DEVICE = "cuda" if torch.cuda.is_available() else "cpu" DTYPE = torch.float16 if DEVICE == "cuda" else torch.float32 print(f"Device: {DEVICE}") print("Loading embedder …") embedder = SentenceTransformer(EMBED_MODEL, device=DEVICE) print("Loading reranker …") reranker = CrossEncoder(RERANKER_MODEL, device=DEVICE) print("Loading generator (flan-t5-base) …") gen_tok = AutoTokenizer.from_pretrained(GEN_MODEL) gen_model = AutoModelForSeq2SeqLM.from_pretrained(GEN_MODEL, torch_dtype=DTYPE).to(DEVICE) print("Loading judge (flan-t5-large) …") judge_tok = AutoTokenizer.from_pretrained(JUDGE_MODEL) judge_model = AutoModelForSeq2SeqLM.from_pretrained(JUDGE_MODEL, torch_dtype=DTYPE).to(DEVICE) print("All models ready.") # ── Tunables ─────────────────────────────────────────────────────────────────── CHUNK_CHARS = 600 # characters per chunk CHUNK_OVERLAP = 100 # overlap between consecutive chunks in same section DENSE_TOPK = 20 BM25_TOPK = 20 RRF_K = 60 RERANK_TOPK = 5 RRF_GATE = 0.30 # min normalised RRF score — below → escalate JUDGE_PASS = 6 # judge score threshold out of 10 GEN_MAX_TOKENS = 350 JUDGE_MAX_TOKENS = 80 # ── Data structures ──────────────────────────────────────────────────────────── @dataclass class Chunk: text: str section_path: str # e.g. "3 · Risk Assessment > 3.2 · Identification" section_index: int # global section counter (for stable citation numbering) para_index: int # paragraph index within the section char_start: int char_end: int @dataclass class DocState: doc_name: str = "" chunks: list = field(default_factory=list) faiss_index: Optional[object] = None bm25_index: Optional[object] = None toc: list[str] = field(default_factory=list) # section headings in order history: list = field(default_factory=list) state = DocState() # ── PDF / text extraction ────────────────────────────────────────────────────── def extract_text_from_pdf(path: str) -> str: doc = fitz.open(path) pages = [] for page in doc: pages.append(page.get_text("text")) doc.close() return "\n".join(pages) def extract_text(file_obj) -> tuple[str, str]: """Returns (raw_text, doc_name). Gradio 4 passes a tempfile object with .name path.""" path = file_obj.name if hasattr(file_obj, "name") else str(file_obj) name = path.split("/")[-1] if name.lower().endswith(".pdf") or path.lower().endswith(".pdf"): raw = extract_text_from_pdf(path) else: with open(path, "r", encoding="utf-8", errors="replace") as f: raw = f.read() return raw, name # ── Section-aware parser ─────────────────────────────────────────────────────── # Matches headings like: "1.", "1.2", "1.2.3", "ARTICLE IV", "Chapter 3", # "Section 4", or ALL-CAPS lines (≤ 60 chars) used as headings in many reports. HEADING_RE = re.compile( r"^(" r"(?:chapter|section|article|part|annex)\s+[\dA-Z][\w\.]*" # named headings r"|[\d]+(?:\.[\d]+)*\.?\s+\S" # numbered 1. / 1.2 / 1.2.3 r"|[A-Z][A-Z\s\-]{3,59}$" # ALL-CAPS titles ≤60 chars r")", re.IGNORECASE | re.MULTILINE, ) def parse_sections(raw: str) -> list[tuple[str, str]]: """ Returns list of (section_heading, section_body) pairs. Heading '' means text before the first heading (preamble). """ lines = raw.splitlines() sections: list[tuple[str, str]] = [] current_heading = "" current_lines: list[str] = [] for line in lines: stripped = line.strip() if not stripped: current_lines.append("") continue if HEADING_RE.match(stripped) and len(stripped) < 120: # flush previous section body = "\n".join(current_lines).strip() if body or current_heading: sections.append((current_heading, body)) current_heading = stripped current_lines = [] else: current_lines.append(stripped) # flush last body = "\n".join(current_lines).strip() if body or current_heading: sections.append((current_heading, body)) # If we found zero headings, treat the whole doc as one section if len(sections) == 1 and sections[0][0] == "": return [("Document", sections[0][1])] return sections def build_section_path(heading: str, idx: int) -> str: """Readable citation path, e.g. '§4 · Risk Assessment'.""" if not heading: return "§Preamble" return f"§{idx + 1} · {heading[:80]}" # ── Chunking ─────────────────────────────────────────────────────────────────── def chunk_section( text: str, section_path: str, section_index: int, ) -> list[Chunk]: """Split section body into overlapping character chunks.""" chunks: list[Chunk] = [] start = 0 para = 0 while start < len(text): end = min(start + CHUNK_CHARS, len(text)) chunks.append(Chunk( text = text[start:end], section_path = section_path, section_index = section_index, para_index = para, char_start = start, char_end = end, )) if end == len(text): break start += CHUNK_CHARS - CHUNK_OVERLAP para += 1 return chunks def ingest_document(raw: str) -> tuple[list[Chunk], object, object, list[str]]: sections = parse_sections(raw) all_chunks: list[Chunk] = [] toc: list[str] = [] for sec_idx, (heading, body) in enumerate(sections): if not body.strip(): continue path = build_section_path(heading, sec_idx) toc.append(path) # prepend heading into each chunk so retrieval is heading-aware headed_body = f"{heading}\n{body}" if heading else body all_chunks += chunk_section(headed_body, path, sec_idx) if not all_chunks: return [], None, None, [] texts = [c.text for c in all_chunks] # Dense index (BGE-base) embeddings = embedder.encode( texts, normalize_embeddings=True, batch_size=32, show_progress_bar=False ).astype("float32") faiss_index = faiss.IndexFlatIP(embeddings.shape[1]) faiss_index.add(embeddings) # Sparse index (BM25) tokenised = [t.lower().split() for t in texts] bm25_index = BM25Okapi(tokenised) return all_chunks, faiss_index, bm25_index, toc # ── Hybrid retrieval + RRF ───────────────────────────────────────────────────── def rrf_fuse(rank_dicts: list[dict[int, int]], k: int = RRF_K) -> dict[int, float]: scores: dict[int, float] = {} for rd in rank_dicts: for doc_id, rank in rd.items(): scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (k + rank) return scores def hybrid_retrieve(query: str) -> list[tuple[Chunk, float]]: chunks = state.chunks faiss_index = state.faiss_index bm25_index = state.bm25_index n = len(chunks) # Dense q_emb = embedder.encode([query], normalize_embeddings=True).astype("float32") _, d_idx = faiss_index.search(q_emb, min(DENSE_TOPK, n)) dense_ranks = {int(d_idx[0][r]): r for r in range(len(d_idx[0])) if d_idx[0][r] >= 0} # BM25 bm25_scores = bm25_index.get_scores(query.lower().split()) bm25_top = np.argsort(bm25_scores)[::-1][:BM25_TOPK] bm25_ranks = {int(i): rank for rank, i in enumerate(bm25_top)} # RRF fused = rrf_fuse([dense_ranks, bm25_ranks]) if not fused: return [] max_score = max(fused.values()) sorted_ids = sorted(fused, key=lambda i: fused[i], reverse=True) return [(chunks[i], fused[i] / max_score) for i in sorted_ids if i < n] def rerank_chunks(query: str, candidates: list[tuple[Chunk, float]]) -> list[tuple[Chunk, float]]: pairs = [[query, c.text] for c, _ in candidates] ce_scores = reranker.predict(pairs, batch_size=16) ranked = sorted(zip(candidates, ce_scores), key=lambda x: x[1], reverse=True) top = ranked[:RERANK_TOPK] sigmoid = lambda x: float(1.0 / (1.0 + np.exp(-x))) return [(chunk, sigmoid(s)) for (chunk, _), s in top] # ── Generation ───────────────────────────────────────────────────────────────── def build_context(top_chunks: list[tuple[Chunk, float]]) -> str: """Build numbered context block with inline section citations.""" parts = [] for i, (c, _) in enumerate(top_chunks, 1): parts.append(f"[{i}] ({c.section_path}, para {c.para_index + 1})\n{c.text}") return "\n\n".join(parts) def generate_answer(query: str, context: str) -> str: prompt = ( "You are a precise document analyst. " "Answer the question using ONLY the numbered context passages below. " "Reference passages by their section and paragraph number. " "Do not use outside knowledge or hallucinate.\n\n" f"Context:\n{context}\n\n" f"Question: {query}\n\n" "Answer (cite sections inline, e.g. 'According to §3 · Risk Assessment, para 2, ...'):" ) inputs = gen_tok(prompt, return_tensors="pt", truncation=True, max_length=1200).to(DEVICE) with torch.no_grad(): out = gen_model.generate( **inputs, max_new_tokens=GEN_MAX_TOKENS, num_beams=4, early_stopping=True, no_repeat_ngram_size=3, length_penalty=1.2, ) return gen_tok.decode(out[0], skip_special_tokens=True).strip() # ── Judge ────────────────────────────────────────────────────────────────────── def judge_answer(query: str, context: str, answer: str) -> tuple[int, str]: """flan-t5-large (770MB) judges the flan-t5-base (250MB) answer. Returns (score, rationale).""" prompt = ( "You are a strict QA evaluator for policy documents.\n" f"Question: {query}\n" f"Retrieved context:\n{context[:2000]}\n" f"Proposed answer: {answer}\n\n" "Score the answer 1–10 on:\n" " • Faithfulness — no claims beyond the context\n" " • Completeness — addresses the full question\n" " • Section accuracy — cites correct sections\n\n" "Reply in EXACTLY this format:\n" "Score: \n" "Rationale: " ) inputs = judge_tok(prompt, return_tensors="pt", truncation=True, max_length=1200).to(DEVICE) with torch.no_grad(): out = judge_model.generate(**inputs, max_new_tokens=JUDGE_MAX_TOKENS, num_beams=2) raw = judge_tok.decode(out[0], skip_special_tokens=True).strip() m_score = re.search(r"Score:\s*(\d+)", raw) score = max(1, min(10, int(m_score.group(1)))) if m_score else 1 m_rat = re.search(r"Rationale:\s*(.+)", raw, re.DOTALL) rationale = m_rat.group(1).strip() if m_rat else raw[:200] return score, rationale # ── Escalation ───────────────────────────────────────────────────────────────── ESCALATION = { "no_doc": "No document loaded. Upload a PDF or paste text first.", "low_retrieval": "No passages found with sufficient relevance to your question.", "gen_error": "Answer generation failed — the question may be out of scope.", "judge_fail": "The generated answer did not pass quality review.", } def escalate(key: str, query: str, detail: str = "") -> str: detail_block = f"\n\n*Judge rationale: {detail}*" if detail else "" return ( f"⚠️ **Escalation — {ESCALATION.get(key, 'Unknown error.')}**" f"{detail_block}\n\n" f"**Try:**\n" f"- Rephrase your question using terms from the document\n" f"- Check the Table of Contents to find the relevant section\n" f"- Ask a narrower, more specific question\n\n" f"> *Your question: {query}*" ) # ── Citation formatter ───────────────────────────────────────────────────────── def fmt_citation_block(top_chunks: list[tuple[Chunk, float]]) -> str: lines = [] for i, (c, score) in enumerate(top_chunks, 1): lines.append( f"**[{i}] {c.section_path}** · para {c.para_index + 1} " f"*(relevance: {score:.2f})*\n" f"> {c.text[:220].strip()}…" ) return "\n\n".join(lines) def suggest_followups(top_chunks: list[tuple[Chunk, float]]) -> str: seen_sections = list(dict.fromkeys(c.section_path for c, _ in top_chunks))[:3] questions = [ f"What are the key requirements in {seen_sections[0]}?" if seen_sections else "What are the key requirements in this section?", "What exceptions or conditions apply here?", "What are the penalties or consequences described?", "What definitions are given for the terms used?", ] return "\n".join(f"- {q}" for q in questions[:3]) # ── Core agent ───────────────────────────────────────────────────────────────── def answer_question(query: str) -> str: query = query.strip() if not query: return "Please enter a question." if not state.chunks or state.faiss_index is None: return escalate("no_doc", query) # 1. Hybrid retrieval candidates = hybrid_retrieve(query) if not candidates or candidates[0][1] < RRF_GATE: return escalate("low_retrieval", query) # 2. Rerank top candidates top_chunks = rerank_chunks(query, candidates) # 3. Build context (numbered, with section labels) context = build_context(top_chunks) # 4. Generate answer — flan-t5-base try: answer = generate_answer(query, context) except Exception: return escalate("gen_error", query) # 5. Judge answer — flan-t5-large try: score, rationale = judge_answer(query, context, answer) except Exception: score, rationale = 5, "Judge unavailable." # 6. Escalate if judge rejects if score < JUDGE_PASS: return escalate("judge_fail", query, detail=rationale) # 7. Return cited answer cite_block = fmt_citation_block(top_chunks[:3]) followups = suggest_followups(top_chunks) state.history.append({"q": query, "a": answer, "score": score}) return ( f"### Answer\n{answer}\n\n" f"*Judge score: {score}/10 · {rationale}*\n\n" f"---\n\n" f"**Cited passages**\n\n{cite_block}\n\n" f"---\n\n" f"**Suggested follow-ups**\n{followups}" ) # ── Document ingestion handler ───────────────────────────────────────────────── def load_document(file_obj, pasted_text: str) -> str: # Prefer uploaded file; fall back to pasted text if file_obj is not None: try: raw, name = extract_text(file_obj) except Exception as e: return f"❌ Could not read file: {e}" elif pasted_text and pasted_text.strip(): raw = pasted_text.strip() name = "pasted-document.txt" else: return "Upload a file or paste text to begin." if len(raw.strip()) < 100: return "❌ Document appears empty or too short." chunks, faiss_index, bm25_index, toc = ingest_document(raw) if not chunks: return "❌ Could not extract any text from the document." state.doc_name = name state.chunks = chunks state.faiss_index = faiss_index state.bm25_index = bm25_index state.toc = toc state.history = [] toc_block = "\n".join(f" {t}" for t in toc[:30]) overflow = f"\n … and {len(toc) - 30} more sections" if len(toc) > 30 else "" return ( f"✅ **{name}** indexed\n" f" {len(chunks)} chunks · {len(toc)} sections\n\n" f"**Table of contents (detected)**\n{toc_block}{overflow}" ) def clear_state() -> tuple: state.doc_name = ""; state.chunks = []; state.faiss_index = None state.bm25_index = None; state.toc = []; state.history = [] return None, "", "", "*Document cleared.*" # ── Gradio UI ────────────────────────────────────────────────────────────────── CSS = """ body { font-family: 'IBM Plex Mono', monospace; } #hdr { background:#0d0d0d; color:#e8e8e8; padding:1.6rem 2.4rem 1.3rem; border-bottom:2px solid #1e1e1e; } #hdr h1 { font-size:1.45rem; font-weight:600; margin:0 0 .2rem; color:#fff; letter-spacing:-.3px; } #hdr p { font-size:.78rem; color:#666; margin:0; } #load-btn { background:#111 !important; color:#e8e8e8 !important; border:1px solid #2a2a2a !important; } #ask-btn { background:#111 !important; color:#e8e8e8 !important; border:1px solid #2a2a2a !important; } #clr-btn { background:transparent !important; color:#555 !important; border:1px solid #222 !important; } """ PIPELINE_INFO = ( "**Pipeline**\n" "1. BGE-base → FAISS (dense, top-20)\n" "2. BM25Okapi (sparse, top-20)\n" "3. RRF fusion k=60\n" "4. BGE-reranker-large (top-5)\n" "5. flan-t5-base → answer (250MB)\n" "6. flan-t5-large → judge (770MB, ≥ 6/10)\n\n" "*Escalates on: no doc · low retrieval · judge < 6*" ) with gr.Blocks(css=CSS, title="PolicyQA") as demo: gr.HTML("""

📋 PolicyQA — Document QA Agent

Upload a policy document or report → ask questions → get answers cited to exact sections

""") with gr.Row(): # ── Left: document loader ────────────────────────────────────────────── with gr.Column(scale=1, min_width=300): gr.Markdown("**01 · Load Document**") file_upload = gr.File( label="Upload PDF or .txt", file_types=[".pdf", ".txt"], ) paste_box = gr.Textbox( label="Or paste document text", placeholder="Paste the full text of your policy document here…", lines=6, ) load_btn = gr.Button("Index Document", elem_id="load-btn") load_out = gr.Markdown("*No document loaded.*") gr.Markdown("---") gr.Markdown(PIPELINE_INFO) clr_btn = gr.Button("Clear & Reset", elem_id="clr-btn") # ── Right: QA interface ──────────────────────────────────────────────── with gr.Column(scale=2): gr.Markdown("**02 · Ask the Agent**") q_box = gr.Textbox( label="Your question", placeholder="What are the reporting obligations under Section 4?", lines=3, ) ask_btn = gr.Button("Ask →", elem_id="ask-btn") ans_box = gr.Markdown("*Index a document first, then ask a question.*") # ── Events ───────────────────────────────────────────────────────────────── load_btn.click( fn=load_document, inputs=[file_upload, paste_box], outputs=load_out, ) ask_btn.click( fn=answer_question, inputs=[q_box], outputs=ans_box, ) q_box.submit( fn=answer_question, inputs=[q_box], outputs=ans_box, ) clr_btn.click( fn=clear_state, inputs=[], outputs=[file_upload, paste_box, q_box, load_out], ) # Fix Gradio 4.x localhost reachability check try: import gradio.networking as _gn _gn.is_localhost = lambda *a, **kw: False except Exception: pass demo.launch(server_name="0.0.0.0", server_port=7860)