Spaces:
Sleeping
Sleeping
File size: 3,260 Bytes
f37e775 486ea50 f37e775 486ea50 f37e775 486ea50 f37e775 486ea50 f37e775 |
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 |
import streamlit as st
import PyPDF2
import tiktoken
import faiss
import numpy as np
import os
from sentence_transformers import SentenceTransformer
import requests
# Load embedding model
embedding_model = SentenceTransformer("all-MiniLM-L6-v2")
# GROQ API configuration (🔐 loaded securely from environment variable)
GROQ_API_KEY = os.getenv("GROQ_API_KEY")
GROQ_URL = "https://api.groq.com/openai/v1/chat/completions"
LLAMA3_MODEL = "llama3-8b-8192"
# Extract text from PDF
def load_pdf(pdf_file):
reader = PyPDF2.PdfReader(pdf_file)
return "".join(page.extract_text() for page in reader.pages)
# Chunk text
def chunk_text(text, chunk_size=500):
words = text.split()
return [' '.join(words[i:i+chunk_size]) for i in range(0, len(words), chunk_size)]
# Generate embeddings
def get_embeddings(chunks):
return embedding_model.encode(chunks)
# Create FAISS index
def create_faiss_index(embeddings):
index = faiss.IndexFlatL2(embeddings.shape[1])
index.add(np.array(embeddings))
return index
# Search index
def search_index(index, query, chunks, top_k=3):
q_embed = embedding_model.encode([query])
_, indices = index.search(np.array(q_embed), top_k)
return [chunks[i] for i in indices[0]]
# Generate answer using GROQ
def generate_answer(prompt):
if not GROQ_API_KEY:
return "🚫 GROQ API key not found. Please set it in environment variables."
headers = {
"Authorization": f"Bearer {GROQ_API_KEY}",
"Content-Type": "application/json"
}
data = {
"model": LLAMA3_MODEL,
"messages": [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": prompt}
]
}
response = requests.post(GROQ_URL, headers=headers, json=data)
response.raise_for_status()
return response.json()["choices"][0]["message"]["content"]
# Streamlit UI
def main():
st.set_page_config("RAG App", layout="centered")
st.title("📄 PDF QA App with LLaMA 3 & GROQ")
uploaded_file = st.file_uploader("Upload a PDF", type="pdf")
if uploaded_file and st.button("Process PDF"):
with st.spinner("Processing..."):
text = load_pdf(uploaded_file)
chunks = chunk_text(text)
embeddings = get_embeddings(chunks)
index = create_faiss_index(embeddings)
st.session_state.chunks = chunks
st.session_state.index = index
st.success("✅ PDF processed and indexed.")
if "index" in st.session_state:
query = st.text_input("Ask a question about the PDF:")
if st.button("Get Answer"):
with st.spinner("Thinking..."):
top_chunks = search_index(st.session_state.index, query, st.session_state.chunks)
context = "\n\n".join(top_chunks)
prompt = f"Use the following context to answer the question:\n\n{context}\n\nQuestion: {query}"
try:
answer = generate_answer(prompt)
st.markdown("### 🧠 Answer:")
st.write(answer)
except requests.exceptions.HTTPError as e:
st.error(f"❌ API Error: {e}")
if __name__ == "__main__":
main()
|