init

.gitignore

@@ -0,0 +1,4 @@
+*.db 
+.env 
+__pycache__
+/chromadb/*

Dockerfile

@@ -0,0 +1,22 @@
+FROM python:3.10-slim
+
+# Set working directory
+WORKDIR /app
+
+# Copy the current directory contents into the container at /app
+COPY requirements.txt /app
+
+# Install any needed packages specified in requirements.txt
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Create a non-root user
+RUN useradd -m appuser
+
+# Create necessary directories and set permissions
+RUN mkdir -p /app/data && \
+    chown -R appuser:appuser /app && \
+    chmod -R 755 /app
+    
+COPY . /app
+
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

app.py

@@ -0,0 +1,35 @@
+from typing import List
+from fastapi import FastAPI, Form, UploadFile, File
+from data_ingetion.data import AdvancedDatabase
+from data_ingetion.pre_processor import read_metadata
+from generator.response import generate_response
+
+db = AdvancedDatabase()
+app = FastAPI()
+
+
+@app.post("/ingest_files")
+async def ingest_files(
+    metadata: str = Form(...),
+    group_name: str = Form(...),
+    files: List[UploadFile] = File(...),
+):
+    """Upload and process multiple files (PDFs and DOCXs) in single endpoint inference asyncronously"""
+    result = await db.ingest(files, metadata, group_name)
+    return result
+
+
+@app.post("/fetch_response")
+async def get_response(query: str, group_name: str, include_chunks: bool = False):
+    """Invoke RAG pipeline with given collection as VectorDB to retrieve context and generate context rich responses"""
+    response = generate_response(query, group_name)
+    if include_chunks:
+        return {"llm_response": response[0], "chunks": response[1]}
+    return {"llm_response": response}
+
+
+@app.get("/get_metadata")
+async def get_metadata():
+    """Fetch Metadata of recently uploaded documents grouped by collections ie. time they were uploaded."""
+    response = read_metadata()
+    return {"Metadata": response}

data_ingetion/data.py

@@ -0,0 +1,39 @@
+import chromadb
+from uuid import uuid4
+
+from .pre_processor import full_processor
+
+from typing import List
+from fastapi import UploadFile
+
+DB_NAME = "chromadb"
+
+
+class AdvancedDatabase:
+    def __init__(self) -> None:
+        self.client = chromadb.PersistentClient(path=DB_NAME)
+
+    async def ingest(self, files: List[UploadFile], user_metadata, collection_name):
+        chunks, metadata, embeddings = await full_processor(
+            files, user_metadata, collection_name
+        )
+
+        collection = self.client.create_collection(name=collection_name)
+
+        collection.add(
+            ids=[str(uuid4()) for _ in range(len(chunks))],
+            embeddings=embeddings,
+            documents=chunks,
+            metadatas=metadata,
+        )
+
+        return {"chunks_added": len(chunks), "collection_name": collection_name}
+
+    def get_context(self, embedding: List[float], group_name: str):
+        collection = self.client.get_collection(name=group_name)
+
+        response = collection.query(
+            query_embeddings=embedding, n_results=5, include=["documents"]
+        )
+
+        return response["documents"][0]

data_ingetion/pre_processor.py

@@ -0,0 +1,145 @@
+from docx import Document
+from pypdf import PdfReader
+import datetime
+from langchain_text_splitters import RecursiveCharacterTextSplitter
+from openai import OpenAI
+import sqlite3
+import json
+
+import os
+from io import BytesIO
+
+from typing import List
+from fastapi import UploadFile
+
+splitter = RecursiveCharacterTextSplitter(chunk_size=225, chunk_overlap=64)
+EMBEDDING_MODEL = "togethercomputer/m2-bert-80M-2k-retrieval"
+api_key = (
+    os.getenv("TOGETHER_API")
+    or "0d1849365485f54f5deb32458276cb348948608da5a89dad0efc780c2d356916"
+)
+ai_client = OpenAI(api_key=api_key, base_url="https://api.together.xyz/v1")
+
+# Setup/Initiate SQLite database for metadata store
+conn = sqlite3.connect("metadata.db")
+cursor = conn.cursor()
+cursor.execute(
+    """
+CREATE TABLE IF NOT EXISTS metadata_store (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    meta_dict TEXT NOT NULL
+)
+"""
+)
+conn.commit()
+
+
+async def full_processor(files: List[UploadFile], user_tags: str, collection_name: str):
+    user_metadata = {"tags": user_tags}
+
+    file_names = [str(file.filename) for file in files]
+    file_types = [name.split(".")[-1] for name in file_names]
+
+    processed_docs = await processor(files, file_types)
+
+    chunks, metadata = create_chunks(processed_docs, file_names, user_metadata)
+
+    response = ai_client.embeddings.create(input=chunks, model=EMBEDDING_MODEL)
+    embeddings = [item.embedding for item in response.data]
+
+    write_metadata(file_names, file_types, user_tags, len(chunks), collection_name)
+
+    return (chunks, metadata, embeddings)
+
+
+async def processor(files: List[UploadFile], file_types: List[str]) -> List[str]:
+    processed_docs = []
+    for idx, file_type in enumerate(file_types):
+        if file_type == "pdf":
+            processed_docs.append(await process_pdf(files[idx]))
+        elif file_type == "docx":
+            processed_docs.append(await process_docx(files[idx]))
+
+    return processed_docs
+
+
+async def process_pdf(file: UploadFile) -> str:
+    data = await file.read()
+    doc = PdfReader(BytesIO(data))
+
+    text = ""
+    for page in doc.pages:
+        text += page.extract_text() or ""
+
+    return text
+
+
+async def process_docx(file: UploadFile) -> str:
+    data = await file.read()
+    doc = Document(BytesIO(data))
+
+    text = ""
+    for para in doc.paragraphs:
+        text += para.text + "\n"
+
+    return text
+
+
+def create_chunks(docs: List[str], file_names: List[str], user_metadata):
+    all_chunks = []
+    all_metadata = []
+
+    dateTime = datetime.datetime.now().strftime("%d/%m/%y-%I")
+
+    for idx, doc in enumerate(docs):
+        chunks = splitter.split_text(doc)
+        all_chunks += chunks
+        metadata = {
+            "file_name": file_names[idx],
+            "timestamp": dateTime,
+            **user_metadata,
+        }
+        all_metadata += [metadata] * len(chunks)
+
+    return all_chunks, all_metadata
+
+
+def write_metadata(file_names, file_types, user_tags, total_chunks, collection_name):
+
+    time = datetime.datetime.now().strftime("%d/%m/%y-%I")
+    pdf, docx, unsupported = 0, 0, 0
+    for i in file_types:
+        if i == "pdf":
+            pdf += 1
+        elif i == "docx":
+            docx += 1
+        else:
+            unsupported += 1
+
+    metadata = {
+        "Collection": collection_name,
+        "FileName": file_names,
+        "TotalPDF": pdf,
+        "TotalDocx": docx,
+        "Unsupported": unsupported,
+        "CustomTag": user_tags,
+        "TotalChunks": total_chunks,
+        "Time": time,
+    }
+    cursor.execute(
+        "INSERT INTO metadata_store (meta_dict) VALUES (?)", (json.dumps(metadata),)
+    )
+    conn.commit()
+
+
+def read_metadata():
+
+    cursor.execute("SELECT meta_dict FROM metadata_store")
+    rows = cursor.fetchall()
+
+    data = []
+    for row in rows:
+        meta = json.loads(row[0])
+        data.append(meta)
+
+    return data

generator/prompts.py

@@ -0,0 +1,4 @@
+SYS_PROMPT = """
+You are an AI assistant powered by a retrieval-augmented generation (RAG) system. You are provided with CONTEXT—semantically relevant information retrieved from a private knowledge base—based on the USER_QUERY.
+Your task is to generate a helpful, accurate, and context-aware response by combining your own knowledge with the retrieved CONTEXT. Always prioritize CONTEXT when it is available, but supplement it with your own understanding when appropriate. Ensure your response is clear, concise, and directly addresses the user's query.
+"""

generator/response.py

@@ -0,0 +1,54 @@
+from openai import OpenAI
+from typing import List
+import os
+
+from .prompts import SYS_PROMPT
+from data_ingetion.data import AdvancedDatabase
+
+api_key = (
+    os.getenv("TOGETHER_API")
+    or "0d1849365485f54f5deb32458276cb348948608da5a89dad0efc780c2d356916"
+)
+client = OpenAI(api_key=api_key, base_url="https://api.together.xyz/v1")
+EMBEDDING_MODEL = "togethercomputer/m2-bert-80M-2k-retrieval"
+
+
+def generate_response(query: str, group_name: str, return_chunks: bool = True):
+    db = AdvancedDatabase()
+    query_embedding = get_embedding(query)
+    context = db.get_context(query_embedding, group_name)
+    response = llm_response(context, query)
+    if return_chunks:
+        return (response, context)
+    return response
+
+
+def llm_response(context: List[str], user_query: str, history={}, stream: bool = False):
+    response = client.chat.completions.create(
+        model="meta-llama/Llama-3.3-70B-Instruct-Turbo",
+        messages=[
+            {
+                "role": "system",
+                "content": SYS_PROMPT,
+            },
+            *history,
+            {
+                "role": "user",
+                "content": f"Query : {user_query} \n\n Context: {context}",
+            },
+        ],
+        stream=False,
+    )
+
+    # Feature to add streaming response
+    # if stream:
+    #    for chunk in response:
+    #        yield chunk.choices[0].delta.content or ""
+    # else:
+    return response.choices[0].message.content
+
+
+def get_embedding(query: str):
+    response = client.embeddings.create(input=query, model=EMBEDDING_MODEL)
+    embeddings = response.data[0].embedding
+    return embeddings

requirements.txt

@@ -0,0 +1,113 @@
+annotated-types==0.7.0
+anyio==4.9.0
+asgiref==3.8.1
+attrs==25.3.0
+backoff==2.2.1
+bcrypt==4.3.0
+build==1.2.2.post1
+cachetools==5.5.2
+certifi==2025.4.26
+charset-normalizer==3.4.2
+chromadb==1.0.12
+click==8.2.1
+coloredlogs==15.0.1
+distro==1.9.0
+dnspython==2.7.0
+durationpy==0.10
+email_validator==2.2.0
+exceptiongroup==1.3.0
+fastapi==0.115.9
+fastapi-cli==0.0.7
+filelock==3.18.0
+flatbuffers==25.2.10
+fsspec==2025.5.1
+google-auth==2.40.3
+googleapis-common-protos==1.70.0
+grpcio==1.72.1
+h11==0.16.0
+hf-xet==1.1.3
+httpcore==1.0.9
+httptools==0.6.4
+httpx==0.28.1
+huggingface-hub==0.32.4
+humanfriendly==10.0
+idna==3.10
+importlib_metadata==8.7.0
+importlib_resources==6.5.2
+Jinja2==3.1.6
+jiter==0.10.0
+jsonpatch==1.33
+jsonpointer==3.0.0
+jsonschema==4.24.0
+jsonschema-specifications==2025.4.1
+kubernetes==32.0.1
+langchain-core==0.3.64
+langchain-text-splitters==0.3.8
+langsmith==0.3.45
+lxml==5.4.0
+markdown-it-py==3.0.0
+MarkupSafe==3.0.2
+mdurl==0.1.2
+mmh3==5.1.0
+mpmath==1.3.0
+numpy==2.2.6
+oauthlib==3.2.2
+onnxruntime==1.22.0
+openai==1.84.0
+opentelemetry-api==1.34.0
+opentelemetry-exporter-otlp-proto-common==1.34.0
+opentelemetry-exporter-otlp-proto-grpc==1.34.0
+opentelemetry-instrumentation==0.55b0
+opentelemetry-instrumentation-asgi==0.55b0
+opentelemetry-instrumentation-fastapi==0.55b0
+opentelemetry-proto==1.34.0
+opentelemetry-sdk==1.34.0
+opentelemetry-semantic-conventions==0.55b0
+opentelemetry-util-http==0.55b0
+orjson==3.10.18
+overrides==7.7.0
+packaging==24.2
+posthog==4.2.0
+protobuf==5.29.5
+pyasn1==0.6.1
+pyasn1_modules==0.4.2
+pydantic==2.11.5
+pydantic_core==2.33.2
+Pygments==2.19.1
+pypdf==5.6.0
+PyPika==0.48.9
+pyproject_hooks==1.2.0
+python-dateutil==2.9.0.post0
+python-docx==1.1.2
+python-dotenv==1.1.0
+python-multipart==0.0.20
+PyYAML==6.0.2
+referencing==0.36.2
+requests==2.32.3
+requests-oauthlib==2.0.0
+requests-toolbelt==1.0.0
+rich==14.0.0
+rich-toolkit==0.14.7
+rpds-py==0.25.1
+rsa==4.9.1
+shellingham==1.5.4
+six==1.17.0
+sniffio==1.3.1
+starlette==0.45.3
+sympy==1.14.0
+tenacity==9.1.2
+tokenizers==0.21.1
+tomli==2.2.1
+tqdm==4.67.1
+typer==0.16.0
+typing-inspection==0.4.1
+typing_extensions==4.14.0
+urllib3==2.4.0
+uvicorn==0.34.3
+uvloop==0.21.0
+watchfiles==1.0.5
+websocket-client==1.8.0
+websockets==15.0.1
+wrapt==1.17.2
+zipp==3.22.0
+zstandard==0.23.0

test/unit_test.py

@@ -0,0 +1,56 @@
+import requests
+from uuid import uuid4
+
+base_url = "http://localhost:7860"
+ingest_endpoint = "/ingest_files"
+generator_endpoint = "/fetch_response"
+get_metadata_endpoint = "/get_metadata"
+
+collection_name = str(uuid4())
+testcase_passed = 0
+total_tests = 3
+
+
+def test_ingest_files():
+    global testcase_passed
+    metadata = {"metadata": "legal,nda,confidential", "group_name": collection_name}
+    with open("test/sample1.pdf", "rb") as f:
+        files = [("files", ("sample1.pdf", f, "application/pdf"))]
+        response = requests.post(base_url + ingest_endpoint, files=files, data=metadata)
+    print("Testing:", ingest_endpoint)
+    print("Status:", response.status_code)
+    print("Response:", response.content)
+    assert response.status_code == 200, "Ingest endpoint failed"
+    testcase_passed += 1
+
+
+def test_llm_generation():
+    global testcase_passed
+    params = {"query": "How was past year performance", "group_name": collection_name}
+    response = requests.post(base_url + generator_endpoint, params=params)
+    print("Testing:", generator_endpoint)
+    print("Status:", response.status_code)
+    print("Response:", response.content)
+    assert response.status_code == 200, "LLM generation endpoint failed"
+    testcase_passed += 1
+
+
+def test_get_metadata():
+    global testcase_passed
+    response = requests.get(base_url + get_metadata_endpoint)
+    print("Testing:", get_metadata_endpoint)
+    print("Status:", response.status_code)
+    print("Response:", response.content)
+    assert response.status_code == 200, "Metadata endpoint failed"
+    testcase_passed += 1
+
+
+if __name__ == "__main__":
+    try:
+        test_ingest_files()
+        test_llm_generation()
+        test_get_metadata()
+    except AssertionError as e:
+        print("Test failed:", e)
+
+    print(f"{testcase_passed}/{total_tests} tests passed")