deploy FastAPI backend

app/auth/jwt_handler.py

@@ -0,0 +1,57 @@
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, Optional
+from jose import jwt, JWTError
+from passlib.hash import argon2
+from fastapi import HTTPException, status
+from app.config import settings
+
+SECRET_KEY = settings.secret_key
+ALGORITHM = settings.algorithm
+ACCESS_TOKEN_EXPIRE_MINUTES = settings.access_token_expire_minutes
+REFRESH_TOKEN_EXPIRE_DAYS = settings.refresh_token_expire_days
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+def create_access_token(subject: str, role: Optional[str] = None) -> str:
+    expire = _now() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    payload = {"sub": subject, "exp": expire, "iat": _now(), "type": "access"}
+    if role:
+        payload["role"] = role
+    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
+
+def create_refresh_token(subject: str) -> str:
+    expire = _now() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
+    payload = {"sub": subject, "exp": expire, "iat": _now(), "type": "refresh"}
+    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
+
+def decode_token(token: str) -> Dict[str, Any]:
+    try:
+        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+    except JWTError:
+        raise ValueError("Invalid token or signature")
+
+def verify_access_token(token: str) -> str:
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        subject: Optional[str] = payload.get("sub")
+        token_type = payload.get("type")
+        if subject is None or token_type != "access":
+            raise credentials_exception
+        return subject
+    except JWTError:
+        raise credentials_exception
+
+def hash_refresh_token(raw_refresh: str) -> str:
+    return argon2.hash(raw_refresh)
+
+def verify_refresh_token(raw_refresh: str, hash_value: str) -> bool:
+    try:
+        return argon2.verify(raw_refresh, hash_value)
+    except Exception:
+        return False

app/auth/models.py

@@ -0,0 +1,28 @@
+from datetime import datetime
+from typing import Optional
+from pydantic import BaseModel, Field, EmailStr
+
+class UserCreate(BaseModel):
+    username: str = Field(..., min_length=3, max_length=50)
+    email: EmailStr = Field(..., description="User email (must be unique)")
+    company: str = Field(default="", max_length=128)
+    password: str = Field(..., min_length=8, description="User password (will be hashed).")
+
+class UserInDB(BaseModel):
+    username: str
+    email: str
+    company: str
+    hashed_password: str
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+
+class UserPublic(BaseModel):
+    username: str
+    email: str
+    company: str
+    created_at: datetime
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str = "bearer"
+    refresh_token: str
+    expires_in: int

app/auth/routes.py

@@ -0,0 +1,145 @@
+from datetime import datetime, timezone
+from typing import Optional
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordRequestForm, OAuth2PasswordBearer
+from motor.motor_asyncio import AsyncIOMotorDatabase
+from passlib.context import CryptContext
+from pydantic import BaseModel
+
+from app.database.connection import get_db
+from app.auth.models import UserCreate, UserPublic, Token
+from app.auth.jwt_handler import (
+    create_access_token,
+    create_refresh_token,
+    decode_token,
+    hash_refresh_token,
+    verify_access_token,
+    verify_refresh_token,
+)
+from app.config import settings
+
+router = APIRouter(prefix="/auth", tags=["Authentication"])
+pwd_context = CryptContext(schemes=["argon2"], deprecated="auto")
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
+
+async def get_current_user(
+    token: str = Depends(oauth2_scheme),
+    db: AsyncIOMotorDatabase = Depends(get_db),
+) -> UserPublic:
+    try:
+        username = verify_access_token(token)
+    except HTTPException as e:
+        raise e
+    except Exception:
+        raise HTTPException(status_code=401, detail="Invalid token")
+    user = await db.users.find_one({"username": username})
+    if not user:
+        raise HTTPException(status_code=401, detail="User not found")
+    return UserPublic(**user)
+
+@router.post("/register", response_model=UserPublic, status_code=status.HTTP_201_CREATED)
+async def register(user: UserCreate, db: AsyncIOMotorDatabase = Depends(get_db)):
+    username = user.username.strip().lower()
+    email = user.email.lower()
+    if await db.users.find_one({"$or": [{"username": username}, {"email": email}]}):
+        raise HTTPException(status_code=400, detail="Username or email already exists")
+    hashed = pwd_context.hash(user.password)
+    doc = {
+        "username": username,
+        "email": email,
+        "company": user.company,
+        "hashed_password": hashed,
+        "created_at": datetime.utcnow(),
+    }
+    await db.users.insert_one(doc)
+    return UserPublic(**doc)
+
+@router.post("/login", response_model=Token)
+async def login(form_data: OAuth2PasswordRequestForm = Depends(), db=Depends(get_db)):
+    username = form_data.username.strip().lower()
+    user = await db.users.find_one({"username": username})
+    if not user or not pwd_context.verify(form_data.password, user["hashed_password"]):
+        raise HTTPException(status_code=401, detail="Incorrect username or password")
+    access_token = create_access_token(username)
+    refresh_token = create_refresh_token(username)
+    payload = decode_token(refresh_token)
+    await db.sessions.insert_one(
+        {
+            "user_id": username,
+            "refresh_token_hash": hash_refresh_token(refresh_token),
+            "created_at": datetime.now(timezone.utc),
+            "expires_at": datetime.fromtimestamp(payload["exp"], tz=timezone.utc),
+            "revoked_at": None,
+        }
+    )
+    return Token(
+        access_token=access_token,
+        refresh_token=refresh_token,
+        expires_in=settings.access_token_expire_minutes * 60,
+    )
+
+class RefreshIn(BaseModel):
+    refresh_token: str
+
+@router.post("/refresh", response_model=Token)
+async def refresh_token(payload: RefreshIn, db=Depends(get_db)):
+    try:
+        decoded = decode_token(payload.refresh_token)
+    except ValueError:
+        raise HTTPException(status_code=401, detail="Invalid refresh token")
+    if decoded.get("type") != "refresh":
+        raise HTTPException(status_code=401, detail="Invalid token type")
+    username = decoded.get("sub")
+    session_doc = await db.sessions.find_one(
+        {
+            "user_id": username,
+            "revoked_at": None,
+            "expires_at": {"$gt": datetime.now(timezone.utc)},
+        },
+        sort=[("created_at", -1)],
+    )
+    if not session_doc or not verify_refresh_token(payload.refresh_token, session_doc["refresh_token_hash"]):
+        raise HTTPException(status_code=401, detail="Refresh token not recognized")
+    new_access = create_access_token(username)
+    new_refresh = create_refresh_token(username)
+    await db.sessions.update_one(
+        {"_id": session_doc["_id"]}, {"$set": {"revoked_at": datetime.now(timezone.utc)}}
+    )
+    payload_new = decode_token(new_refresh)
+    await db.sessions.insert_one(
+        {
+            "user_id": username,
+            "refresh_token_hash": hash_refresh_token(new_refresh),
+            "created_at": datetime.now(timezone.utc),
+            "expires_at": datetime.fromtimestamp(payload_new["exp"], tz=timezone.utc),
+            "revoked_at": None,
+        }
+    )
+    return Token(access_token=new_access, refresh_token=new_refresh, expires_in=settings.access_token_expire_minutes * 60)
+
+@router.post("/logout")
+async def logout(payload: RefreshIn, db=Depends(get_db)):
+    try:
+        decoded = decode_token(payload.refresh_token)
+    except ValueError:
+        return {"ok": True}
+    username = decoded.get("sub")
+    session_doc = await db.sessions.find_one(
+        {
+            "user_id": username,
+            "revoked_at": None,
+            "expires_at": {"$gt": datetime.now(timezone.utc)},
+        },
+        sort=[("created_at", -1)],
+    )
+    if not session_doc:
+        return {"ok": True}
+    if verify_refresh_token(payload.refresh_token, session_doc["refresh_token_hash"]):
+        await db.sessions.update_one(
+            {"_id": session_doc["_id"]}, {"$set": {"revoked_at": datetime.now(timezone.utc)}}
+        )
+    return {"ok": True}
+
+@router.get("/profile", response_model=UserPublic)
+async def read_users_me(current_user: UserPublic = Depends(get_current_user)):
+    return current_user

app/config.py

@@ -0,0 +1,17 @@
+import os
+from pydantic_settings import BaseSettings
+
+class Settings(BaseSettings):
+    mongo_uri: str = os.getenv("MONGO_URI")
+    database_name: str = os.getenv("DATABASE_NAME")
+    groq_api_key: str = os.getenv("GROQ_API_KEY")
+    secret_key: str = os.getenv("SECRET_KEY")
+    algorithm: str = os.getenv("ALGORITHM")
+    access_token_expire_minutes: int = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", 30))
+    refresh_token_expire_days: int = int(os.getenv("REFRESH_TOKEN_EXPIRE_DAYS", 7))
+
+    class Config:
+        env_file = ".env"
+        env_file_encoding = "utf-8"
+
+settings = Settings()

app/database/connection.py

@@ -0,0 +1,7 @@
+from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase
+from app.config import settings
+
+client = AsyncIOMotorClient(settings.mongo_uri)
+
+async def get_db() -> AsyncIOMotorDatabase:
+    return client[settings.database_name]

app/database/schemas.py

@@ -0,0 +1,73 @@
+from typing import List, Optional, Dict, Any
+from datetime import datetime
+from pydantic import BaseModel, Field, EmailStr
+
+# User Schemas (adapted)
+class UserCreate(BaseModel):
+    username: str = Field(..., min_length=3, max_length=50)
+    email: EmailStr = Field(..., description="User email (must be unique)")
+    company: str = Field(default="", max_length=128)
+    password: str = Field(..., min_length=8, description="User password (will be hashed).")
+
+class UserDB(BaseModel):
+    id: Optional[str] = Field(None, alias="_id")
+    username: str
+    email: str
+    password_hash: str = Field(alias="hashed_password")
+    company: str
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    updated_at: datetime = Field(default_factory=datetime.utcnow)
+    is_active: bool = True
+    roles: List[str] = Field(default_factory=lambda: ["user"])
+
+class SessionDB(BaseModel):
+    id: Optional[str] = Field(None, alias="_id")
+    user_id: str
+    refresh_token_hash: str
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    expires_at: datetime
+    revoked_at: Optional[datetime] = None
+    meta: Dict[str, Any] = Field(default_factory=dict)
+
+# Conversation Schemas
+class Message(BaseModel):
+    role: str
+    content: str
+
+class ConversationDB(BaseModel):
+    id: Optional[str] = Field(None, alias="_id")
+    user_id: str
+    messages: List[Message] = Field(default_factory=list)
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    updated_at: datetime = Field(default_factory=datetime.utcnow)
+    meta: Dict[str, Any] = Field(default_factory=dict)  # e.g., {"model": "llama-3.1-8b-instant"}
+
+# Audit Log (optional, for security)
+class AuditLogDB(BaseModel):
+    id: Optional[str] = Field(None, alias="_id")
+    user_id: Optional[str] = None
+    action: str
+    ip: Optional[str] = None
+    user_agent: Optional[str] = None
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    meta: Dict[str, Any] = Field(default_factory=dict)
+
+# MongoDB Indexes for Performance
+MONGO_INDEXES = {
+    "users": [
+        {"keys": [("username", 1)], "unique": True},
+        {"keys": [("email", 1)], "unique": True},
+        {"keys": [("created_at", -1)]},
+    ],
+    "sessions": [
+        {"keys": [("user_id", 1), ("created_at", -1)]},
+        {"keys": [("expires_at", 1)]},
+    ],
+    "conversations": [
+        {"keys": [("user_id", 1), ("created_at", -1)]},
+    ],
+    "audit_logs": [
+        {"keys": [("user_id", 1), ("created_at", -1)]},
+        {"keys": [("action", 1), ("created_at", -1)]},
+    ],
+}

app/logging_config.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+import json, sys, os, logging
+from logging.config import dictConfig
+from uvicorn.config import LOG_LEVELS
+
+SERVICE_NAME = os.getenv("SERVICE_NAME", "backend")
+ENV = os.getenv("ENV", "production")
+
+class JsonFormatter(logging.Formatter):
+    def format(self, record: logging.LogRecord) -> str:
+        base = {
+            "level": record.levelname,
+            "logger": record.name,
+            "msg": record.getMessage(),
+            "time": self.formatTime(record, self.datefmt),
+            "service": SERVICE_NAME,
+            "env": ENV,
+        }
+        if record.exc_info:
+            base["exc_info"] = self.formatException(record.exc_info)
+        return json.dumps(base, ensure_ascii=False)
+
+def setup_logging():
+    log_level = os.getenv("LOG_LEVEL", "INFO").upper()
+    if log_level not in LOG_LEVELS:
+        log_level = "INFO"
+
+    dictConfig({
+        "version": 1,
+        "disable_existing_loggers": False,
+        "formatters": {
+            "json": {"()": JsonFormatter},
+            "plain": {"format": "%(levelname)s:%(name)s:%(message)s"},
+        },
+        "handlers": {
+            "console": {
+                "class": "logging.StreamHandler",
+                "stream": sys.stdout,
+                "formatter": "json" if ENV == "production" else "plain",
+                "level": log_level,
+            },
+        },
+        "loggers": {
+            "uvicorn": {"handlers": ["console"], "level": log_level, "propagate": False},
+            "uvicorn.error": {"handlers": ["console"], "level": log_level, "propagate": False},
+            "uvicorn.access": {"handlers": ["console"], "level": log_level, "propagate": False},
+            "fastapi": {"handlers": ["console"], "level": log_level, "propagate": False},
+            "": {"handlers": ["console"], "level": log_level},  # root logger
+        },
+    })

app/main.py

@@ -0,0 +1,24 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from app.auth.routes import router as auth_router
+from app.rag.routes import router as rag_router
+from app.logging_config import setup_logging
+
+setup_logging()
+
+app = FastAPI(title="GrokRAG API", description="SaaS RAG Chat with Groq", version="1.1.0")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["http://localhost:3000"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+app.include_router(auth_router)
+app.include_router(rag_router, prefix="/rag")
+
+@app.get("/")
+async def root():
+    return {"message": "Welcome to GrokRAG API"}

app/rag/models.py

@@ -0,0 +1,39 @@
+from typing import List, Optional
+from pydantic import BaseModel, Field
+from bson import ObjectId
+from datetime import datetime
+
+ALLOWED_MODELS = [
+    "allam-2-7b",  # Fixed typo from "allam-2-7b"
+    "groq/compound",
+    "groq/compound-mini",
+    "llama-3.1-8b-instant",
+    "llama-3.3-70b-versatile",
+    "llama-3.1-70b-versatile",  # Assuming typo from "llama-3.3-70b-versatile"
+    "meta-llama/llama-4-maverick-17b-128e-instruct",
+    "meta-llama/llama-4-scout-17b-16e-instruct",
+    "meta-llama/llama-guard-4-12b",
+    "meta-llama/llama-prompt-guard-2-22m",
+    "meta-llama/llama-prompt-guard-2-86m",
+    "moonshotai/kimi-k2-instruct",
+    "moonshotai/kimi-k2-instruct-0905",
+    "openai/gpt-oss-120b",
+    "openai/gpt-oss-20b",
+    "openai/gpt-oss-safeguard-20b",
+    "qwen/qwen3-32b",
+]
+
+class Message(BaseModel):
+    role: str
+    content: str
+
+class Conversation(BaseModel):
+    id: Optional[str] = None  # str(ObjectId)
+    user_id: str
+    messages: List[Message] = []
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+
+class ChatRequest(BaseModel):
+    model: str = Field(..., description="Groq model", enum=ALLOWED_MODELS)
+    enable_web_search: bool = False
+    message: str = Field(..., min_length=1)

app/rag/rag_processor.py

@@ -0,0 +1,151 @@
+import os
+import re
+import logging
+from typing import List, Tuple, Optional
+import faiss
+from sentence_transformers import SentenceTransformer
+from PyPDF2 import PdfReader
+from docx import Document
+import pytesseract
+from PIL import Image
+import io
+import openpyxl
+import pandas as pd
+from duckduckgo_search import DDGS
+from fastapi import UploadFile
+
+logger = logging.getLogger(__name__)
+
+_EMBED_MODEL_NAME = "sentence-transformers/all-MiniLM-L6-v2"
+_embedder: Optional[SentenceTransformer] = None
+
+def _get_embedder() -> SentenceTransformer:
+    global _embedder
+    if _embedder is None:
+        logger.info(f"Loading embedding model: {_EMBED_MODEL_NAME}")
+        _embedder = SentenceTransformer(_EMBED_MODEL_NAME)
+    return _embedder
+
+# Enhanced File Extraction
+def extract_text(file: UploadFile) -> str:
+    ext = os.path.splitext(file.filename)[1].lower()
+    content = file.file.read()
+    file_bytes = io.BytesIO(content)
+    if ext == ".pdf":
+        try:
+            reader = PdfReader(file_bytes)
+            return "\n".join(page.extract_text() or "" for page in reader.pages)
+        except Exception as e:
+            logger.error(f"PDF extract failed: {e}")
+            return ""
+    elif ext == ".docx":
+        try:
+            doc = Document(file_bytes)
+            return "\n".join(p.text for p in doc.paragraphs if p.text)
+        except Exception as e:
+            logger.error(f"DOCX extract failed: {e}")
+            return ""
+    elif ext in [".xlsx", ".xls"]:
+        try:
+            wb = openpyxl.load_workbook(file_bytes, read_only=True, data_only=True)
+            text = []
+            for sheet in wb:
+                for row in sheet.iter_rows(values_only=True):
+                    text.append(" ".join(str(cell) for cell in row if cell is not None))
+            return "\n".join(text)
+        except Exception as e:
+            logger.error(f"Excel extract failed: {e}")
+            return ""
+    elif ext == ".csv":
+        try:
+            df = pd.read_csv(file_bytes)
+            return df.to_string()
+        except Exception as e:
+            logger.error(f"CSV extract failed: {e}")
+            return ""
+    elif ext in [".jpg", ".jpeg", ".png", ".gif"]:  # OCR for images
+        try:
+            img = Image.open(file_bytes)
+            return pytesseract.image_to_string(img)
+        except Exception as e:
+            logger.error(f"Image OCR failed: {e}")
+            return ""
+    else:  # Fallback text
+        try:
+            return content.decode("utf-8", errors="ignore")
+        except Exception as e:
+            logger.error(f"Text extract failed: {e}")
+            return ""
+
+def clean_text(text: str) -> str:
+    t = re.sub(r"[ \t]+", " ", text)
+    t = re.sub(r"\n{3,}", "\n\n", t)
+    return t.strip()
+
+def chunk_text(text: str, max_tokens: int = 400, overlap: int = 50) -> List[str]:
+    text = clean_text(text)
+    if not text:
+        return []
+    words = text.split()
+    chunks, start = [], 0
+    while start < len(words):
+        end = min(len(words), start + max_tokens)
+        chunk = " ".join(words[start:end]).strip()
+        if chunk:
+            chunks.append(chunk)
+        if end == len(words):
+            break
+        start = max(0, end - overlap)
+    return chunks
+
+class RagIndex:
+    def __init__(self, index: faiss.IndexFlatIP, dim: int, chunks: List[str]):
+        self.index = index
+        self.dim = dim
+        self.chunks = chunks
+
+def build_faiss_index(chunks: List[str]) -> RagIndex:
+    emb = _get_embedder()
+    vectors = emb.encode(chunks, convert_to_numpy=True, normalize_embeddings=True)
+    dim = vectors.shape[1]
+    index = faiss.IndexFlatIP(dim)
+    index.add(vectors)
+    return RagIndex(index=index, dim=dim, chunks=chunks)
+
+def search(index: RagIndex, query: str, top_k: int = 6) -> List[Tuple[str, float]]:
+    emb = _get_embedder()
+    q = emb.encode([query], convert_to_numpy=True, normalize_embeddings=True)
+    D, I = index.index.search(q, top_k)
+    hits = []
+    for score, idx in zip(D[0], I[0]):
+        if idx == -1:
+            continue
+        hits.append((index.chunks[idx], float(score)))
+    return hits
+
+def build_context_from_files(files: List[UploadFile], prompt: str, top_k: int = 6) -> str:
+    all_text = []
+    for file in files:
+        txt = extract_text(file)
+        if txt:
+            all_text.append(txt)
+        file.file.seek(0)  # Reset
+    big_text = "\n\n".join(all_text)
+    chunks = chunk_text(big_text, max_tokens=450, overlap=80)
+    if not chunks:
+        return ""
+    idx = build_faiss_index(chunks)
+    hits = search(idx, prompt, top_k=top_k)
+    context_sections = [f"[DOC#{i} score={score:.3f}]\n{chunk}" for i, (chunk, score) in enumerate(hits, 1)]
+    return "\n\n".join(context_sections)
+
+# Web search tool
+def web_search(query: str) -> str:
+    try:
+        with DDGS() as ddgs:
+            results = [r for r in ddgs.text(query, max_results=5)]
+        sections = [f"[WEB#{i}] Title: {r['title']}\nSnippet: {r['body']}\nURL: {r['href']}" for i, r in enumerate(results, 1)]
+        return "\n\n".join(sections) if sections else "No results found."
+    except Exception as e:
+        logger.error(f"Web search failed: {e}")
+        return "Web search error."

app/rag/routes.py

@@ -0,0 +1,168 @@
+from typing import List
+from fastapi import APIRouter, Depends, HTTPException, status, UploadFile, Form
+from fastapi.responses import StreamingResponse
+from motor.motor_asyncio import AsyncIOMotorDatabase
+from bson import ObjectId
+from groq import Groq
+import json
+import logging
+from datetime import datetime
+
+from app.database.connection import get_db
+from app.database.schemas import ConversationDB
+from app.auth.routes import get_current_user
+from app.auth.models import UserPublic
+from app.rag.models import ALLOWED_MODELS, Message
+from app.rag.rag_processor import build_context_from_files, web_search
+from app.config import settings
+
+router = APIRouter(tags=["RAG Chat"])
+logger = logging.getLogger(__name__)
+
+SYSTEM_PROMPT = """You are a helpful assistant. Use the provided context if relevant. If web search is enabled and you need up-to-date information, use the web_search tool. Reason step-by-step before deciding to use tools."""
+
+WEB_SEARCH_TOOL = {
+    "type": "function",
+    "function": {
+        "name": "web_search",
+        "description": "Search the web using DuckDuckGo for up-to-date information.",
+        "parameters": {
+            "type": "object",
+            "properties": {"query": {"type": "string", "description": "The search query"}},
+            "required": ["query"],
+        },
+    },
+}
+
+@router.post("/conversations", status_code=status.HTTP_201_CREATED)
+async def create_conversation(
+    current_user: UserPublic = Depends(get_current_user),
+    db: AsyncIOMotorDatabase = Depends(get_db),
+):
+    conv = ConversationDB(user_id=current_user.username)
+    result = await db.conversations.insert_one(conv.dict(exclude={"id"}))
+    conv.id = str(result.inserted_id)
+    return {"conversation_id": conv.id}
+
+@router.get("/conversations/{conv_id}")
+async def get_conversation(
+    conv_id: str,
+    current_user: UserPublic = Depends(get_current_user),
+    db: AsyncIOMotorDatabase = Depends(get_db),
+):
+    try:
+        oid = ObjectId(conv_id)
+    except:
+        raise HTTPException(status_code=400, detail="Invalid conversation ID")
+    conv = await db.conversations.find_one({"_id": oid, "user_id": current_user.username})
+    if not conv:
+        raise HTTPException(status_code=404, detail="Conversation not found")
+    conv["id"] = str(conv["_id"])
+    del conv["_id"]
+    return conv
+
+@router.post("/conversations/{conv_id}/messages")
+async def send_message(
+    conv_id: str,
+    model: str = Form(...),
+    enable_web_search: bool = Form(False),
+    message: str = Form(...),
+    files: List[UploadFile] = None,
+    current_user: UserPublic = Depends(get_current_user),
+    db: AsyncIOMotorDatabase = Depends(get_db),
+):
+    if model not in ALLOWED_MODELS:
+        raise HTTPException(status_code=400, detail="Invalid model")
+    try:
+        oid = ObjectId(conv_id)
+    except:
+        raise HTTPException(status_code=400, detail="Invalid conversation ID")
+    conv = await db.conversations.find_one({"_id": oid, "user_id": current_user.username})
+    if not conv:
+        raise HTTPException(status_code=404, detail="Conversation not found")
+
+    # Load messages
+    messages = [Message(**m) for m in conv.get("messages", [])]
+
+    # Build RAG context if files
+    rag_context = ""
+    if files:
+        rag_context = build_context_from_files(files, message)
+    
+    # System prompt with context
+    system_msg = {"role": "system", "content": SYSTEM_PROMPT + (f"\n\nContext: {rag_context}" if rag_context else "")}
+
+    # Append user message
+    user_msg = Message(role="user", content=message)
+    messages.append(user_msg)
+
+    # Groq client
+    client = Groq(api_key=settings.groq_api_key)
+
+    # Tools if enabled
+    tools = [WEB_SEARCH_TOOL] if enable_web_search else None
+
+    # Tool loop for reasoning and multiple calls (up to 3 iterations)
+    chat_history = [
+        system_msg if isinstance(system_msg, dict) else system_msg.dict()
+    ] + [
+            m if isinstance(m, dict) else m.dict() for m in messages
+            ]
+    max_tool_loops = 3
+    for _ in range(max_tool_loops):
+        completion = client.chat.completions.create(
+            model=model,
+            messages=chat_history,
+            temperature=1,
+            max_tokens=8192,
+            top_p=1,
+            stream=False,
+            stop=None,
+            tools=tools,
+        )
+        choice = completion.choices[0].message
+        if not choice.tool_calls:
+            # No more tools, prepare to stream
+            break
+        for tool_call in choice.tool_calls:
+            if tool_call.function.name == "web_search":
+                args = json.loads(tool_call.function.arguments)
+                query = args["query"]
+                result = web_search(query)
+                tool_response = {
+                    "role": "tool",
+                    "tool_call_id": tool_call.id,
+                    "name": "web_search",
+                    "content": result,
+                }
+                chat_history.append(tool_response)
+    else:
+        logger.warning("Max tool loops reached")
+        raise HTTPException(status_code=500, detail="Too many tool calls")
+
+    # Final streaming call
+    completion = client.chat.completions.create(
+        model=model,
+        messages=chat_history,
+        temperature=1,
+        max_tokens=8192,
+        top_p=1,
+        stream=True,
+        stop=None,
+    )
+
+    # Stream response
+    async def generate():
+        response_content = ""
+        for chunk in completion:
+            content = chunk.choices[0].delta.content or ""
+            response_content += content
+            yield content
+        # Save to DB
+        messages.append(Message(role="assistant", content=response_content))
+        await db.conversations.update_one(
+            {"_id": oid},
+            {"$set": {"messages": [m.dict() for m in messages], "updated_at": datetime.utcnow()}}
+        )
+
+    return StreamingResponse(generate(), media_type="text/event-stream")

app/request.py

@@ -0,0 +1,142 @@
+"""import requests
+import json
+
+# Base URL
+BASE_URL = "http://localhost:8000"
+
+# 1. Register a new user
+register_data = {
+    "username": "testuser",
+    "email": "test@example.com",
+    "company": "Test Co",
+    "password": "securepassword123"
+}
+response = requests.post(f"{BASE_URL}/auth/register", json=register_data)
+print("Register:", response.json())
+
+# 2. Login (get access/refresh tokens)
+login_data = {
+    "username": "testuser",
+    "password": "securepassword123"
+}
+response = requests.post(f"{BASE_URL}/auth/login", data=login_data)
+tokens = response.json()
+access_token = tokens["access_token"]
+refresh_token = tokens["refresh_token"]
+print("Login:", tokens)
+
+# Headers for authenticated requests
+headers = {"Authorization": f"Bearer {access_token}"}
+
+# 3. Create a conversation
+response = requests.post(f"{BASE_URL}/rag/conversations", headers=headers)
+conv_id = response.json()["conversation_id"]
+print("Conversation ID:", conv_id)
+
+# 4. Send a message (with optional files, web search)
+# Example: Text-only message
+files = []  # Or: [('files', open('doc.pdf', 'rb'))] for uploads
+data = {
+    "model": "llama-3.1-8b-instant",
+    "enable_web_search": True,
+    "message": "What is the capital of France?"
+}
+response = requests.post(
+    f"{BASE_URL}/rag/conversations/{conv_id}/messages",
+    headers=headers,
+    data=data,
+    files=files if files else None,
+    stream=True
+)
+for chunk in response.iter_content(chunk_size=1024):
+    if chunk:
+        print(chunk.decode(), end='', flush=True)  # Streaming output
+
+# 5. Get conversation history
+response = requests.get(f"{BASE_URL}/rag/conversations/{conv_id}", headers=headers)
+print("History:", response.json())
+
+# 6. Refresh token
+refresh_data = {"refresh_token": refresh_token}
+response = requests.post(f"{BASE_URL}/auth/refresh", json=refresh_data)
+new_tokens = response.json()
+print("New Tokens:", new_tokens)
+
+# 7. Logout
+logout_data = {"refresh_token": refresh_token}
+response = requests.post(f"{BASE_URL}/auth/logout", json=logout_data)
+print("Logout:", response.json())"""
+import requests
+import json
+
+# Base URL
+BASE_URL = "http://localhost:8000"
+
+# 1. Login (get access/refresh tokens) - Change credentials if needed
+login_data = {
+    "username": "testuser",        # Update if your username is different
+    "password": "securepassword123"  # Update with your actual password
+}
+response = requests.post(f"{BASE_URL}/auth/login", data=login_data)
+
+if response.status_code != 200:
+    print("Login Failed:", response.status_code, response.text)
+else:
+    tokens = response.json()
+    access_token = tokens["access_token"]
+    refresh_token = tokens["refresh_token"]
+    print("Login Success:", tokens)
+
+    # Headers for authenticated requests
+    headers = {"Authorization": f"Bearer {access_token}"}
+
+    # 2. Create a conversation
+    response = requests.post(f"{BASE_URL}/rag/conversations", headers=headers)
+    if response.status_code == 201:
+        conv_id = response.json()["conversation_id"]
+        print("Conversation Created - ID:", conv_id)
+    else:
+        print("Failed to create conversation:", response.status_code, response.text)
+        conv_id = None
+
+    if conv_id:
+        # 3. Send a message (text-only example)
+        data = {
+            "model": "llama-3.1-8b-instant",   # Change model if desired (from ALLOWED_MODELS)
+            "enable_web_search": "true",       # "true" or "false" as string for form data
+            "message": "What is the capital of France?"
+        }
+        # Optional: Add files for document RAG
+        # files = [('files', open('your_document.pdf', 'rb'))]
+
+        response = requests.post(
+            f"{BASE_URL}/rag/conversations/{conv_id}/messages",
+            headers=headers,
+            data=data,
+            # files=files if 'files' in locals() else None,
+            stream=True
+        )
+
+        print("\n--- Assistant Response ---")
+        if response.status_code == 200:
+            for chunk in response.iter_content(chunk_size=1024, decode_unicode=True):
+                if chunk:
+                    print(chunk, end='', flush=True)
+            print("\n--- End of Response ---")
+        else:
+            print("Message Send Failed:", response.status_code)
+            print("Response:", response.text)
+
+        # 4. Get conversation history
+        response = requests.get(f"{BASE_URL}/rag/conversations/{conv_id}", headers=headers)
+        print("\nConversation History:", json.dumps(response.json(), indent=2))
+
+    # 5. Refresh token (optional)
+    refresh_data = {"refresh_token": refresh_token}
+    response = requests.post(f"{BASE_URL}/auth/refresh", json=refresh_data)
+    print("Token Refresh:", response.json() if response.status_code == 200 else response.text)
+
+    # 6. Logout (optional)
+    logout_data = {"refresh_token": refresh_token}
+    response = requests.post(f"{BASE_URL}/auth/logout", json=logout_data)
+    print("Logout:", response.json())

requirements.txt

@@ -0,0 +1,17 @@
+fastapi
+uvicorn==0.24.0
+pydantic==2.7.2
+motor==3.3.2
+passlib[argon2]==1.7.4
+python-jose[cryptography]==3.3.0
+sentence-transformers>=2.2.2,<3
+faiss-cpu==1.7.4
+PyPDF2==3.0.1
+python-docx==1.1.0
+duckduckgo-search==6.2.13
+huggingface_hub>=0.17.0,<1.0
+transformers<5,>=4.41.2
+tokenizers<0.20,>=0.19.1
+groq==0.11.0  # Or latest: pip install groq --upgrade
+python-multipart==0.0.9
+faiss-cpu==1.7.4