user uploads supported

app.py

@@ -3,146 +3,141 @@ import logging
 import gradio as gr
 import requests
 
-# Suppress HuggingFace tokenizer parallelism warning
 os.environ["TOKENIZERS_PARALLELISM"] = "false"
 
-# Setup logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
-# FastAPI endpoint configuration
 FASTAPI_URL = os.getenv("FASTAPI_URL", "http://127.0.0.1:8000")
 CHAT_ENDPOINT = f"{FASTAPI_URL}/chat"
+UPLOAD_ENDPOINT = f"{FASTAPI_URL}/upload"
+RESET_ENDPOINT = f"{FASTAPI_URL}/reset"
 
+SUPPORTED_TYPES = [".pdf", ".docx", ".txt", ".md", ".csv"]
+
+
+def upload_files(files) -> str:
+    if not files:
+        return "No files selected."
+    try:
+        multipart = []
+        for f in files:
+            path = f if isinstance(f, str) else f.name
+            filename = os.path.basename(path)
+            with open(path, "rb") as fp:
+                multipart.append(("files", (filename, fp.read(), "application/octet-stream")))
+
+        resp = requests.post(UPLOAD_ENDPOINT, files=multipart, timeout=120)
+        resp.raise_for_status()
+        return resp.json().get("status", "Files processed.")
+    except requests.exceptions.RequestException as e:
+        return f"Upload failed: {e}"
+
+
+def reset_documents() -> str:
+    try:
+        resp = requests.post(RESET_ENDPOINT, timeout=10)
+        resp.raise_for_status()
+        return resp.json().get("status", "Documents cleared.")
+    except requests.exceptions.RequestException as e:
+        return f"Reset failed: {e}"
+
+
+def determine_source(text: str) -> str:
+    lower = text.lower()
+    if any(k in lower for k in ["search", "web", "duckduckgo", "internet", "online"]):
+        return "Web Search"
+    if any(k in lower for k in ["uploaded", "file:", "page", "document", "policy", "pdf"]):
+        return "Uploaded Documents (RAG)"
+    return "RAG + Web Search"
 
-def determine_source(response_text) -> str:
-    """
-    Attempt to determine if the response came from RAG or Web Search.
-    Looks for keywords or source indicators in the response.
-    """
-    # Convert list to string if needed
-    if isinstance(response_text, list):
-        response_text = " ".join(str(item) for item in response_text)
-    
-    response_lower = str(response_text).lower()
-    
-    # Check for web search indicators
-    if any(keyword in response_lower for keyword in ["search", "web", "duckduckgo", "internet", "online"]):
-        return "Source: Web Search"
-    
-    # Check for RAG/document indicators
-    if "source:" in response_lower and "page" in response_lower:
-        return "Source: Internal Documents (RAG)"
-    
-    if any(keyword in response_lower for keyword in ["document", "policy", "pdf", "internal", "knowledge base"]):
-        return "Source: Internal Documents (RAG)"
-    
-    # Default fallback - cannot determine
-    return "Source: Mixed (RAG + Web Search)"
 
 def process_query(message: str, chat_history: list) -> tuple[list, str]:
-    """
-    Process user query by calling the FastAPI /chat endpoint.
-    
-    Args:
-        message: User query
-        chat_history: Current chat history in Gradio format
-        
-    Returns:
-        Tuple of (updated_chat_history, status_message)
-    """
+    if not message.strip():
+        return chat_history, "Please enter a question."
     try:
-        logger.info(f"Sending query to FastAPI: {message}")
-        response = requests.post(CHAT_ENDPOINT, json={"query": message}, timeout=60)
-        response.raise_for_status()
-        payload = response.json()
-        response_text = payload.get("response", "")
-        
-        if isinstance(response_text, list):
-            response_text = "\n".join(str(item) for item in response_text)
-        else:
-            response_text = str(response_text)
-        
+        resp = requests.post(CHAT_ENDPOINT, json={"query": message}, timeout=120)
+        resp.raise_for_status()
+        response_text = resp.json().get("response", "")
+
         source = determine_source(response_text)
-        full_response = f"{response_text}\n\n--- {source} ---"
-        
+        full_response = f"{response_text}\n\n--- Source: {source} ---"
+
         chat_history.append({"role": "user", "content": message})
         chat_history.append({"role": "assistant", "content": full_response})
-        
-        logger.info("Response received from FastAPI")
-        return chat_history, f"Query processed. {source}"
+        return chat_history, f"Done — answered via {source}"
     except requests.exceptions.RequestException as e:
-        error_msg = f"FastAPI request failed: {str(e)}"
-        logger.error(error_msg)
-        chat_history.append({"role": "user", "content": message})
-        chat_history.append({"role": "assistant", "content": f"ERROR: {error_msg}"})
-        return chat_history, error_msg
-    except Exception as e:
-        error_msg = f"Error processing query: {str(e)}"
-        logger.error(error_msg)
+        error = f"Request failed: {e}"
         chat_history.append({"role": "user", "content": message})
-        chat_history.append({"role": "assistant", "content": f"ERROR: {error_msg}"})
-        return chat_history, error_msg
+        chat_history.append({"role": "assistant", "content": f"ERROR: {error}"})
+        return chat_history, error
+
 
 def clear_chat() -> tuple[list, str]:
-    """Clear chat history."""
     return [], ""
 
-# Build Gradio Interface
+
+# --- UI ---
 with gr.Blocks(title="Agentic RAG Knowledge Search") as demo:
-    gr.Markdown("# 🔍 Agentic RAG Knowledge Search")
-    gr.Markdown("Ask questions answered from internal documents (RAG) or the live web — the agent decides.")
+    gr.Markdown("# Agentic RAG Knowledge Search")
+    gr.Markdown(
+        "Upload your documents, then ask questions. "
+        "The agent searches your files first, then the web if needed."
+    )
+
+    # File upload panel
+    with gr.Group():
+        gr.Markdown("### Upload Documents")
+        gr.Markdown(f"Supported: {', '.join(SUPPORTED_TYPES)}")
+        with gr.Row():
+            file_input = gr.File(
+                label="Select Files",
+                file_count="multiple",
+                file_types=SUPPORTED_TYPES,
+            )
+        with gr.Row():
+            upload_btn = gr.Button("Process Files", variant="primary")
+            reset_btn = gr.Button("Clear Uploaded Documents", variant="secondary")
+        upload_status = gr.Textbox(label="Upload Status", interactive=False, lines=2)
+
+    gr.Markdown("---")
+
+    # Chat panel
     with gr.Group():
+        gr.Markdown("### Ask a Question")
         chatbot = gr.Chatbot(
-            label="Conversation History",
-            show_label=True,
-            height=400
+            label="Conversation",
+            height=420,
         )
-    
-    with gr.Row():
-        user_input = gr.Textbox(
-            placeholder="Ask a question about policy documents or anything on the web...",
-            label="Your Question",
-            lines=2
-        )
-    
-    with gr.Row():
-        submit_btn = gr.Button("Submit", variant="primary")
-        clear_btn = gr.Button("Clear", variant="secondary")
-    
-    status_output = gr.Textbox(
-        label="Status",
-        interactive=False,
-        lines=1
-    )
-    
-    # Wire up chat interactions
+        with gr.Row():
+            user_input = gr.Textbox(
+                placeholder="Ask anything about your documents or the web...",
+                label="Your Question",
+                lines=2,
+                scale=4,
+            )
+            submit_btn = gr.Button("Submit", variant="primary", scale=1)
+        with gr.Row():
+            clear_btn = gr.Button("Clear Chat", variant="secondary")
+        status_output = gr.Textbox(label="Status", interactive=False, lines=1)
+
+    # Wiring
+    upload_btn.click(fn=upload_files, inputs=[file_input], outputs=[upload_status])
+    reset_btn.click(fn=reset_documents, inputs=[], outputs=[upload_status])
+
     submit_btn.click(
         fn=process_query,
         inputs=[user_input, chatbot],
-        outputs=[chatbot, status_output]
-    ).then(
-        fn=lambda: "",
-        inputs=[],
-        outputs=[user_input]
-    )
-    
-    clear_btn.click(
-        fn=clear_chat,
-        inputs=[],
-        outputs=[chatbot, status_output]
-    )
-    
-    # Allow Enter key to submit
+        outputs=[chatbot, status_output],
+    ).then(fn=lambda: "", inputs=[], outputs=[user_input])
+
     user_input.submit(
         fn=process_query,
         inputs=[user_input, chatbot],
-        outputs=[chatbot, status_output]
-    ).then(
-        fn=lambda: "",
-        inputs=[],
-        outputs=[user_input]
-    )
+        outputs=[chatbot, status_output],
+    ).then(fn=lambda: "", inputs=[], outputs=[user_input])
+
+    clear_btn.click(fn=clear_chat, inputs=[], outputs=[chatbot, status_output])
 
 if __name__ == "__main__":
     demo.launch(server_name="0.0.0.0", server_port=7860, share=False, theme=gr.themes.Soft())

requirements.txt

@@ -15,6 +15,7 @@ langchain-huggingface>=0.1.2
 sentence-transformers>=3.2.0
 faiss-cpu>=1.7.4
 pypdf>=5.1.0
+docx2txt>=0.8
 duckduckgo-search==5.3.1
 pydantic>=2.9.0
 requests>=2.32.0

src/agent.py

@@ -5,52 +5,51 @@ from langchain_community.tools import DuckDuckGoSearchRun
 from langgraph.prebuilt import create_react_agent
 from dotenv import load_dotenv
 from src.rag_engine import KnowledgeBase
+from src.file_processor import FileProcessor
 
 load_dotenv()
 
-# --- Configuration ---
-PDF_PATH = os.path.join("data", "policy.pdf")
+# --- Shared state ---
+# file_processor is imported and mutated by main.py's /upload endpoint
+file_processor = FileProcessor()
 
-kb = KnowledgeBase(pdf_path=PDF_PATH)
+# Optional fallback KB (original policy.pdf); silently skipped if missing
+_PDF_PATH = os.path.join("data", "policy.pdf")
+_fallback_kb = KnowledgeBase(pdf_path=_PDF_PATH)
 try:
-    kb.load_and_index()
+    _fallback_kb.load_and_index()
 except Exception as e:
-    print(f"PDF Load skipped: {e}")
+    print(f"Fallback KB skipped: {e}")
 
-# --- Define Tools ---
+# --- Tools ---
 
 @tool
-def lookup_internal_policy(query: str) -> str:
-    """Useful for answering questions about specific internal policies, documents, laws, or the PDF file."""
-    return kb.retrieve(query)
+def lookup_documents(query: str) -> str:
+    """Search the user-uploaded documents for relevant information.
+    Use this for questions about content in any uploaded files."""
+    if file_processor.has_documents():
+        result = file_processor.retrieve(query)
+        if result:
+            return result
+    # Fall back to original KB when no uploads exist
+    return _fallback_kb.retrieve(query)
 
-# Initialize the tool ONCE (Global scope)
 search_tool = DuckDuckGoSearchRun()
 
 @tool
 def search_web(query: str) -> str:
-    """Useful for finding current events, news, or general knowledge not in the internal docs."""
-    # Use the pre-initialized tool
+    """Search the web for current events, news, or general knowledge not in uploaded documents."""
     try:
         return search_tool.run(query)
     except Exception as e:
         return f"Search failed: {e}"
 
-# --- Initialize Agent ---
+# --- Agent factory ---
 
 def get_agent_executor():
     if not os.getenv("GOOGLE_API_KEY"):
         raise ValueError("GOOGLE_API_KEY not found in .env file")
 
     print("Initializing Gemini Agent (Model: gemini-2.5-flash-lite)...")
-    llm = ChatGoogleGenerativeAI(
-        model="gemini-2.5-flash-lite",
-        temperature=0
-    )
-    
-    tools = [lookup_internal_policy, search_web]
-    
-    # Create the Agent (LangGraph)
-    agent = create_react_agent(llm, tools)
-    
-    return agent
+    llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash-lite", temperature=0)
+    return create_react_agent(llm, [lookup_documents, search_web])

src/file_processor.py

@@ -1,136 +1,111 @@
-"""
-File Processor for handling user-uploaded documents.
-Processes PDFs, creates FAISS indices for semantic search.
-"""
-
 import logging
-from langchain_community.document_loaders import PyPDFLoader
+import os
+from langchain_community.document_loaders import (
+    PyPDFLoader,
+    TextLoader,
+    CSVLoader,
+    Docx2txtLoader,
+)
 from langchain_text_splitters import RecursiveCharacterTextSplitter
 from langchain_community.vectorstores import FAISS
 from langchain_huggingface import HuggingFaceEmbeddings
 
 logger = logging.getLogger(__name__)
 
+SUPPORTED_EXTENSIONS = {".pdf", ".txt", ".md", ".csv", ".docx"}
 
-class FileProcessor:
-    """
-    Handles processing of user-uploaded PDF files and creates searchable indices.
-    """
 
-    def __init__(self, embedding_model: str = "all-MiniLM-L6-v2"):
-        """
-        Initialize FileProcessor with embeddings.
+def _loader_for(file_path: str):
+    ext = os.path.splitext(file_path)[1].lower()
+    if ext == ".pdf":
+        return PyPDFLoader(file_path)
+    if ext in (".txt", ".md"):
+        return TextLoader(file_path, encoding="utf-8")
+    if ext == ".csv":
+        return CSVLoader(file_path)
+    if ext == ".docx":
+        return Docx2txtLoader(file_path)
+    return None
 
-        Args:
-            embedding_model: HuggingFace embedding model name
-        """
-        logger.info(f"Initializing FileProcessor with model: {embedding_model}")
+
+class FileProcessor:
+    def __init__(self, embedding_model: str = "all-MiniLM-L6-v2"):
         self.embeddings = HuggingFaceEmbeddings(model_name=embedding_model)
         self.vector_store = None
         self.status = "No files processed"
-
-    def process_files(self, files: list) -> str:
-        """
-        Process uploaded PDF files and create a FAISS index.
-
-        Args:
-            files: List of file objects from Gradio
-
-        Returns:
-            Status message
-        """
-        if not files:
-            self.status = "No files uploaded"
-            return "No files selected. Using internal docs + web search."
-
-        try:
-            logger.info(f"Processing {len(files)} uploaded file(s)...")
-            all_docs = []
-
-            # Load all PDF files
-            for file_obj in files:
-                file_path = file_obj if isinstance(file_obj, str) else file_obj.name
-
-                if file_path.lower().endswith(".pdf"):
-                    logger.info(f"Loading PDF: {file_path}")
-                    loader = PyPDFLoader(file_path)
-                    docs = loader.load()
-                    all_docs.extend(docs)
-                else:
-                    logger.warning(f"Skipping non-PDF file: {file_path}")
-
-            if not all_docs:
-                self.status = "No PDFs found in uploads"
-                return "No valid PDF files uploaded. Using internal docs + web search."
-
-            # Split documents into chunks
-            logger.info(f"Splitting {len(all_docs)} documents into chunks...")
-            text_splitter = RecursiveCharacterTextSplitter(
-                chunk_size=1000, chunk_overlap=200
+        self._splitter = RecursiveCharacterTextSplitter(
+            chunk_size=1000, chunk_overlap=200
+        )
+
+    def process_files(self, file_paths: list[str]) -> str:
+        """Index a list of file paths into the FAISS vector store."""
+        if not file_paths:
+            return "No files provided."
+
+        all_docs = []
+        skipped = []
+
+        for path in file_paths:
+            ext = os.path.splitext(path)[1].lower()
+            if ext not in SUPPORTED_EXTENSIONS:
+                skipped.append(os.path.basename(path))
+                continue
+            try:
+                loader = _loader_for(path)
+                docs = loader.load()
+                for d in docs:
+                    d.metadata["source_file"] = os.path.basename(path)
+                all_docs.extend(docs)
+                logger.info(f"Loaded {len(docs)} page(s) from {os.path.basename(path)}")
+            except Exception as e:
+                logger.error(f"Failed to load {path}: {e}")
+                skipped.append(os.path.basename(path))
+
+        if not all_docs:
+            self.status = "No supported files could be loaded"
+            return (
+                f"Could not load any files. "
+                f"Supported types: {', '.join(sorted(SUPPORTED_EXTENSIONS))}. "
+                + (f"Skipped: {', '.join(skipped)}" if skipped else "")
             )
-            chunks = text_splitter.split_documents(all_docs)
 
-            # Create FAISS index
-            logger.info(f"Creating FAISS index from {len(chunks)} chunks...")
-            self.vector_store = FAISS.from_documents(chunks, self.embeddings)
+        chunks = self._splitter.split_documents(all_docs)
 
-            self.status = (
-                f"Indexed {len(chunks)} chunks from {len(all_docs)} pages"
-            )
-            logger.info(self.status)
-            return f"✓ SUCCESS: {self.status} - Your uploaded documents will be searched first!"
+        if self.vector_store is None:
+            self.vector_store = FAISS.from_documents(chunks, self.embeddings)
+        else:
+            # Merge into existing index so previous uploads are retained
+            new_store = FAISS.from_documents(chunks, self.embeddings)
+            self.vector_store.merge_from(new_store)
 
-        except Exception as e:
-            error_msg = f"Error processing files: {str(e)}"
-            logger.error(error_msg)
-            self.status = "Error loading files"
-            return f"ERROR: {error_msg} - Falling back to internal docs + web search."
+        file_count = len(file_paths) - len(skipped)
+        self.status = f"Indexed {len(chunks)} chunks from {file_count} file(s)"
+        note = f" (skipped: {', '.join(skipped)})" if skipped else ""
+        logger.info(self.status)
+        return f"Done: {self.status}{note}"
 
     def retrieve(self, query: str, k: int = 4) -> str:
-        """
-        Retrieve relevant chunks from uploaded files.
-
-        Args:
-            query: Search query
-            k: Number of results to return
-
-        Returns:
-            Retrieved content or empty string if no index
-        """
         if not self.vector_store:
             return ""
-
         try:
             docs = self.vector_store.similarity_search(query, k=k)
             if not docs:
                 return ""
             return "\n\n".join(
-                [f"[Uploaded Document] {d.page_content}" for d in docs]
+                f"[File: {d.metadata.get('source_file', '?')}] {d.page_content}"
+                for d in docs
             )
         except Exception as e:
-            logger.error(f"Error retrieving from uploaded files: {e}")
+            logger.error(f"Retrieval error: {e}")
             return ""
 
     def has_documents(self) -> bool:
-        """
-        Check if vector store has documents.
-
-        Returns:
-            True if documents are loaded, False otherwise
-        """
         return self.vector_store is not None
 
     def get_status(self) -> str:
-        """
-        Get current processing status.
-
-        Returns:
-            Status message
-        """
         return self.status
 
     def reset(self) -> None:
-        """Reset the file processor and clear loaded documents."""
         self.vector_store = None
         self.status = "No files processed"
-        logger.info("File processor reset")
+        logger.info("FileProcessor reset")

src/main.py

@@ -1,62 +1,87 @@
-from fastapi import FastAPI, HTTPException
-from pydantic import BaseModel
-from src.agent import get_agent_executor
+import os
+import shutil
+import tempfile
 import logging
+from fastapi import FastAPI, HTTPException, UploadFile, File
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+from typing import List
+from src.agent import get_agent_executor, file_processor
 
-# Setup logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 app = FastAPI(
     title="Agentic RAG Service",
-    description="An AI Microservice that routes between Internal Docs and Web Search using LangGraph.",
-    version="2.0"
+    description="AI microservice that routes between uploaded documents and web search.",
+    version="3.0",
 )
 
-# Initialize Agent
 try:
     agent_executor = get_agent_executor()
 except Exception as e:
     logger.error(f"Failed to initialize agent: {e}")
     agent_executor = None
 
-# --- API Models ---
+
 class QueryRequest(BaseModel):
     query: str
 
+
 class QueryResponse(BaseModel):
     response: str
 
-# --- Routes ---
 
 @app.get("/")
 async def root():
     return {
-        "status": "active", 
-        "service": "Agentic Knowledge Search (LangGraph)", 
-        "docs_url": "/docs"
+        "status": "active",
+        "service": "Agentic Knowledge Search",
+        "docs_url": "/docs",
+        "uploaded_docs": file_processor.get_status(),
     }
 
+
+@app.post("/upload")
+async def upload_files(files: List[UploadFile] = File(...)):
+    """Accept uploaded files, index them for RAG, return a status message."""
+    tmp_dir = tempfile.mkdtemp()
+    try:
+        saved_paths = []
+        for upload in files:
+            dest = os.path.join(tmp_dir, upload.filename)
+            with open(dest, "wb") as f:
+                shutil.copyfileobj(upload.file, f)
+            saved_paths.append(dest)
+            logger.info(f"Saved upload: {upload.filename}")
+
+        status = file_processor.process_files(saved_paths)
+        logger.info(f"Processing result: {status}")
+        return JSONResponse({"status": status})
+    except Exception as e:
+        logger.error(f"Upload error: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+    finally:
+        shutil.rmtree(tmp_dir, ignore_errors=True)
+
+
+@app.post("/reset")
+async def reset_documents():
+    """Clear all uploaded documents from the index."""
+    file_processor.reset()
+    return {"status": "Uploaded documents cleared."}
+
+
 @app.post("/chat", response_model=QueryResponse)
 async def chat(request: QueryRequest):
     if not agent_executor:
-        raise HTTPException(status_code=500, detail="Agent not initialized (Check API Keys)")
-    
+        raise HTTPException(status_code=500, detail="Agent not initialized (check API key)")
+
     try:
-        logger.info(f"Received query: {request.query}")
-        
-        # LangGraph Input Format
-        # pass a dictionary with "messages"
-        inputs = {"messages": [("user", request.query)]}
-        
-        # Invoke the agent
-        result = agent_executor.invoke(inputs)
-        
-        # The result contains the entire conversation state. 
-        # The last message is the AI's answer.
+        logger.info(f"Query: {request.query}")
+        result = agent_executor.invoke({"messages": [("user", request.query)]})
         last_message = result["messages"][-1]
 
-        # Gemini may return content as a list of typed blocks; extract plain text
         content = last_message.content
         if isinstance(content, list):
             content = " ".join(
@@ -66,11 +91,11 @@ async def chat(request: QueryRequest):
             )
 
         return QueryResponse(response=str(content))
-        
     except Exception as e:
-        logger.error(f"Error processing query: {e}")
+        logger.error(f"Chat error: {e}")
         raise HTTPException(status_code=500, detail=str(e))
 
+
 if __name__ == "__main__":
     import uvicorn
-    uvicorn.run(app, host="0.0.0.0", port=8000)
+    uvicorn.run(app, host="0.0.0.0", port=8000)