स्थानीय सामग्रीको लागि Q&A बोट

लक्ष्य

Nepal-आधारित छोटो-NGO को आन्तरिक HR/policy assistant, जुन:

• 5-8 नीति कागजातहरू (Nepali PDFs, अंग्रेजी mixed) माथि जवाफ दिन्छ।
• नेपाली वा अंग्रेजी मा प्रश्नहरू लिन्छ।
• हर तथ्यको लागि source कागजात cite गर्दछ।
• यदि कागजातहरूले प्रश्न कभर गर्दैनन् भने जवाफ दिन इन्कार गर्दछ।
• संवादात्मक follow-ups सम्हाल्दछ (“HR policy को बारेमा फेरि भन्नुहोस्” वा “र maternity को बारेमा?”)।
• नराम्रो जवाफ (thumbs down) को लागि हलुका feedback loop प्रस्ताव गर्दछ।

एक साधारण Streamlit UI यसलाई तपाईंको सहकर्मीलाई देखाउन पर्याप्त बनाउँछ।

पूर्ण कोड

तीन फाइलहरू। पहिलो, indexer.py, जुन कागजातहरूलाई index मा एक पटक भर्दछ:

# indexer.py
import os
import pickle
from pathlib import Path
from dotenv import load_dotenv
from openai import OpenAI
from pypdf import PdfReader

load_dotenv()
openai_client = OpenAI()


def extract_pdf(path: str) -> str:
    reader = PdfReader(path)
    return "\n".join(page.extract_text() for page in reader.pages)


def chunk(text: str, chunk_size: int = 400, overlap: int = 80) -> list[str]:
    words = text.split()
    chunks = []
    step = chunk_size - overlap
    for i in range(0, len(words), step):
        piece = " ".join(words[i:i + chunk_size])
        if len(piece.split()) < 50:
            continue
        chunks.append(piece)
    return chunks


def embed_batch(texts: list[str]) -> list[list[float]]:
    response = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=texts,
    )
    return [d.embedding for d in response.data]


def build_index(pdf_dir: str, index_out: str = "index.pkl") -> None:
    all_chunks = []

    for path in Path(pdf_dir).glob("*.pdf"):
        doc_id = path.stem
        text = extract_pdf(str(path))
        for i, piece in enumerate(chunk(text)):
            all_chunks.append({
                "chunk_id": f"{doc_id}#{i}",
                "doc_id": doc_id,
                "text": piece,
            })

    print(f"Chunks: {len(all_chunks)}")

    vectors = []
    for i in range(0, len(all_chunks), 100):
        batch = [c["text"] for c in all_chunks[i:i + 100]]
        vectors.extend(embed_batch(batch))
        print(f"Embedded {i + len(batch)} / {len(all_chunks)}")

    for chunk_data, vector in zip(all_chunks, vectors):
        chunk_data["vector"] = vector

    with open(index_out, "wb") as f:
        pickle.dump(all_chunks, f)

    print(f"Index saved to {index_out}")


if __name__ == "__main__":
    import sys
    build_index(sys.argv[1] if len(sys.argv) > 1 else "docs")

python indexer.py docs

एक पटक चलेको छ। नविकरण गरिएका कागजातहरू पछि आउँदा नै फेरि चलाउनुहोस्।

दोस्रो फाइल, niyam.py, वास्तविक assistant class हो:

# niyam.py
import pickle
import os
from dotenv import load_dotenv
from openai import OpenAI
from anthropic import Anthropic
import numpy as np

load_dotenv()
openai_client = OpenAI()
anthropic_client = Anthropic()


SYSTEM_PROMPT = """You are niyam, an assistant for an NGO's internal policies.
Answer questions using ONLY the documents provided below. Follow these rules:

1. Answer in the same language the user asked (Nepali/English).
2. If the answer is in the documents, provide it clearly. At the END of your
   answer, list the source documents as: "Sources: [doc1, doc2]"
3. If the answer is NOT clearly in the documents, respond exactly:
   "मैले नीति कागजातहरूमा त्यसको जवाफ भेटिनँ। कृपया HR टिमलाई सम्पर्क गर्नुहोस्।"
4. Do not invent policies, fees, timelines, or names.
5. Keep answers brief (2-4 sentences) unless the question demands detail.
6. If the user has asked a follow-up (e.g. "क्या यो सधैँ लागू हुन्छ?"),
   consider the prior conversation and current documents together.
"""


class Niyam:
    def __init__(self, index_path: str = "index.pkl"):
        with open(index_path, "rb") as f:
            self.chunks = pickle.load(f)
        self.history: list[dict] = []  # rolling conversation history

    def embed(self, text: str) -> list[float]:
        return openai_client.embeddings.create(
            model="text-embedding-3-small",
            input=text,
        ).data[0].embedding

    def cosine(self, v1: list[float], v2: list[float]) -> float:
        a, b = np.array(v1), np.array(v2)
        return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

    def search(self, query: str, top_k: int = 3) -> list[dict]:
        query_vector = self.embed(query)
        scored = [
            {**c, "score": self.cosine(query_vector, c["vector"])}
            for c in self.chunks
        ]
        scored.sort(key=lambda x: x["score"], reverse=True)
        return scored[:top_k]

    def ask(self, question: str) -> dict:
        hits = self.search(question, top_k=4)
        top_score = hits[0]["score"]

        if top_score < 0.32:
            answer = (
                "मैले नीति कागजातहरूमा त्यसको जवाफ भेटिनँ। "
                "कृपया HR टिमलाई सम्पर्क गर्नुहोस्।"
            )
            return {
                "answer": answer,
                "sources": [],
                "confidence": "low",
                "top_score": top_score,
            }

        context = "\n\n---\n\n".join(
            f"[Document: {h['doc_id']}]\n{h['text']}" for h in hits
        )

        # Include short prior turns
        messages = list(self.history[-4:])  # last 2 exchanges
        messages.append({
            "role": "user",
            "content": f"Documents:\n{context}\n\n---\n\nQuestion: {question}",
        })

        response = anthropic_client.messages.create(
            model="claude-3-5-sonnet-latest",
            max_tokens=500,
            system=SYSTEM_PROMPT,
            messages=messages,
        )

        answer = response.content[0].text

        self.history.append({"role": "user", "content": question})
        self.history.append({"role": "assistant", "content": answer})

        sources = sorted({h["doc_id"] for h in hits[:3]})
        confidence = "high" if top_score > 0.6 else "medium"

        return {
            "answer": answer,
            "sources": sources,
            "confidence": confidence,
            "top_score": top_score,
        }

    def feedback(self, question: str, answer: str, thumbs: str) -> None:
        """Log user feedback for later review."""
        with open("feedback.log", "a") as f:
            f.write(f"[{thumbs}] Q: {question}\nA: {answer[:200]}\n---\n")


if __name__ == "__main__":
    bot = Niyam()
    print("niyam — नीति सहायक। बाहिर निस्कन 'exit' टाइप गर्नुहोस्।")
    while True:
        q = input("\nतपाईंको प्रश्न: ").strip()
        if q.lower() in {"exit", "quit"}:
            break
        result = bot.ask(q)
        print(f"\n{result['answer']}")
        if result["sources"]:
            print(f"\n[स्रोतहरू: {', '.join(result['sources'])} | "
                  f"विश्वास: {result['confidence']}]")

तेस्रो फाइल, app.py, एउटा साना Streamlit UI:

# app.py
import streamlit as st
from niyam import Niyam


st.set_page_config(page_title="niyam", page_icon="📘")
st.title("📘 niyam — नीति सहायक")

if "bot" not in st.session_state:
    st.session_state.bot = Niyam()
if "messages" not in st.session_state:
    st.session_state.messages = []

for msg in st.session_state.messages:
    with st.chat_message(msg["role"]):
        st.markdown(msg["content"])
        if msg.get("sources"):
            st.caption(f"Sources: {', '.join(msg['sources'])} · "
                       f"{msg.get('confidence', '')}")

prompt = st.chat_input("नीति बारे प्रश्न सोध्नुहोस्...")
if prompt:
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

    with st.chat_message("assistant"):
        with st.spinner("खोज्दै..."):
            result = st.session_state.bot.ask(prompt)
        st.markdown(result["answer"])
        if result["sources"]:
            st.caption(f"Sources: {', '.join(result['sources'])} · "
                       f"{result['confidence']}")

        col1, col2 = st.columns([1, 20])
        with col1:
            if st.button("👍", key=f"up_{len(st.session_state.messages)}"):
                st.session_state.bot.feedback(prompt, result["answer"], "up")
                st.toast("धन्यवाद!")
            if st.button("👎", key=f"down_{len(st.session_state.messages)}"):
                st.session_state.bot.feedback(prompt, result["answer"], "down")
                st.toast("Feedback लिइयो।")

    st.session_state.messages.append({
        "role": "assistant",
        "content": result["answer"],
        "sources": result["sources"],
        "confidence": result["confidence"],
    })

streamlit run app.py

अब तपाईंसँग एउटा real, प्रयोगयोग्य policy assistant छ। यसलाई तपाईंको सहकर्मीलाई देखाउनुहोस् र तिनीहरू यसलाई प्रयोग गर्न सक्छन्।

Design विकल्पहरू, स्पष्ट गरिएको

आउनुहोस् केही निर्णयहरू के गरे र किन जान्नुहोस्:

Search threshold 0.32। यो पर्याप्त कम छ कि अधिकांश legitimate प्रश्नहरू पास हुन्छ, र पर्याप्त उच्च छ कि पूर्णत: असम्बन्धित प्रश्नहरू (“मौसम कस्तो छ?”) सही रूपमा अस्वीकार गरिन्छन्। आफ्नो कागजातहरूमा tune गर्नुहोस्। रैंडम सट्टामा प्रश्नहरूसँग test गर्नुहोस् जुन कागजातहरूमा हुनुहुन्न, र threshold लाई adjust गर्नुहोस् जबसम्म ती अस्वीकृत नहुन्।

Rolling conversation history (last 4 messages)। यसले follow-ups सम्हाल्न पर्याप्त context दिन्छ (“त्यो सधैँ लागू हुन्छ?”) token खर्च exploding बिना। सिमित सम्मेलनको लागि सिट टिपिकल विकल्प।

Confidence तह UI मा दृश्यमान। यो प्रयोगकर्ताहरूलाई प्रत्येक जवाफलाई कति भरोसा गर्ने भन्ने संकेत गर्दछ। ठूला दृश्यमान संकेत खराब जवाफलाई पेश गर्न भन्दा राम्रो हो जुन high-confidence देखिन्छ।

Thumbs up/down feedback। एक हप्ता पछि, feedback.log एक gold-mine हो: सबै thumbs-down cases हरू छन् जुन बिग्रिएका छन्। तिनीहरू review गर्नुहोस्, retrieval quality सुधार गर्नुहोस्, वा कागजातहरू update गर्नुहोस्।

niyam लाई वास्तविक बनाउनको लागि तपाईंले के थप्नुहुन्छ

यो सहायक साना संकेत परीक्षणको लागि तयार छ। एक साना NGO साठि इसलाई real deploy गर्नको लागि, थप्नुहोस्:

• Authentication। एउटा साधारण HTTP basic auth वा magic link। कर्मचारीहरू मात्र प्रवेश गर्नुपर्दछ।
• कागजात नविकरण pipeline। एउटा साप्ताहिक cron job जुन indexer.py लाई फेरि चलाउँछ। नयाँ policy लगायत वा पुरानो updated हुँदा।
• Analytics। एक साना pandas dashboard जुन feedback.log लाई पार्स गर्दछ र show गर्दछ: प्रति दिन प्रश्नहरू, thumbs-down rate, थम्ब्स-डाउन का साथ प्रश्नहरूको उदाहरणहरू।
• PII redaction। यदि प्रयोगकर्ताले प्रश्नमा नाम टाइप गर्दछ, त्यसलाई log गर्दा पहिले scrub गर्नुहोस्। यो privacy hygiene हो।
• कस्तूरी कागजातको लागि access control। केही कागजातहरू (जस्तै executive compensation) मात्र क्रोमो निश्चित प्रयोगकर्ताहरूको लागि हुनुपर्दछ। एक roles field ले कागजातहरूको प्रति र search-time filtering लाई थप्नुहोस्।

यी तिनीहरू सबै standard software engineering elaborations हुन् — जो कुरा तपाईंलाई कुनै पनि real साना app दिनको लागि आवश्यक हुन्छ। LLM stuff (embeddings, RAG, generation) पहिले नै काम गर्दैछ।

niyam लाई सुधार गर्ने केही तरिकाहरू

आफ्नो हातहरू असल हुँदा, यहाँ 3 उन्नत सुधार छन्:

1. Query expansion। एउटा प्रश्न लिनुहोस्, LLM लाई 3 प्रकारले rephrase गर्न भन्नुहोस्, र प्रत्येक rephrased संस्करण को लागि semantic search गर्नुहोस्। परिणामहरूको union लिनुहोस्। यसले "मातृ बिदा" लाई "स्तनपान बिदा" सँग match गर्न मद्दत गर्दछ जुन शब्दहरू पर्याप्त फरक हुन्छन् मूल similarity कम हुन्छ।

2. Reranking। प्रारम्भिक cosine search ले शीर्ष-20 chunks फर्काउँछ। एउटा हल्का reranker (जस्तै Cohere को API-based reranker) प्रयोग गर्दै तिनीहरू लाई फेरि रैंक गर्नुहोस्, top-3 भन्दा राम्रो प्राप्त गर्न। यो एक powerful, low-effort quality boost हो।

3. Multi-hop RAG। केही प्रश्नहरूको लागि “अघि X के हुन्छ र त्यसपछि Y बारेमा” जस्ता कागजातहरूलाई पछ्याउन आवश्यक पर्दछ। पहिले “X” खोज्नुहोस्, त्यसपछि LLM लाई त्यसको जवाफ प्रयोग गरेर “Y” खोज्नुहोस्, त्यसपछि दुबैको जवाफ फिर्ता गर्नुहोस्। जटिल तर वास्तविक-संसार Q&A को लागि शक्तिशाली।

अब के आउँछ

तपाईंले Chapter 5 समाप्त गर्नुभयो। तपाईंसँग एक real, चलिरहेको RAG प्रणाली छ। Chapter 6 (अन्तिम chapter) खुला-अन्त्य AI सहायक निर्माण गर्ने कलामा जान्दछ: कहिले narrow scope साथ जान्ने, कहिले tools र बाह्य APIs थप्ने, र यसले चलिरहेको उत्पादमा जान्दा safety, bias, र Nepali-context परीक्षणको लागि discipline।