Create app.py

app.py

@@ -1,110 +1,59 @@
-import os
-import requests
-import fitz  # PyMuPDF for PDF reading
-import faiss
-import numpy as np
+import json
 import gradio as gr
 from sentence_transformers import SentenceTransformer
+import faiss
+import numpy as np
 from huggingface_hub import InferenceClient
 
-# 🔹 Define PDF Directory and Chunk Size
-PDF_DIR = "./pdfs"
-CHUNK_SIZE = 2500  # Larger chunks for better context
-
-# 🔹 Ensure Directory Exists
-os.makedirs(PDF_DIR, exist_ok=True)
-
-# 🔹 Direct URLs for PDF Downloads (Colorado Policy Documents)
-PDF_FILES = {
-    "SNAP 10 CCR 2506-1.pdf": "https://huggingface.co/spaces/tstone87/ccr-colorado/resolve/main/SNAP%2010%20CCR%202506-1%20.pdf?download=true",
-    "Med 10 CCR 2505-10 8.100.pdf": "https://huggingface.co/spaces/tstone87/ccr-colorado/resolve/main/Med%2010%20CCR%202505-10%208.100.pdf?download=true",
-}
+# 🔹 Load JSON Data with Colorado Food Stamp Information
+DATA_FILE = "colorado_foodstamps.json"
 
-# 🔹 Function to Download PDFs
+def load_json_data():
+    try:
+        with open(DATA_FILE, "r", encoding="utf-8") as f:
+            return json.load(f)
+    except FileNotFoundError:
+        return {"error": "Data file not found! Please upload a valid JSON file."}
 
-def download_pdfs():
-    for filename, url in PDF_FILES.items():
-        pdf_path = os.path.join(PDF_DIR, filename)
-        if not os.path.exists(pdf_path):
-            print(f"📥 Downloading {filename}...")
-            try:
-                response = requests.get(url, stream=True)
-                response.raise_for_status()
-                with open(pdf_path, "wb") as f:
-                    for chunk in response.iter_content(chunk_size=8192):
-                        f.write(chunk)
-                print(f"✅ Downloaded {filename}")
-            except Exception as e:
-                print(f"❌ Error downloading {filename}: {e}")
+data = load_json_data()
 
-# 🔹 Function to Extract Text from PDFs
+# 🔹 Initialize FAISS for Searching Relevant Answers
+model = SentenceTransformer("multi-qa-mpnet-base-dot-v1")
 
-def extract_text_from_pdfs():
-    all_text = ""
-    for pdf_file in os.listdir(PDF_DIR):
-        if pdf_file.endswith(".pdf"):
-            pdf_path = os.path.join(PDF_DIR, pdf_file)
-            doc = fitz.open(pdf_path)
-            for page in doc:
-                all_text += page.get_text("text") + "\n"
-    return all_text
-
-# 🔹 Initialize FAISS Index
-
-def initialize_faiss():
-    download_pdfs()
-    text_data = extract_text_from_pdfs()
-    if not text_data:
-        raise ValueError("❌ No text extracted from PDFs!")
-    
-    chunks = [text_data[i:i+CHUNK_SIZE] for i in range(0, len(text_data), CHUNK_SIZE)]
-    model = SentenceTransformer("multi-qa-mpnet-base-dot-v1")
-    embeddings = np.array([model.encode(chunk) for chunk in chunks])
+def create_faiss_index(data):
+    texts = list(data.values())
+    embeddings = np.array([model.encode(text) for text in texts])
     index = faiss.IndexFlatL2(embeddings.shape[1])
     index.add(embeddings)
-    print("✅ FAISS index initialized.")
-    return index, chunks
-
-# 🔹 Initialize FAISS on Startup
-index, chunks = initialize_faiss()
+    return index, texts
 
-# 🔹 Function to Search FAISS
+index, texts = create_faiss_index(data)
 
-def search_policy(query, top_k=3):
-    query_embedding = SentenceTransformer("multi-qa-mpnet-base-dot-v1").encode(query).reshape(1, -1)
+# 🔹 Function to Search FAISS for Relevant Answers
+def search_faiss(query, top_k=1):
+    query_embedding = model.encode(query).reshape(1, -1)
     distances, indices = index.search(query_embedding, top_k)
-    return "\n\n".join([chunks[i] for i in indices[0] if i < len(chunks)])
+    return texts[indices[0][0]] if indices[0][0] < len(texts) else "No relevant information found."
 
-# 🔹 Hugging Face LLM Client
+# 🔹 Hugging Face API for Additional Responses
 client = InferenceClient("HuggingFaceH4/zephyr-7b-beta")
 
-# 🔹 Function to Handle Chat Responses
-
-def respond(message, history):
-    messages = [{"role": "system", "content": "You are a chatbot specializing in Colorado public assistance programs."}]
-    for val in history:
-        if val[0]:
-            messages.append({"role": "user", "content": val[0]})
-        if val[1]:
-            messages.append({"role": "assistant", "content": val[1]})
-    
-    policy_context = search_policy(message)
-    if policy_context:
-        messages.append({"role": "assistant", "content": f"📄 **Colorado Policy Info:**\n\n{policy_context}"})
-    
-    messages.append({"role": "user", "content": message})
+def get_huggingface_response(query):
+    messages = [{"role": "system", "content": "Provide accurate food stamp information for Colorado."},
+                {"role": "user", "content": query}]
     response = ""
     for message in client.chat_completion(messages, max_tokens=512, stream=True, temperature=0.7, top_p=0.95):
-        token = message.choices[0].delta.content
-        response += token
-        yield response
-
-# 🔹 Gradio Chat Interface (Colorado-Themed)
-demo = gr.ChatInterface(
-    respond,
-    textbox=gr.Textbox(placeholder="Ask about Colorado public assistance programs...", interactive=True, show_label=False),
-    submit_btn=gr.Button("Send"),
-    chatbot=gr.Chatbot(),
-)
+        response += message.choices[0].delta.content
+    return response
+
+# 🔹 Main Chatbot Function
+def chatbot_response(message, history):
+    relevant_info = search_faiss(message)
+    if "No relevant information found." not in relevant_info:
+        return relevant_info
+    return get_huggingface_response(message)
+
+# 🔹 Gradio Chat Interface
+demo = gr.ChatInterface(chatbot_response, textbox=gr.Textbox(placeholder="Ask about Colorado food stamps..."))
 
 demo.launch()