Knowledge Agent

""" web_app.py — FastAPI web UI + REST API for the Knowledge Agent ============================================================== Run with: uvicorn web_app:app --host 0.0.0.0 --port 8000 --reload Endpoints: GET / → Web UI (chat interface) POST /api/ask → JSON Q&A GET /api/stream → SSE streaming answer POST /api/index → Trigger re-indexing POST /api/upload → Upload and index a new document GET /api/docs-list → List indexed documents GET /api/stats → Knowledge base stats DELETE /api/doc/{filename} → Remove a document """ import os import shutil import asyncio from pathlib import Path from typing import Optional from fastapi import FastAPI, UploadFile, File, HTTPException, Query from fastapi.responses import HTMLResponse, StreamingResponse, JSONResponse from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from knowledge_api import ( query_knowledge, index_docs, index_single, get_knowledge_stats, list_indexed_docs, delete_doc, stream_knowledge_answer, ) DOCS_DIR = os.getenv("DOCS_DIR", "./documents") # ── App setup ───────────────────────────────────────────────────────────────── app = FastAPI( title="Knowledge Agent", description="Personal RAG system over your documents", version="1.0.0", ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # ── Request / Response models ───────────────────────────────────────────────── class AskRequest(BaseModel): question: str top_k: int = 5 class IndexRequest(BaseModel): docs_dir: Optional[str] = None force: bool = False # ── REST API endpoints ──────────────────────────────────────────────────────── @app.post("/api/ask") async def ask(req: AskRequest): """Ask a question and get a grounded answer from your documents.""" if not req.question.strip(): raise HTTPException(400, "Question cannot be empty.") result = query_knowledge(req.question, top_k=req.top_k) return result @app.get("/api/stream") async def stream( question: str = Query(..., description="Your question"), top_k: int = Query(5, description="Number of chunks to retrieve"), ): """ SSE endpoint — streams the answer token by token. Used by the web UI for a typing-cursor effect. """ if not question.strip(): raise HTTPException(400, "Question cannot be empty.") def event_generator(): for token in stream_knowledge_answer(question, top_k=top_k): # SSE format: "data: \n\n" yield f"data: {token}\n\n" yield "data: [DONE]\n\n" return StreamingResponse( event_generator(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", }, ) @app.post("/api/index") async def trigger_index(req: IndexRequest): """Trigger indexing of the documents directory.""" docs_dir = req.docs_dir or DOCS_DIR # Run in a thread pool so we don't block the event loop loop = asyncio.get_event_loop() await loop.run_in_executor( None, lambda: index_docs(docs_dir=docs_dir, force=req.force) ) stats = get_knowledge_stats() return {"message": "Indexing complete.", "stats": stats} @app.post("/api/upload") async def upload_document(file: UploadFile = File(...)): """ Upload a document file; save it to the documents dir and index it. Supports: .pdf, .docx, .txt, .md """ allowed = {".pdf", ".docx", ".doc", ".txt", ".md", ".markdown"} ext = Path(file.filename).suffix.lower() if ext not in allowed: raise HTTPException(400, f"Unsupported file type: {ext}. Allowed: {allowed}") os.makedirs(DOCS_DIR, exist_ok=True) dest = os.path.join(DOCS_DIR, file.filename) # Save file with open(dest, "wb") as f: shutil.copyfileobj(file.file, f) # Index it loop = asyncio.get_event_loop() await loop.run_in_executor(None, lambda: index_single(dest)) return { "message": f"'{file.filename}' uploaded and indexed.", "path": dest, } @app.get("/api/docs-list") async def docs_list(): """List all documents currently in the knowledge base.""" return {"documents": list_indexed_docs()} @app.get("/api/stats") async def stats(): """Return knowledge base statistics.""" return get_knowledge_stats() @app.delete("/api/doc/{filename}") async def remove_doc(filename: str): """Remove a document and all its chunks from the index.""" delete_doc(filename) return {"message": f"Deleted '{filename}' from the knowledge base."} # ── Web UI ──────────────────────────────────────────────────────────────────── @app.get("/", response_class=HTMLResponse) async def ui(): """Serve the single-page chat interface.""" return HTMLResponse(content=WEB_UI_HTML) # ── HTML for the web UI (self-contained, no build step needed) ──────────────── WEB_UI_HTML = """ Knowledge Agent

knowledge_agent

🧠

Ask anything about your documents Add files via the sidebar, then start asking questions.

Ready

"""