How to Build a RAG System Over Indian Government Data with Claude

pip install langchain langchain-openai langchain-community \
    chromadb pypdf requests streamlit beautifulsoup4

# collect_rbi.py
import requests
import os
import time
from bs4 import BeautifulSoup
from pathlib import Path
import pypdf
import json

DATA_DIR = Path("./rbi_data")
PDF_DIR = DATA_DIR / "pdfs"
TEXT_DIR = DATA_DIR / "texts"

PDF_DIR.mkdir(parents=True, exist_ok=True)
TEXT_DIR.mkdir(parents=True, exist_ok=True)

HEADERS = {
    "User-Agent": "Mozilla/5.0 (compatible; ResearchBot/1.0)",
}

RBI_BASE = "https://www.rbi.org.in"


def get_circular_links(year: int) -> list[dict]:
    """Scrape circular listing page for a given year."""
    url = f"{RBI_BASE}/scripts/BS_CircularIndexDisplay.aspx?year={year}"
    response = requests.get(url, headers=HEADERS, timeout=30)
    soup = BeautifulSoup(response.text, "html.parser")

    circulars = []
    # RBI's table structure: each row has date, circular number, subject, PDF link
    for row in soup.find_all("tr"):
        cells = row.find_all("td")
        if len(cells) < 3:
            continue

        pdf_link = row.find("a", href=lambda h: h and ".pdf" in h.lower())
        if not pdf_link:
            continue

        href = pdf_link.get("href", "")
        full_url = href if href.startswith("http") else f"{RBI_BASE}{href}"

        circulars.append({
            "date": cells[0].get_text(strip=True) if cells else "",
            "circular_number": cells[1].get_text(strip=True) if len(cells) > 1 else "",
            "subject": cells[2].get_text(strip=True) if len(cells) > 2 else "",
            "pdf_url": full_url,
            "year": year,
        })

    return circulars


def download_and_extract(circular: dict) -> dict | None:
    """Download PDF and extract text."""
    # Sanitise filename from circular number
    safe_name = circular["circular_number"].replace("/", "_").replace(" ", "_")
    pdf_path = PDF_DIR / f"{safe_name}.pdf"
    text_path = TEXT_DIR / f"{safe_name}.txt"

    # Skip if already processed
    if text_path.exists():
        return None

    try:
        # Download PDF
        response = requests.get(circular["pdf_url"], headers=HEADERS, timeout=30)
        response.raise_for_status()
        pdf_path.write_bytes(response.content)

        # Extract text
        reader = pypdf.PdfReader(str(pdf_path))
        text = "\n".join(page.extract_text() or "" for page in reader.pages)
        text = text.strip()

        if len(text) < 100:  # skip nearly-empty extractions
            return None

        text_path.write_text(text, encoding="utf-8")

        return {
            **circular,
            "text_path": str(text_path),
            "char_count": len(text),
        }

    except Exception as e:
        print(f"Failed to process {circular['circular_number']}: {e}")
        return None


def collect_circulars(years: list[int] = None):
    if years is None:
        years = [2022, 2023, 2024, 2025, 2026]

    all_circulars = []
    for year in years:
        print(f"Fetching {year} circular listings...")
        circulars = get_circular_links(year)
        print(f"  Found {len(circulars)} circulars")
        all_circulars.extend(circulars)

    print(f"\nDownloading {len(all_circulars)} circulars...")
    processed = []
    for i, circular in enumerate(all_circulars):
        result = download_and_extract(circular)
        if result:
            processed.append(result)
            print(f"  [{i+1}/{len(all_circulars)}] {circular['circular_number']}")
        time.sleep(0.5)  # be polite to RBI's servers

    # Save manifest
    manifest_path = DATA_DIR / "manifest.json"
    manifest_path.write_text(json.dumps(processed, indent=2))
    print(f"\nDone. Processed {len(processed)} circulars. Manifest at {manifest_path}")


if __name__ == "__main__":
    collect_circulars()

python collect_rbi.py

# build_index.py
import json
from pathlib import Path
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

DATA_DIR = Path("./rbi_data")
CHROMA_DIR = "./rbi_chroma"

# Load manifest
manifest = json.loads((DATA_DIR / "manifest.json").read_text())

# Prepare documents with metadata
from langchain.schema import Document

documents = []
for circular in manifest:
    text_path = Path(circular["text_path"])
    if not text_path.exists():
        continue

    text = text_path.read_text(encoding="utf-8")

    # Preserve structured metadata for citation
    metadata = {
        "circular_number": circular["circular_number"],
        "date": circular["date"],
        "subject": circular["subject"],
        "year": str(circular["year"]),
        "source": circular["pdf_url"],
    }

    documents.append(Document(page_content=text, metadata=metadata))

print(f"Loaded {len(documents)} circulars")

# Split into chunks
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " "],
)
chunks = splitter.split_documents(documents)
print(f"Split into {len(chunks)} chunks")

# Create embeddings using AICredits.in endpoint
# AICredits exposes an OpenAI-compatible endpoint — use text-embedding-3-small
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    openai_api_key=os.environ["AICREDITS_API_KEY"],
    openai_api_base="https://api.aicredits.in/v1",
)

# Build ChromaDB index
print("Building ChromaDB index (this takes a few minutes for large datasets)...")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory=CHROMA_DIR,
)
vectorstore.persist()
print(f"Index saved to {CHROMA_DIR}")

# qa_system.py
import os
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

CHROMA_DIR = "./rbi_chroma"

# Load the existing ChromaDB index
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    openai_api_key=os.environ["AICREDITS_API_KEY"],
    openai_api_base="https://api.aicredits.in/v1",
)

vectorstore = Chroma(
    persist_directory=CHROMA_DIR,
    embedding_function=embeddings,
)

# Claude Sonnet via AICredits.in
llm = ChatOpenAI(
    model="anthropic/claude-sonnet-4-6",
    openai_api_key=os.environ["AICREDITS_API_KEY"],
    openai_api_base="https://api.aicredits.in/v1",
    temperature=0,  # deterministic for regulatory Q&A
)

# Custom prompt that keeps Claude grounded in the source material
PROMPT_TEMPLATE = """You are a regulatory research assistant specialising in RBI (Reserve Bank of India) circulars and policy documents.

Answer the question based ONLY on the provided RBI circular excerpts. Do not use any knowledge outside these excerpts.

If the answer is not found in the provided circulars, say: "I could not find specific guidance on this in the available circulars. You should check the RBI website directly at rbi.org.in for current guidance."

Always cite the circular number and date when you provide information.

Context from RBI circulars:
{context}

Question: {question}

Answer (cite circular numbers and dates):"""

PROMPT = PromptTemplate(
    template=PROMPT_TEMPLATE,
    input_variables=["context", "question"],
)

# Build the QA chain
retriever = vectorstore.as_retriever(
    search_type="mmr",       # Maximum Marginal Relevance — reduces redundant results
    search_kwargs={"k": 5},  # retrieve 5 most relevant chunks
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True,
    chain_type_kwargs={"prompt": PROMPT},
)


def ask_rbi(question: str) -> dict:
    """Query the RBI circular database."""
    result = qa_chain.invoke({"query": question})

    # Extract unique source circulars for display
    sources = []
    seen = set()
    for doc in result["source_documents"]:
        circular_num = doc.metadata.get("circular_number", "Unknown")
        if circular_num not in seen:
            seen.add(circular_num)
            sources.append({
                "circular": circular_num,
                "date": doc.metadata.get("date", ""),
                "subject": doc.metadata.get("subject", ""),
            })

    return {
        "answer": result["result"],
        "sources": sources,
    }


if __name__ == "__main__":
    # Quick test
    result = ask_rbi("What are RBI's current regulations on digital lending apps?")
    print(result["answer"])
    print("\nSources used:")
    for s in result["sources"]:
        print(f"  - {s['circular']} ({s['date']}): {s['subject']}")

# app.py
import streamlit as st
from qa_system import ask_rbi

st.set_page_config(
    page_title="RBI Circular Search",
    page_icon="🏦",
    layout="wide",
)

st.title("RBI Circular Q&A")
st.markdown("Ask questions about RBI regulations. Answers are grounded in RBI circulars from 2022-2026.")

query = st.text_input(
    "Your question",
    placeholder="What did RBI say about BNPL regulations in 2024?",
)

if st.button("Search", type="primary") and query:
    with st.spinner("Searching RBI circulars..."):
        result = ask_rbi(query)

    st.markdown("### Answer")
    st.write(result["answer"])

    if result["sources"]:
        st.markdown("### Source circulars")
        for source in result["sources"]:
            with st.expander(f"{source['circular']} — {source['date']}"):
                st.write(f"**Subject:** {source['subject']}")

AICREDITS_API_KEY=your_key streamlit run app.py

# Add to your MCP server or standalone script
import requests

def get_startup_stats(sector: str = None, state: str = None) -> dict:
    """Query DPIIT startup registry."""
    DPIIT_API = "https://api.startupindia.gov.in/sih/api/search/profiles/startup"
    params = {
        "startupSector": sector,
        "state": state,
        "pageNo": 0,
        "pageSize": 20,
    }
    headers = {
        "Authorization": f"Bearer {os.environ['DPIIT_API_KEY']}",
        "Content-Type": "application/json",
    }
    response = requests.get(DPIIT_API, params=params, headers=headers)
    return response.json()