Upload folder using huggingface_hub

app/config.py

@@ -0,0 +1,22 @@
+"""Application configuration, loaded from environment variables."""
+import os
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+BASE_DIR = Path(__file__).resolve().parent.parent
+
+GROQ_API_KEY = os.getenv("GROQ_API_KEY", "")
+GROQ_MODEL = os.getenv("GROQ_MODEL", "llama-3.1-8b-instant")
+
+CHROMA_DIR = os.getenv("CHROMA_DIR", str(BASE_DIR / "data" / "chroma"))
+COLLECTION_NAME = os.getenv("COLLECTION_NAME", "python_qa")
+
+TOP_K = int(os.getenv("TOP_K", "5"))
+# Cosine distance above which a retrieved doc is considered irrelevant.
+RELEVANCE_THRESHOLD = float(os.getenv("RELEVANCE_THRESHOLD", "0.50"))
+MAX_QUESTION_LEN = int(os.getenv("MAX_QUESTION_LEN", "2000"))
+ANSWER_CACHE_SIZE = int(os.getenv("ANSWER_CACHE_SIZE", "256"))
+LLM_TIMEOUT_SECONDS = float(os.getenv("LLM_TIMEOUT_SECONDS", "30"))

app/main.py

@@ -0,0 +1,75 @@
+"""FastAPI service exposing the Python Q&A RAG pipeline."""
+import logging
+import time
+from collections import OrderedDict
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI, HTTPException
+from fastapi.responses import RedirectResponse
+from groq import GroqError
+
+from app import config
+from app.rag import RAGPipeline, Retriever
+from app.schemas import AskRequest, AskResponse, HealthResponse
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# question -> response dict, evicted FIFO once full.
+_answer_cache: OrderedDict[str, dict] = OrderedDict()
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    retriever = Retriever()
+    app.state.pipeline = RAGPipeline(retriever)
+    logger.info("Index loaded: %d documents", retriever.count())
+    yield
+
+
+app = FastAPI(
+    title="Python Programming Q&A Assistant",
+    description=(
+        "RAG-powered Q&A over the Stack Overflow Python dataset "
+        "(Kaggle: stackoverflow/pythonquestions), answered by Llama on Groq."
+    ),
+    version="1.0.0",
+    lifespan=lifespan,
+)
+
+
+@app.get("/", include_in_schema=False)
+async def root():
+    return RedirectResponse(url="/docs")
+
+
+@app.get("/health", response_model=HealthResponse)
+async def health():
+    return HealthResponse(
+        status="ok",
+        index_size=app.state.pipeline.retriever.count(),
+        model=config.GROQ_MODEL,
+    )
+
+
+@app.post("/ask", response_model=AskResponse)
+async def ask(req: AskRequest):
+    cache_key = f"{req.question.strip().lower()}|{req.top_k}"
+    if cache_key in _answer_cache:
+        cached = _answer_cache[cache_key]
+        return AskResponse(**{**cached, "cached": True, "latency_ms": 0})
+
+    start = time.perf_counter()
+    try:
+        result = await app.state.pipeline.ask(req.question, top_k=req.top_k)
+    except GroqError as e:
+        logger.exception("LLM call failed")
+        raise HTTPException(status_code=502, detail=f"LLM provider error: {e}") from e
+
+    result["latency_ms"] = int((time.perf_counter() - start) * 1000)
+
+    _answer_cache[cache_key] = result
+    if len(_answer_cache) > config.ANSWER_CACHE_SIZE:
+        _answer_cache.popitem(last=False)
+
+    return AskResponse(**result)

app/rag.py

@@ -0,0 +1,115 @@
+"""RAG pipeline: Chroma retrieval over Stack Overflow Python Q&A + Groq Llama generation."""
+import logging
+
+import chromadb
+from groq import AsyncGroq
+
+from app import config
+
+logger = logging.getLogger(__name__)
+
+SYSTEM_PROMPT = """\
+You are a Python programming Q&A assistant for data science learners. You answer \
+questions using the provided Stack Overflow excerpts as your primary source of truth.
+
+Rules:
+- Ground your answer in the provided context. When you use information from an \
+excerpt, cite it inline as [1], [2], etc. matching the excerpt numbers.
+- Include short, runnable code examples where they help.
+- If the context does not contain enough information, say so explicitly and then \
+give your best general answer, clearly marked as not sourced from the context.
+- If the question is not about Python programming, politely say you only answer \
+Python programming questions and do not attempt to answer it.
+- Be concise and accurate. Prefer modern Python 3 idioms; if an excerpt shows \
+Python 2 syntax, modernise it and mention that you did.
+"""
+
+USER_PROMPT_TEMPLATE = """\
+Stack Overflow excerpts:
+
+{context}
+
+Learner's question: {question}
+"""
+
+
+class Retriever:
+    """Thin wrapper around a persistent Chroma collection of SO Python Q&A pairs."""
+
+    def __init__(self, chroma_dir: str = config.CHROMA_DIR,
+                 collection_name: str = config.COLLECTION_NAME):
+        client = chromadb.PersistentClient(path=chroma_dir)
+        self.collection = client.get_collection(collection_name)
+
+    def count(self) -> int:
+        return self.collection.count()
+
+    def query(self, question: str, k: int = config.TOP_K) -> list[dict]:
+        res = self.collection.query(
+            query_texts=[question],
+            n_results=k,
+            include=["documents", "metadatas", "distances"],
+        )
+        hits = []
+        for doc, meta, dist in zip(res["documents"][0], res["metadatas"][0], res["distances"][0]):
+            hits.append({
+                "text": doc,
+                "title": meta["title"],
+                "url": meta["url"],
+                "answer_score": meta["answer_score"],
+                "tags": meta.get("tags", ""),
+                # Cosine distance -> similarity in [0, 1].
+                "relevance": round(max(0.0, 1.0 - dist), 4),
+                "distance": dist,
+            })
+        return hits
+
+
+def build_context(hits: list[dict]) -> str:
+    blocks = []
+    for i, h in enumerate(hits, start=1):
+        blocks.append(f"[{i}] {h['title']} (answer score: {h['answer_score']})\n{h['text']}")
+    return "\n\n---\n\n".join(blocks)
+
+
+class RAGPipeline:
+    def __init__(self, retriever: Retriever):
+        self.retriever = retriever
+        self.llm = AsyncGroq(api_key=config.GROQ_API_KEY)
+
+    async def ask(self, question: str, top_k: int = config.TOP_K) -> dict:
+        hits = self.retriever.query(question, k=top_k)
+        # RELEVANCE_THRESHOLD is a minimum similarity (0-1); distance = 1 - similarity.
+        relevant = [h for h in hits if h["relevance"] >= config.RELEVANCE_THRESHOLD]
+        grounded = len(relevant) > 0
+        # When nothing passes the threshold, pass all hits anyway —
+        # the prompt instructs the model to flag unsourced answers and decline off-topic ones.
+        context_hits = relevant if grounded else hits
+
+        completion = await self.llm.chat.completions.create(
+            model=config.GROQ_MODEL,
+            messages=[
+                {"role": "system", "content": SYSTEM_PROMPT},
+                {"role": "user", "content": USER_PROMPT_TEMPLATE.format(
+                    context=build_context(context_hits), question=question)},
+            ],
+            temperature=0.2,
+            max_tokens=1024,
+            timeout=config.LLM_TIMEOUT_SECONDS,
+        )
+        answer = completion.choices[0].message.content
+
+        return {
+            "answer": answer,
+            "sources": [
+                {
+                    "title": h["title"],
+                    "url": h["url"],
+                    "relevance": h["relevance"],
+                    "answer_score": h["answer_score"],
+                }
+                for h in context_hits
+            ],
+            "grounded": grounded,
+            "model": config.GROQ_MODEL,
+        }

app/schemas.py

@@ -0,0 +1,44 @@
+"""Pydantic request/response models for the API."""
+from pydantic import BaseModel, Field
+
+from app import config
+
+
+class AskRequest(BaseModel):
+    question: str = Field(
+        ...,
+        min_length=3,
+        max_length=config.MAX_QUESTION_LEN,
+        description="A Python programming question in natural language.",
+        examples=["How do I merge two dictionaries in Python?"],
+    )
+    top_k: int = Field(
+        default=config.TOP_K,
+        ge=1,
+        le=10,
+        description="Number of Stack Overflow Q&A pairs to retrieve for grounding.",
+    )
+
+
+class Source(BaseModel):
+    title: str
+    url: str
+    relevance: float = Field(description="Similarity score in [0, 1]; higher is more relevant.")
+    answer_score: int = Field(description="Stack Overflow vote count of the answer used.")
+
+
+class AskResponse(BaseModel):
+    answer: str
+    sources: list[Source]
+    grounded: bool = Field(
+        description="False when no sufficiently relevant Stack Overflow context was found."
+    )
+    model: str
+    latency_ms: int
+    cached: bool = False
+
+
+class HealthResponse(BaseModel):
+    status: str
+    index_size: int
+    model: str