Rename app.py to app/main.py

app.py

@@ -1,432 +0,0 @@
-import os
-import re
-import glob
-import shutil
-import threading
-import logging
-from typing import List, Tuple, Optional
-
-import faiss
-import numpy as np
-import yaml
-from sentence_transformers import SentenceTransformer
-from fastapi import FastAPI, UploadFile, File, HTTPException
-from fastapi.staticfiles import StaticFiles
-from fastapi.responses import FileResponse, JSONResponse
-from pydantic import BaseModel
-
-# ─── Logging ──────────────────────────────────────────────────────────────────
-logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
-logger = logging.getLogger(__name__)
-
-# ─── Config ───────────────────────────────────────────────────────────────────
-_CONFIG_PATH = "./config.yaml"
-
-def load_config(path: str = _CONFIG_PATH) -> dict:
-    defaults = {
-        "kb": {"directory": "./kb", "index_directory": "./.cache/faiss"},
-        "models": {
-            "embedding": "sentence-transformers/all-MiniLM-L6-v2",
-            "qa": "google/flan-t5-small",
-        },
-        "chunking": {"chunk_size": 500, "overlap": 100},
-        "retrieval": {"top_k": 5, "min_similarity": 0.25, "max_sentences": 6},
-        "client": {"name": "Self-Service KB Assistant"},
-        "messages": {
-            "welcome": "Ask me anything about the documents in the knowledge base!",
-        },
-        "quick_actions": [],
-    }
-    if os.path.exists(path):
-        try:
-            with open(path, "r", encoding="utf-8") as f:
-                user_cfg = yaml.safe_load(f) or {}
-            for key, val in user_cfg.items():
-                if isinstance(val, dict) and key in defaults:
-                    defaults[key].update(val)
-                else:
-                    defaults[key] = val
-            logger.info(f"Config loaded from {path}")
-        except Exception as e:
-            logger.warning(f"Could not load {path}: {e}. Using defaults.")
-    return defaults
-
-CONFIG               = load_config()
-KB_DIR               = CONFIG["kb"]["directory"]
-INDEX_DIR            = CONFIG["kb"]["index_directory"]
-EMBEDDING_MODEL_NAME = CONFIG["models"]["embedding"]
-QA_MODEL_NAME        = CONFIG["models"]["qa"]
-CHUNK_SIZE           = CONFIG["chunking"]["chunk_size"]
-CHUNK_OVERLAP        = CONFIG["chunking"]["overlap"]
-TOP_K                = CONFIG["retrieval"]["top_k"]
-MIN_SIMILARITY       = CONFIG["retrieval"]["min_similarity"]
-MAX_SENTENCES        = CONFIG["retrieval"]["max_sentences"]
-MAX_QUERY_CHARS      = 800
-ALLOWED_EXTENSIONS   = {".txt", ".md", ".pdf", ".docx", ".doc"}
-MAX_FILE_SIZE_MB     = 10
-_IDX_PATH            = os.path.join(INDEX_DIR, "kb.index")
-_META_PATH           = os.path.join(INDEX_DIR, "kb_meta.npy")
-
-# ─── File loading ──────────────────────────────────────────────────────────────
-def load_file_text(path: str) -> str:
-    ext = os.path.splitext(path)[1].lower()
-    if ext == ".pdf":
-        from PyPDF2 import PdfReader
-        reader = PdfReader(path)
-        return "\n".join(p.extract_text() or "" for p in reader.pages)
-    elif ext in (".docx", ".doc"):
-        import docx as python_docx
-        doc = python_docx.Document(path)
-        return "\n".join(p.text for p in doc.paragraphs if p.text.strip())
-    else:
-        with open(path, "r", encoding="utf-8", errors="replace") as f:
-            return f.read()
-
-def load_kb_documents(kb_dir: str) -> List[Tuple[str, str]]:
-    docs = []
-    os.makedirs(kb_dir, exist_ok=True)
-    for pat in ["*.txt", "*.md", "*.pdf", "*.docx", "*.doc"]:
-        for path in sorted(glob.glob(os.path.join(kb_dir, pat))):
-            try:
-                text = load_file_text(path)
-                if text.strip():
-                    docs.append((os.path.basename(path), text))
-            except Exception as e:
-                logger.warning(f"Skipped {path}: {e}")
-    return docs
-
-# ─── Chunking ──────────────────────────────────────────────────────────────────
-def chunk_text(text: str, chunk_size: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> List[str]:
-    if not text:
-        return []
-    overlap = min(overlap, chunk_size - 1)
-    sentences = re.compile(r'(?<=[.!?])\s+').split(text.strip())
-    chunks, current, current_len = [], [], 0
-    for sent in sentences:
-        sent = sent.strip()
-        if not sent:
-            continue
-        slen = len(sent)
-        if slen > chunk_size:
-            if current:
-                chunks.append(" ".join(current))
-                current, current_len = [], 0
-            start = 0
-            while start < slen:
-                chunks.append(sent[start:min(start + chunk_size, slen)])
-                start += chunk_size - overlap
-            continue
-        if current_len + slen + 1 > chunk_size:
-            if current:
-                chunks.append(" ".join(current))
-            carry, carry_len = [], 0
-            for s in reversed(current):
-                if carry_len + len(s) + 1 > overlap:
-                    break
-                carry.insert(0, s)
-                carry_len += len(s) + 1
-            current, current_len = carry, carry_len
-        current.append(sent)
-        current_len += slen + 1
-    if current:
-        chunks.append(" ".join(current))
-    return [c for c in chunks if len(c.strip()) > 20]
-
-def clean_context(text: str) -> str:
-    lines, cleaned, seen = text.splitlines(), [], set()
-    for line in lines:
-        l = re.sub(r'^#+\s*', '', line.strip())
-        l = re.sub(r'^\d+[\.\)]\s*', '', l)
-        l = re.sub(r'^[-*]\s*', '', l)
-        if len(l) < 5 or l in seen:
-            continue
-        words = l.split()
-        if words and len(words) <= 10 and sum(1 for w in words if w[:1].isupper()) >= len(words) - 1:
-            continue
-        seen.add(l)
-        cleaned.append(l)
-    return "\n".join(cleaned)
-
-# ─── KB Index (FAISS) ──────────────────────────────────────────────────────────
-class KBIndex:
-    def __init__(self):
-        self.embedder: Optional[SentenceTransformer] = None
-        self.chunks:   List[str] = []
-        self.sources:  List[str] = []
-        self.index:    Optional[faiss.Index] = None
-        self._lock = threading.Lock()
-        logger.info("Loading embedding model…")
-        self.embedder = SentenceTransformer(EMBEDDING_MODEL_NAME)
-        logger.info("Embedding model loaded.")
-        os.makedirs(INDEX_DIR, exist_ok=True)
-        self._load_or_build()
-
-    def _load_or_build(self):
-        if os.path.exists(_IDX_PATH) and os.path.exists(_META_PATH):
-            try:
-                self.index = faiss.read_index(_IDX_PATH)
-                meta = np.load(_META_PATH, allow_pickle=True).item()
-                self.chunks  = list(meta["chunks"])
-                self.sources = list(meta["sources"])
-                logger.info(f"✅ FAISS index loaded — {len(self.chunks)} chunks.")
-                return
-            except Exception as e:
-                logger.warning(f"Could not load index: {e}. Rebuilding…")
-        self._build()
-
-    def _build(self):
-        docs = load_kb_documents(KB_DIR)
-        if not docs:
-            demo = ("demo_content.txt",
-                    "Welcome to the Self-Service KB Assistant. "
-                    "Upload your own documents via the Knowledge Base panel. "
-                    "The assistant will index them and answer questions from their content.")
-            docs = [demo]
-        all_chunks, all_sources = [], []
-        for source, text in docs:
-            for chunk in chunk_text(text):
-                all_chunks.append(chunk)
-                all_sources.append(source)
-        if not all_chunks:
-            self.index = None; self.chunks = []; self.sources = []
-            return
-        logger.info(f"Encoding {len(all_chunks)} chunks…")
-        embeddings = self.embedder.encode(
-            all_chunks, show_progress_bar=False,
-            convert_to_numpy=True, batch_size=64
-        ).astype("float32")
-        faiss.normalize_L2(embeddings)
-        index = faiss.IndexFlatIP(embeddings.shape[1])
-        index.add(embeddings)
-        self.index = index; self.chunks = all_chunks; self.sources = all_sources
-        try:
-            faiss.write_index(index, _IDX_PATH)
-            np.save(_META_PATH, {"chunks": np.array(all_chunks, dtype=object),
-                                 "sources": np.array(all_sources, dtype=object)})
-            logger.info("FAISS index saved.")
-        except Exception as e:
-            logger.warning(f"Could not save index: {e}")
-
-    def rebuild(self) -> str:
-        with self._lock:
-            for p in (_IDX_PATH, _META_PATH):
-                if os.path.exists(p): os.remove(p)
-            self._build()
-        if not self.chunks:
-            return "Index rebuilt (empty — no documents found)."
-        return f"Index rebuilt — {len(self.chunks)} chunks from {len(set(self.sources))} files."
-
-    def retrieve(self, query: str) -> List[Tuple[str, str, float]]:
-        if not query.strip() or self.index is None or not self.chunks:
-            return []
-        q = self.embedder.encode([query], convert_to_numpy=True).astype("float32")
-        faiss.normalize_L2(q)
-        scores, idxs = self.index.search(q, min(TOP_K, len(self.chunks)))
-        return [
-            (self.chunks[i], self.sources[i], float(s))
-            for s, i in zip(scores[0], idxs[0])
-            if 0 <= i < len(self.chunks) and float(s) >= MIN_SIMILARITY
-        ]
-
-    def best_sentences(self, query: str, results: List[Tuple[str, str, float]]) -> str:
-        sents = []
-        for chunk, _, _ in results:
-            for s in re.split(r'(?<=[.!?])\s+|\n+', clean_context(chunk)):
-                if len(s.strip()) >= 25:
-                    sents.append(s.strip())
-        if not sents:
-            return ""
-        q_emb = self.embedder.encode([query], convert_to_numpy=True).astype("float32")
-        s_emb = self.embedder.encode(sents,  convert_to_numpy=True).astype("float32")
-        faiss.normalize_L2(q_emb); faiss.normalize_L2(s_emb)
-        sims = np.dot(s_emb, q_emb.T).reshape(-1)
-        q_words = {w.lower() for w in re.findall(r"\w+", query) if len(w) > 3}
-        lex = np.array([len(q_words & {w.lower() for w in re.findall(r"\w+", s) if len(w) > 3})
-                        for s in sents], dtype=float)
-        top = np.argsort(-(1.5 * sims + 0.5 * lex))
-        chosen, seen = [], set()
-        for i in top:
-            s = sents[i].strip()
-            if s and s not in seen:
-                chosen.append(s); seen.add(s)
-            if len(chosen) >= MAX_SENTENCES:
-                break
-        return "\n".join(chosen)
-
-# ─── QA Model ─────────────────────────────────────────────────────────────────
-qa_tokenizer = None
-qa_model     = None
-qa_available = False
-_qa_lock     = threading.Lock()
-
-try:
-    from transformers import AutoTokenizer, AutoModelForSeq2SeqLM
-    logger.info(f"Loading QA model: {QA_MODEL_NAME}…")
-    qa_tokenizer = AutoTokenizer.from_pretrained(QA_MODEL_NAME)
-    qa_model     = AutoModelForSeq2SeqLM.from_pretrained(QA_MODEL_NAME)
-    qa_model.eval()
-    qa_available = True
-    logger.info(f"✅ QA model loaded ({QA_MODEL_NAME})")
-except Exception as e:
-    logger.warning(f"⚠️ Could not load QA model: {e}. Using extractive fallback.")
-
-def generate_answer(question: str, context: str) -> Optional[str]:
-    if not qa_available:
-        return None
-    prompt = (
-        "You are a warm, helpful assistant. Answer using ONLY the context below. "
-        "Open with a friendly sentence. Introduce any steps with a sentence first. "
-        "Keep tone conversational. If context is insufficient, say so honestly.\n\n"
-        f"Context:\n{context}\n\nQuestion: {question}\n\nAnswer:"
-    )
-    try:
-        with _qa_lock:
-            inputs = qa_tokenizer(prompt, return_tensors="pt", truncation=True, max_length=768)
-            outputs = qa_model.generate(**inputs, max_new_tokens=220, temperature=1.0, do_sample=False)
-            answer = qa_tokenizer.decode(outputs[0], skip_special_tokens=True).strip()
-        return answer or None
-    except Exception as e:
-        logger.error(f"QA error: {e}")
-        return None
-
-# ─── Init index ───────────────────────────────────────────────────────────────
-logger.info("Initialising KB index…")
-kb_index = KBIndex()
-
-# ─── RAG pipeline ─────────────────────────────────────────────────────────────
-def build_answer(query: str, history: list) -> dict:
-    """Returns {"answer": str, "sources": [str]}"""
-    results = kb_index.retrieve(query)
-    if not results:
-        return {"answer": "I couldn't find anything relevant for your question. Try rephrasing or using different keywords.", "sources": []}
-
-    context = kb_index.best_sentences(query, results)
-    if not context:
-        return {"answer": "I found some content but couldn't extract relevant sentences. Please try rephrasing.", "sources": []}
-
-    # Collect unique sources above threshold
-    seen_sources: List[str] = []
-    for _, src, score in results:
-        if src not in seen_sources and score >= MIN_SIMILARITY:
-            seen_sources.append(src)
-        if len(seen_sources) == 3:
-            break
-
-    answer = generate_answer(query, context)
-    if not answer:
-        answer = "Here's the most relevant information I found:\n\n" + "\n".join(
-            f"• {s}" for s in context.split("\n") if s.strip()
-        )
-    return {"answer": answer, "sources": seen_sources}
-
-# ─── FastAPI app ───────────────────────────────────────────────────────────────
-app = FastAPI(title="Self-Service KB Assistant")
-app.mount("/site", StaticFiles(directory="site"), name="site")
-
-# ── Pydantic models ────────────────────────────────────────────────────────────
-class ChatRequest(BaseModel):
-    message: str
-    history: list = []
-
-class DeleteRequest(BaseModel):
-    filename: str
-
-class RenameRequest(BaseModel):
-    old_name: str
-    new_name: str
-
-# ── Routes ─────────────────────────────────────────────────────────────────────
-@app.get("/")
-async def root():
-    return FileResponse("static/index.html")
-
-@app.get("/api/config")
-async def get_config():
-    quick = [q["query"] for q in CONFIG.get("quick_actions", []) if "query" in q] or [
-        "What makes a good knowledge base article?",
-        "How could a KB assistant help support agents?",
-        "Why is self-service important for customer support?",
-    ]
-    return {
-        "name":         CONFIG["client"]["name"],
-        "welcome":      CONFIG["messages"].get("welcome", "Ask me anything!"),
-        "quick_actions": quick,
-    }
-
-@app.post("/api/chat")
-async def chat(req: ChatRequest):
-    if not req.message.strip():
-        return {"answer": "Please ask a question.", "sources": []}
-    try:
-        return build_answer(req.message.strip()[:MAX_QUERY_CHARS], req.history)
-    except Exception as e:
-        logger.error(f"Chat error: {e}")
-        raise HTTPException(status_code=500, detail="Something went wrong. Please try again.")
-
-@app.get("/api/files")
-async def list_files():
-    files = []
-    for pat in ["*.txt", "*.md", "*.pdf", "*.docx", "*.doc"]:
-        for p in sorted(glob.glob(os.path.join(KB_DIR, pat))):
-            name = os.path.basename(p)
-            size = os.path.getsize(p)
-            files.append({"name": name, "size": size})
-    return {"files": files, "count": len(files)}
-
-@app.post("/api/files/upload")
-async def upload_files(files: List[UploadFile] = File(...)):
-    os.makedirs(KB_DIR, exist_ok=True)
-    saved, skipped = [], []
-    for file in files:
-        name = os.path.basename(file.filename or "")
-        ext  = os.path.splitext(name)[1].lower()
-        if ext not in ALLOWED_EXTENSIONS:
-            skipped.append(f"{name} (unsupported type)")
-            continue
-        content = await file.read()
-        if len(content) / (1024 * 1024) > MAX_FILE_SIZE_MB:
-            skipped.append(f"{name} (exceeds {MAX_FILE_SIZE_MB} MB)")
-            continue
-        try:
-            with open(os.path.join(KB_DIR, name), "wb") as f:
-                f.write(content)
-            saved.append(name)
-        except Exception as e:
-            skipped.append(f"{name} (error: {e})")
-    msg = ""
-    if saved:
-        msg = kb_index.rebuild()
-    return {"saved": saved, "skipped": skipped, "message": msg}
-
-@app.delete("/api/files/{filename}")
-async def delete_file(filename: str):
-    filename = os.path.basename(filename)
-    if os.path.splitext(filename)[1].lower() not in ALLOWED_EXTENSIONS:
-        raise HTTPException(status_code=400, detail="Invalid file type")
-    target = os.path.join(KB_DIR, filename)
-    if not os.path.exists(target):
-        raise HTTPException(status_code=404, detail="File not found")
-    os.remove(target)
-    msg = kb_index.rebuild()
-    return {"deleted": filename, "message": msg}
-
-@app.put("/api/files/rename")
-async def rename_file(req: RenameRequest):
-    old = os.path.basename(req.old_name.strip())
-    new = os.path.basename(req.new_name.strip())
-    if not old or not new:
-        raise HTTPException(status_code=400, detail="Both names required")
-    if os.path.splitext(new)[1].lower() not in ALLOWED_EXTENSIONS:
-        raise HTTPException(status_code=400, detail="New name must have a supported extension")
-    src = os.path.join(KB_DIR, old)
-    if not os.path.exists(src):
-        raise HTTPException(status_code=404, detail=f"File not found: {old}")
-    os.rename(src, os.path.join(KB_DIR, new))
-    msg = kb_index.rebuild()
-    return {"renamed_to": new, "message": msg}
-
-# ─── Entry point ──────────────────────────────────────────────────────────────
-if __name__ == "__main__":
-    import uvicorn
-    port = int(os.environ.get("PORT", 7860))
-    uvicorn.run("app:app", host="0.0.0.0", port=port, reload=False)

app/main.py

@@ -0,0 +1,33 @@
+from pathlib import Path
+from fastapi import FastAPI
+from fastapi.staticfiles import StaticFiles
+
+from app.core.lifecycle import lifespan
+from app.routers.chat import router as chat_router
+from app.routers.health import router as health_router
+from app.routers.evaluation import router as evaluation_router
+
+app = FastAPI(
+    title="Self-Service KB Assistant",
+    version="1.0.0",
+    lifespan=lifespan,
+)
+
+# Routers
+app.include_router(chat_router, prefix="/api/chat", tags=["chat"])
+app.include_router(health_router, tags=["health"])
+app.include_router(evaluation_router, prefix="/api/evaluation", tags=["evaluation"])
+
+# Safe static mounting
+SITE_DIR = Path("site")
+
+if SITE_DIR.exists():
+    app.mount("/site", StaticFiles(directory=SITE_DIR), name="site")
+
+
+@app.get("/")
+async def root():
+    return {
+        "status": "running",
+        "service": "Self-Service KB Assistant"
+    }