Spaces:
Running
Running
File size: 4,014 Bytes
0864cdf |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 |
import os
import requests
import fitz # PyMuPDF for PDF reading
import faiss
import numpy as np
import gradio as gr
from sentence_transformers import SentenceTransformer
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",
}
# 🔹 Function to Download PDFs
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}")
# 🔹 Function to Extract Text from PDFs
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])
index = faiss.IndexFlatL2(embeddings.shape[1])
index.add(embeddings)
print("✅ FAISS index initialized.")
return index, chunks
# 🔹 Initialize FAISS on Startup
index, chunks = initialize_faiss()
# 🔹 Function to Search FAISS
def search_policy(query, top_k=3):
query_embedding = SentenceTransformer("multi-qa-mpnet-base-dot-v1").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)])
# 🔹 Hugging Face LLM Client
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})
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(),
)
demo.launch()
|