initial deploy

Dockerfile

@@ -0,0 +1,12 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY backend/ backend/
+
+EXPOSE 7860
+
+CMD ["uvicorn", "backend.main:app", "--host", "0.0.0.0", "--port", "7860"]

backend/ingestion.py

@@ -0,0 +1,21 @@
+import os
+from dataclasses import dataclass
+
+from fastapi import UploadFile
+
+
+@dataclass
+class FileMetadata:
+    filename: str
+    file_type: str
+    size_bytes: int
+
+
+async def read_upload(file: UploadFile) -> tuple[bytes, FileMetadata]:
+    content = await file.read()
+    ext = os.path.splitext(file.filename or "")[-1].lower()
+    return content, FileMetadata(
+        filename=file.filename or "unknown",
+        file_type=ext or "unknown",
+        size_bytes=len(content),
+    )

backend/main.py

@@ -0,0 +1,146 @@
+from pathlib import Path
+from dotenv import load_dotenv
+load_dotenv(Path(__file__).resolve().parent.parent / ".env")
+
+import asyncio
+import os
+import re
+from fastapi import FastAPI, UploadFile, File, Form, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+
+from .ingestion import read_upload
+from .parser import parse_file
+from .rag import RAGEngine
+
+app = FastAPI(title="Document Ingestion API", version="1.0.0")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+rag_engine = RAGEngine(embed_provider=os.environ.get("EMBED_PROVIDER", "gemini"))
+
+
+class ChatMessage(BaseModel):
+    role: str
+    content: str
+
+
+class QueryRequest(BaseModel):
+    query: str
+    history: list[ChatMessage] = []
+    top_k: int = 3
+
+
+class IngestTextRequest(BaseModel):
+    text: str
+    chunk_size: int = 180
+    chunk_overlap: int = 40
+
+
+def _compute_recommendations(text: str, num_chunks: int) -> dict:
+    sentences = [s.strip() for s in re.split(r"[.!?\n]+", text) if s.strip()]
+    avg_len = sum(len(s) for s in sentences) / max(len(sentences), 1)
+    text_len = len(text)
+
+    if text_len > 5_000_000:
+        rec_cs = max(500, min(1000, int(avg_len * 8 / 10) * 10))
+    elif text_len > 1_000_000:
+        rec_cs = max(300, min(800, int(avg_len * 6 / 10) * 10))
+    else:
+        rec_cs = max(50, min(1000, int(avg_len * 4 / 10) * 10))
+
+    rec_co = max(0, min(200, int(rec_cs * 0.2 / 5) * 5))
+
+    if num_chunks <= 20:
+        rec_tk = 3
+    elif num_chunks <= 100:
+        rec_tk = 5
+    elif num_chunks <= 500:
+        rec_tk = 7
+    else:
+        rec_tk = 10
+
+    return {"chunk_size": rec_cs, "chunk_overlap": rec_co, "top_k": rec_tk}
+
+
+@app.post("/upload")
+async def upload_file(
+    file: UploadFile = File(...),
+    chunk_size: int = Form(180),
+    chunk_overlap: int = Form(40),
+):
+    content, metadata = await read_upload(file)
+
+    try:
+        text = parse_file(metadata.filename, content)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except Exception:
+        raise HTTPException(status_code=500, detail="Failed to process the file.")
+
+    rag_engine.chunk_size = chunk_size
+    rag_engine.chunk_overlap = chunk_overlap
+    num_chunks = rag_engine.start_ingest(text)
+
+    # Run CPU-bound embedding in a background thread
+    loop = asyncio.get_running_loop()
+    loop.run_in_executor(None, rag_engine._do_embed)
+
+    recs = _compute_recommendations(text, num_chunks)
+
+    return {
+        "text": text,
+        "metadata": {
+            "filename": metadata.filename,
+            "file_type": metadata.file_type,
+            "size_bytes": metadata.size_bytes,
+            "chunk_size": chunk_size,
+            "chunk_overlap": chunk_overlap,
+        },
+        "num_chunks": num_chunks,
+        "recommended_chunk_size": recs["chunk_size"],
+        "recommended_chunk_overlap": recs["chunk_overlap"],
+        "recommended_top_k": recs["top_k"],
+    }
+
+
+@app.post("/ingest_text")
+async def ingest_text(body: IngestTextRequest):
+    if not body.text.strip():
+        raise HTTPException(status_code=400, detail="Text is empty.")
+    rag_engine.chunk_size = body.chunk_size
+    rag_engine.chunk_overlap = body.chunk_overlap
+    num_chunks = rag_engine.start_ingest(body.text)
+
+    loop = asyncio.get_running_loop()
+    loop.run_in_executor(None, rag_engine._do_embed)
+
+    return {"num_chunks": num_chunks}
+
+
+@app.get("/ingest_status")
+async def ingest_status():
+    return rag_engine.ingest_progress
+
+
+@app.get("/config")
+async def config():
+    return {"embed_provider": rag_engine.embed_provider}
+
+
+@app.post("/query")
+async def query_document(body: QueryRequest):
+    if not rag_engine.chunks:
+        raise HTTPException(status_code=400, detail="No document has been ingested yet. Upload a file first.")
+    if not rag_engine.is_ready:
+        raise HTTPException(status_code=409, detail="Document is still being embedded. Please wait.")
+    if not body.query.strip():
+        raise HTTPException(status_code=400, detail="Query cannot be empty.")
+    history = [{"role": m.role, "content": m.content} for m in body.history]
+    rag_engine.top_k = body.top_k
+    return rag_engine.answer(body.query, history=history)

backend/parser.py

@@ -0,0 +1,54 @@
+import csv
+import io
+import os
+
+import fitz
+
+SUPPORTED_TYPES = {".txt", ".pdf", ".csv"}
+
+_EXTRACTORS: dict[str, callable] = {}
+
+
+def _decode(content: bytes) -> str:
+    try:
+        return content.decode("utf-8")
+    except UnicodeDecodeError:
+        return content.decode("latin-1")
+
+
+def _extract_txt(content: bytes) -> str:
+    return _decode(content)
+
+
+def _extract_pdf(content: bytes) -> str:
+    with fitz.open(stream=content, filetype="pdf") as doc:
+        return "\n".join(page.get_text() for page in doc)
+
+
+def _extract_csv(content: bytes) -> str:
+    rows = list(csv.reader(io.StringIO(_decode(content))))
+    if not rows:
+        return ""
+    return "\n".join(" | ".join(cell.strip() for cell in row) for row in rows)
+
+
+_EXTRACTORS = {
+    ".txt": _extract_txt,
+    ".pdf": _extract_pdf,
+    ".csv": _extract_csv,
+}
+
+
+def parse_file(filename: str, content: bytes) -> str:
+    if not content:
+        raise ValueError("Uploaded file is empty.")
+
+    ext = os.path.splitext(filename)[-1].lower()
+    extractor = _EXTRACTORS.get(ext)
+    if extractor is None:
+        raise ValueError(f"Unsupported file type '{ext}'. Supported: {', '.join(sorted(SUPPORTED_TYPES))}")
+
+    text = extractor(content)
+    if not text.strip():
+        raise ValueError("No readable text found in the file.")
+    return text

backend/rag.py

@@ -0,0 +1,205 @@
+import logging
+import os
+import threading
+import time
+from dataclasses import dataclass
+
+import google.generativeai as genai
+import numpy as np
+import requests as _requests
+from huggingface_hub import InferenceClient
+from langchain_text_splitters import RecursiveCharacterTextSplitter
+from sklearn.metrics.pairwise import cosine_similarity
+
+log = logging.getLogger(__name__)
+
+FALLBACK_MODELS = [
+    "mistralai/Mistral-7B-Instruct-v0.3",
+    "Qwen/Qwen2.5-7B-Instruct",
+    "meta-llama/Llama-3.1-8B-Instruct",
+]
+
+SYSTEM_PROMPT = (
+    "You are a helpful RAG assistant. Use only the provided context when possible. "
+    "If context is insufficient, say what is missing. Keep answers concise."
+)
+
+EMBED_PROVIDERS = {
+    "gemini": {"model": "models/text-embedding-004", "batch_size": 100},
+    "hf": {"model": "sentence-transformers/all-MiniLM-L6-v2", "batch_size": 32},
+}
+
+
+@dataclass
+class RetrievedChunk:
+    index: int
+    score: float
+    text: str
+
+
+class RAGEngine:
+    def __init__(
+        self,
+        embed_provider: str = "gemini",
+        chunk_size: int = 180,
+        chunk_overlap: int = 40,
+        top_k: int = 3,
+        llm_model: str = "mistralai/Mistral-7B-Instruct-v0.3",
+    ):
+        self.chunk_size = chunk_size
+        self.chunk_overlap = chunk_overlap
+        self.top_k = top_k
+        self.llm_model = llm_model
+        self.embed_provider = embed_provider
+        self.chunks: list[str] = []
+        self.chunk_embeddings: np.ndarray | None = None
+        self._ingest_progress: dict = {"state": "idle", "embedded": 0, "total": 0}
+        self._embed_lock = threading.Lock()
+        self._ingest_generation = 0
+        self._configure_gemini()
+
+    def _configure_gemini(self):
+        api_key = os.environ.get("GOOGLE_API_KEY")
+        if api_key:
+            genai.configure(api_key=api_key)
+
+    def _embed_texts(self, texts: list[str], task_type: str = "RETRIEVAL_DOCUMENT") -> np.ndarray:
+        """Embed a batch of texts using the configured provider."""
+        if self.embed_provider == "gemini":
+            return self._embed_gemini(texts, task_type)
+        return self._embed_hf(texts)
+
+    def _embed_gemini(self, texts: list[str], task_type: str) -> np.ndarray:
+        result = genai.embed_content(
+            model=EMBED_PROVIDERS["gemini"]["model"],
+            content=texts,
+            task_type=task_type,
+        )
+        return np.array(result["embedding"])
+
+    def _embed_hf(self, texts: list[str]) -> np.ndarray:
+        hf_token = os.environ.get("HF_TOKEN", "")
+        model = EMBED_PROVIDERS["hf"]["model"]
+        resp = _requests.post(
+            f"https://api-inference.huggingface.co/pipeline/feature-extraction/{model}",
+            headers={"Authorization": f"Bearer {hf_token}"},
+            json={"inputs": texts, "options": {"wait_for_model": True}},
+            timeout=120,
+        )
+        resp.raise_for_status()
+        return np.array(resp.json())
+
+    @property
+    def ingest_progress(self) -> dict:
+        return dict(self._ingest_progress)
+
+    @property
+    def is_ready(self) -> bool:
+        return (
+            bool(self.chunks)
+            and self.chunk_embeddings is not None
+            and self._ingest_progress.get("state") != "running"
+        )
+
+    def ingest(self, text: str) -> int:
+        """Synchronous ingest: chunk + embed. Returns chunk count."""
+        self._do_split(text)
+        self._do_embed()
+        return len(self.chunks)
+
+    def start_ingest(self, text: str) -> int:
+        """Fast phase: split text into chunks. Returns chunk count.
+        Call _do_embed() in a thread pool to complete ingestion."""
+        self._do_split(text)
+        return len(self.chunks)
+
+    def _do_split(self, text: str):
+        splitter = RecursiveCharacterTextSplitter(
+            chunk_size=self.chunk_size,
+            chunk_overlap=self.chunk_overlap,
+        )
+        self.chunks = splitter.split_text(text)
+        self.chunk_embeddings = None
+        self._ingest_generation += 1
+        total = len(self.chunks)
+        self._ingest_progress = {"state": "running", "embedded": 0, "total": total}
+        log.info("Chunked into %d pieces (size=%d). Embedding…", total, self.chunk_size)
+
+    def _do_embed(self):
+        gen = self._ingest_generation
+        batch_size = EMBED_PROVIDERS.get(self.embed_provider, {}).get("batch_size", 32)
+        try:
+            with self._embed_lock:
+                if gen != self._ingest_generation:
+                    return
+                total = len(self.chunks)
+                if total == 0:
+                    self._ingest_progress = {"state": "done", "embedded": 0, "total": 0}
+                    return
+
+                embeddings, t0 = [], time.time()
+                for i in range(0, total, batch_size):
+                    if gen != self._ingest_generation:
+                        return
+                    batch = self.chunks[i : i + batch_size]
+                    embeddings.append(self._embed_texts(batch, task_type="RETRIEVAL_DOCUMENT"))
+                    done = min(i + batch_size, total)
+                    self._ingest_progress["embedded"] = done
+                    elapsed = time.time() - t0
+                    rate = done / elapsed if elapsed > 0 else 0
+                    remaining = (total - done) / rate if rate > 0 else 0
+                    log.info("Embedded %d/%d (%.0f/s, ~%.0fs left)", done, total, rate, remaining)
+
+                self.chunk_embeddings = np.vstack(embeddings)
+                self._ingest_progress["state"] = "done"
+                log.info("Embedding done in %.1fs", time.time() - t0)
+        except Exception as e:
+            log.exception("Embedding failed")
+            self._ingest_progress = {"state": "error", "embedded": 0, "total": 0, "error": str(e)}
+
+    def retrieve(self, query: str) -> list[RetrievedChunk]:
+        if not self.chunks or self.chunk_embeddings is None:
+            return []
+        query_vec = self._embed_texts([query], task_type="RETRIEVAL_QUERY")
+        scores = cosine_similarity(query_vec, self.chunk_embeddings)[0]
+        top_idx = scores.argsort()[::-1][: self.top_k]
+        return [
+            RetrievedChunk(index=int(i), score=float(scores[i]), text=self.chunks[i])
+            for i in top_idx
+        ]
+
+    def answer(self, query: str, history: list[dict] | None = None) -> dict:
+        retrieved = self.retrieve(query)
+        chunks_payload = [{"index": c.index, "score": round(c.score, 4), "text": c.text} for c in retrieved]
+
+        context_block = "\n\n".join(
+            f"Chunk {c.index} (score={c.score:.4f}):\n{c.text}" for c in retrieved
+        )
+
+        hf_token = os.environ.get("HF_TOKEN")
+        if not hf_token:
+            return {"answer": "HF_TOKEN not set.", "chunks": chunks_payload, "model_used": None}
+
+        messages = self._build_messages(query, context_block, history)
+        return self._call_llm(hf_token, messages, chunks_payload)
+
+    def _build_messages(self, query: str, context: str, history: list[dict] | None) -> list[dict]:
+        messages: list[dict] = [{"role": "system", "content": SYSTEM_PROMPT}]
+        if history:
+            messages.extend({"role": m["role"], "content": m["content"]} for m in history)
+        messages.append({"role": "user", "content": f"Question:\n{query}\n\nContext:\n{context}"})
+        return messages
+
+    def _call_llm(self, token: str, messages: list[dict], chunks_payload: list[dict]) -> dict:
+        client = InferenceClient(api_key=token)
+        # Deduplicated ordered candidate list
+        candidates = list(dict.fromkeys([self.llm_model] + FALLBACK_MODELS))
+
+        for model in candidates:
+            try:
+                resp = client.chat_completion(model=model, messages=messages, max_tokens=350, temperature=0.2)
+                return {"answer": resp.choices[0].message.content, "chunks": chunks_payload, "model_used": model}
+            except Exception:
+                continue
+
+        return {"answer": "All candidate models failed.", "chunks": chunks_payload, "model_used": None}

requirements.txt

@@ -0,0 +1,12 @@
+fastapi
+uvicorn[standard]
+python-multipart
+python-dotenv
+PyMuPDF
+streamlit
+requests
+langchain-text-splitters
+google-generativeai
+scikit-learn
+huggingface-hub
+numpy