Spaces:
Configuration error
Configuration error
| """ | |
| rag.py — Retrieval and answer generation for the Digital KP 2030 assistant. | |
| Pipeline: | |
| question -> embed -> FAISS search -> top-k page-tagged chunks | |
| -> grounded prompt -> LLM -> answer + page citations | |
| The LLM is reached through an OpenAI-compatible client, so the same code works | |
| with Groq, OpenRouter, Together or OpenAI itself — you only change two secrets. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import os | |
| from dataclasses import dataclass | |
| from pathlib import Path | |
| import numpy as np | |
| INDEX_DIR = Path(__file__).parent / "index" | |
| # Read from environment / Hugging Face Secrets. Never hard-code a key. | |
| LLM_API_KEY = os.environ.get("LLM_API_KEY", "") | |
| LLM_BASE_URL = os.environ.get("LLM_BASE_URL", "https://api.groq.com/openai/v1") | |
| LLM_MODEL = os.environ.get("LLM_MODEL", "llama-3.3-70b-versatile") | |
| TOP_K = 6 | |
| MIN_SCORE = 0.22 # below this the match is too weak to be worth citing | |
| SYSTEM_PROMPT = """You are the Digital KP Assistant. You answer questions about one \ | |
| document only: the Khyber Pakhtunkhwa Digital Transformation Policy & Roadmap 2030 \ | |
| ("Digital KP 2030"). | |
| Rules you must follow without exception: | |
| 1. Answer ONLY from the numbered excerpts provided. Never use outside knowledge about \ | |
| Khyber Pakhtunkhwa, Pakistan, or digital policy in general. | |
| 2. Cite the page for every factual claim, in the form (p. 27). Take page numbers from \ | |
| the excerpt headers. If one claim draws on two excerpts, cite both: (pp. 25, 27). | |
| 3. If the excerpts do not contain the answer, say exactly: "I could not find that in the \ | |
| Digital KP 2030 document." Then, if useful, name the section that looks closest. Do not \ | |
| guess, and do not fill the gap with plausible-sounding policy language. | |
| 4. If the excerpts disagree with each other, say so and cite both pages rather than \ | |
| silently picking one. This document does contain a few internal inconsistencies. | |
| 5. Quote sparingly — at most one short phrase per excerpt. Otherwise paraphrase. | |
| 6. Reply in the language the user wrote in. If they write in Urdu or Pashto, answer in \ | |
| that language but keep official names (DASTAK, Khyber Pass, KPITB, DIGIT) in English. | |
| 7. Be concise and concrete. Prefer the document's own terminology: goals, pillars, \ | |
| sub-pillars, thematic domains, interventions, KPIs.""" | |
| class Source: | |
| page: int | |
| text: str | |
| score: float | |
| class Assistant: | |
| """Loads the prebuilt index once and answers questions against it.""" | |
| def __init__(self) -> None: | |
| import faiss | |
| from sentence_transformers import SentenceTransformer | |
| meta_path = INDEX_DIR / "meta.json" | |
| if not meta_path.exists(): | |
| raise FileNotFoundError( | |
| "index/ not found. Run `python ingest.py` locally and commit the " | |
| "index/ folder alongside the app." | |
| ) | |
| self.meta = json.loads(meta_path.read_text(encoding="utf-8")) | |
| self.chunks = json.loads((INDEX_DIR / "chunks.json").read_text(encoding="utf-8")) | |
| self.index = faiss.read_index(str(INDEX_DIR / "faiss.index")) | |
| self.embedder = SentenceTransformer(self.meta["embed_model"]) | |
| # ---------------- retrieval ---------------- | |
| def search(self, question: str, k: int = TOP_K) -> list[Source]: | |
| vector = self.embedder.encode( | |
| [question], normalize_embeddings=True | |
| ).astype("float32") | |
| scores, ids = self.index.search(vector, k) | |
| results: list[Source] = [] | |
| for score, idx in zip(scores[0], ids[0]): | |
| if idx < 0 or score < MIN_SCORE: | |
| continue | |
| record = self.chunks[int(idx)] | |
| results.append(Source(page=record["page"], text=record["text"], score=float(score))) | |
| return results | |
| # ---------------- generation ---------------- | |
| def _build_context(sources: list[Source]) -> str: | |
| blocks = [] | |
| for i, s in enumerate(sources, start=1): | |
| blocks.append(f"[Excerpt {i} — page {s.page}]\n{s.text}") | |
| return "\n\n---\n\n".join(blocks) | |
| def answer(self, question: str, history: list[dict] | None = None) -> dict: | |
| """ | |
| Returns {"answer": str, "sources": list[Source], "error": str | None}. | |
| `history` is a list of {"role", "content"} dicts for follow-up questions. | |
| """ | |
| sources = self.search(question) | |
| if not sources: | |
| return { | |
| "answer": "I could not find that in the Digital KP 2030 document. " | |
| "Try rephrasing with terms the document uses — for example " | |
| "*goal*, *pillar*, *thematic domain*, *intervention* or *KPI*.", | |
| "sources": [], | |
| "error": None, | |
| } | |
| if not LLM_API_KEY: | |
| return { | |
| "answer": "No LLM key is configured, so I can only show the passages I " | |
| "retrieved. Set `LLM_API_KEY` in the Space secrets to enable " | |
| "generated answers.", | |
| "sources": sources, | |
| "error": "missing_key", | |
| } | |
| messages = [{"role": "system", "content": SYSTEM_PROMPT}] | |
| for turn in (history or [])[-4:]: # keep the last two exchanges | |
| messages.append({"role": turn["role"], "content": turn["content"]}) | |
| messages.append({ | |
| "role": "user", | |
| "content": ( | |
| f"Excerpts from Digital KP 2030:\n\n{self._build_context(sources)}\n\n" | |
| f"---\n\nQuestion: {question}" | |
| ), | |
| }) | |
| try: | |
| from openai import OpenAI | |
| client = OpenAI(api_key=LLM_API_KEY, base_url=LLM_BASE_URL) | |
| completion = client.chat.completions.create( | |
| model=LLM_MODEL, | |
| messages=messages, | |
| temperature=0.1, # low: this is a factual lookup task | |
| max_tokens=900, | |
| ) | |
| return { | |
| "answer": completion.choices[0].message.content.strip(), | |
| "sources": sources, | |
| "error": None, | |
| } | |
| except Exception as exc: # noqa: BLE001 | |
| return { | |
| "answer": "The language model could not be reached. The retrieved " | |
| "passages are shown below so you can still read the source.", | |
| "sources": sources, | |
| "error": f"{type(exc).__name__}: {exc}", | |
| } | |
| EXAMPLE_QUESTIONS = [ | |
| "What is the vision of Digital KP 2030?", | |
| "Explain the DIGIT 2030 framework.", | |
| "What are the nine strategic goals?", | |
| "Who chairs the KP Digital Council, and how often does it meet?", | |
| "What does the roadmap say about AI governance and ethics?", | |
| "How many interventions sit under each strategic pillar?", | |
| "What are the 2030 KPIs for digital payments?", | |
| "What is DASTAK and how does it relate to Khyber Pass?", | |
| "What is expected of a Departmental Implementation Forum?", | |
| "What targets exist for upskilling government officials?", | |
| ] | |