Create app.py

app.py

@@ -0,0 +1,265 @@
+from flask import Flask, render_template, request, jsonify
+import faiss
+import numpy as np
+import json
+from sentence_transformers import SentenceTransformer
+from langchain.prompts import PromptTemplate
+from langchain_groq import ChatGroq
+import re
+import faiss
+import numpy as np
+import json
+from sentence_transformers import SentenceTransformer
+from dotenv import load_dotenv
+import fitz  # PyMuPDF for text extraction
+from pdf2image import convert_from_path
+import json
+import os
+load_dotenv()
+
+def extract_text_images(pdf_path, output_dir="static/output_images"):
+    doc = fitz.open(pdf_path)
+    data = []
+    
+    if not os.path.exists(output_dir):
+        os.makedirs(output_dir)
+    
+    for page_num in range(len(doc)):
+        page = doc[page_num]
+        text = page.get_text("text")
+        
+        images = page.get_images(full=True)
+        image_paths = []
+        
+        for img_index, img in enumerate(images):
+            xref = img[0]
+            base_image = doc.extract_image(xref)
+            image_bytes = base_image["image"]
+            image_ext = base_image["ext"]
+            image_filename = f"{output_dir}/page_{page_num+1}_img_{img_index+1}.{image_ext}"
+            
+            with open(image_filename, "wb") as img_file:
+                img_file.write(image_bytes)
+            
+            image_paths.append(image_filename)
+        
+        data.append({"page": page_num + 1, "text": text, "images": image_paths})
+
+    with open("pdf_data.json", "w") as f:
+        json.dump(data, f, indent=4)
+
+    return "Extraction completed!"
+
+pdf_path = "./Exelsys easyHR v10 User Guide.pdf"
+extract_text_images(pdf_path)
+
+
+# Load Hugging Face model
+model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
+
+def get_embedding(text):
+    return model.encode(text, convert_to_numpy=True)
+
+def store_embeddings():
+    with open("pdf_data.json") as f:
+        data = json.load(f)
+
+    dimension = 384  
+    index = faiss.IndexFlatL2(dimension)
+    metadata = []
+
+    for i, entry in enumerate(data):
+        embedding = np.array(get_embedding(entry["text"])).astype("float32")
+        index.add(np.array([embedding]))
+        metadata.append({"page": entry["page"], "text": entry["text"], "images": entry["images"]})
+
+    faiss.write_index(index, "faiss_index.bin")
+
+    with open("metadata.json", "w") as f:
+        json.dump(metadata, f, indent=4)
+
+    return "Embeddings stored successfully!"
+
+store_embeddings()
+
+
+
+app = Flask(__name__)
+
+# Load Model and FAISS Index
+model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
+index = faiss.read_index("faiss_index.bin")
+
+groq_api_key = os.getenv('GROQ_API_KEY')
+model_name = "llama-3.3-70b-versatile"
+
+llm = ChatGroq(
+            temperature=0, 
+            groq_api_key=groq_api_key, 
+            model_name=model_name
+        )
+
+with open("metadata.json") as f:
+    metadata = json.load(f)
+
+
+def categorize_query(query):
+    """
+    Categorizes user queries into different types (greetings, small talk, unrelated, etc.).
+    """
+    query = query.lower().strip()
+
+    # Greetings
+    greeting_patterns = [
+        r"\bhello\b", r"\bhi\b", r"\bhey\b", r"\bhola\b", r"\bgreetings\b",
+        r"\bwhat('s| is) up\b", r"\bhowdy\b", r"\bhiya\b", r"\byo\b",
+        r"\bgood (morning|afternoon|evening|day|night)\b",
+        r"\bhow (are|r) you\b", r"\bhow's it going\b", r"\bhow have you been\b",
+        r"\bhope you are (doing )?(well|good|fine)\b", r"\bnice to meet you\b",
+        r"\bpleased to meet you\b"
+    ]
+
+    # Thank-you messages
+    thank_you_patterns = [
+        r"\bthank(s| you)\b", r"\bthanks a lot\b", r"\bthanks so much\b",
+        r"\bthank you very much\b", r"\bappreciate it\b", r"\bmuch obliged\b",
+        r"\bgrateful\b", r"\bcheers\b"
+    ]
+
+    # Small talk
+    small_talk_patterns = [
+        r"\bhow (are|r) you\b", r"\bhow's your day\b", r"\bwhat's up\b",
+        r"\bhow's it going\b", r"\bhow have you been\b", r"\bhope you are well\b"
+    ]
+
+    # Unrelated topics
+    unrelated_patterns = [
+        r"\btell me a joke\b", r"\bwho won\b", r"\bwhat is ai\b", r"\bexplain blockchain\b"
+    ]
+
+
+    # Goodbye messages
+    goodbye_patterns = [
+        r"\bbye\b", r"\bgoodbye\b", r"\bsee you\b", r"\bhave a nice day\b"
+    ]
+
+    # Rude or inappropriate messages
+    rude_patterns = [
+        r"\bstupid\b", r"\bdumb\b", r"\buseless\b", r"\bshut up\b"
+    ]
+
+    if any(re.search(pattern, query) for pattern in greeting_patterns):
+        return "greeting"
+    if any(re.search(pattern, query) for pattern in thank_you_patterns):
+        return "thank_you"
+    if any(re.search(pattern, query) for pattern in small_talk_patterns):
+        return "small_talk"
+    if any(re.search(pattern, query) for pattern in unrelated_patterns):
+        return "unrelated"
+    if any(re.search(pattern, query) for pattern in goodbye_patterns):
+        return "goodbye"
+    if any(re.search(pattern, query) for pattern in rude_patterns):
+        return "rude"
+
+    return "normal"
+
+# Function to Search for Relevant Answers
+def search_text(query, top_k=2):
+    query_embedding = np.array(model.encode(query, convert_to_numpy=True)).astype("float32").reshape(1, -1)
+    distances, indices = index.search(query_embedding, top_k)
+
+    results = []
+    for idx in indices[0]:
+        if idx >= 0:
+            results.append(metadata[idx])
+    
+    return results
+
+# Serve HTML Page
+@app.route("/")
+def home():
+    return render_template("index.html")
+
+@app.route("/query", methods=["POST"])
+def query_pdf():
+    query = request.json.get("query")
+
+    query_type = categorize_query(query)
+
+    if query_type == "greeting":
+        return jsonify({"text": "Hello! How can I assist you with Exelsys EasyHR?", "images": []})
+
+    if query_type == "thank_you":
+        return jsonify({"text": "You're welcome! How can I assist you further?", "images": []})
+
+    if query_type == "small_talk":
+        return jsonify({"text": "I'm here to assist with Exelsys EasyHR. How can I help?", "images": []})
+
+    if query_type == "unrelated":
+        return jsonify({"text": "I'm here to assist with Exelsys easyHR queries only.", "images": []})
+
+    if query_type == "vague":
+        return jsonify({"text": "Could you please provide more details?", "images": []})
+
+    if query_type == "goodbye":
+        return jsonify({"text": "You're welcome! Have a great day!", "images": []})
+
+    if query_type == "rude":
+        return jsonify({"text": "I'm here to assist you professionally.", "images": []})
+
+
+
+    # Search for relevant PDF content using FAISS
+    results = search_text(query, top_k=3)
+
+    if not results:
+        return jsonify({"text": "No relevant results found in the PDF.", "images": []})
+
+    # Merge multiple text results
+    retrieved_text = "\n\n---\n\n".join([res["text"] for res in results])
+    print(retrieved_text)
+    
+    prompt_extract = PromptTemplate.from_template(
+        """
+        ### YOU ARE AN EXELSYS EASYHR GUIDE ASSISTANT:
+        ### INSTRUCTIONS:
+        - Your job is to provide step-by-step guidance for the following user query.
+        - Base your response **only** on the retrieved context from the PDF.
+        - If no relevant information is found, simply respond with: "Not found."
+        - If the user greets you (e.g., "Hello", "Hi", "Good morning"), respond politely but keep it brief.
+        - If the query is unrelated to Exelsys easyHR, respond with: "I'm here to assist with Exelsys easyHR queries only."
+
+        ### USER QUERY:
+        {query}
+
+        ### CONTEXT FROM PDF:
+        {retrieved_text}
+
+        ### ANSWER:
+        """
+    )
+
+    # Chain the prompt with ChatGroq
+    chain_extract = prompt_extract | llm
+    chat_response = chain_extract.invoke({"query": query, "retrieved_text": retrieved_text})
+
+    # Convert response to string
+    response_text = str(chat_response.content)
+
+    # Determine if images should be included
+    retrieved_images = []
+    if "Not found." not in response_text and "I'm here to assist" not in response_text:
+        retrieved_images = [img for res in results if "images" in res for img in res["images"]]
+
+    # Final response JSON
+    response = {
+        "text": response_text,
+        "images": retrieved_images
+    }
+
+    return jsonify(response)
+
+if __name__ == "__main__":
+    app.run(host="0.0.0.0", port=7860)
+
+