Spaces:
Sleeping
Sleeping
File size: 3,662 Bytes
c4b7a63 f0c6ef0 c4b7a63 |
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 112 113 114 |
import os
from typing import List, Optional
import gradio as gr
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import TextLoader
from langchain.docstore.document import Document
from langchain.chains import RetrievalQA
from langchain.llms.base import LLM
from groq import Groq
import pypdf # PyMuPDF
# --- Custom LLM class using Groq ---
class GroqLLM(LLM):
model: str = "llama3-8b-8192"
api_key: str = "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" # <-- PUT YOUR GROQ API KEY HERE
temperature: float = 0.0
def _call(self, prompt: str, stop: Optional[List[str]] = None) -> str:
client = Groq(api_key=self.api_key)
messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": prompt}
]
response = client.chat.completions.create(
model=self.model,
messages=messages,
temperature=self.temperature,
)
return response.choices[0].message.content
@property
def _llm_type(self) -> str:
return "groq-llm"
# --- RAG Setup ---
retriever = None
qa_chain = None
def extract_text_from_pdf(file_path: str) -> str:
doc = fitz.open(file_path)
text = ""
for page in doc:
text += page.get_text()
doc.close()
return text
def process_file(file_obj):
global retriever, qa_chain
ext = os.path.splitext(file_obj.name)[1].lower()
try:
# Load content
if ext == ".pdf":
text = extract_text_from_pdf(file_obj.name)
elif ext == ".txt":
with open(file_obj.name, "r", encoding="utf-8") as f:
text = f.read()
else:
return "❌ Unsupported file format. Please upload a .txt or .pdf file."
# Create document chunks
document = Document(page_content=text)
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
docs = splitter.split_documents([document])
# Vectorstore with HuggingFace embeddings
embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")
vectorstore = Chroma.from_documents(docs, embedding=embeddings, persist_directory="rag_store")
retriever = vectorstore.as_retriever()
qa_chain = RetrievalQA.from_chain_type(
llm=GroqLLM(),
retriever=retriever,
return_source_documents=True
)
return "✅ File processed successfully. You can now ask questions."
except Exception as e:
return f"❌ Error processing file: {e}"
def ask_question(query):
if qa_chain is None:
return "⚠ Please upload a file first."
result = qa_chain({"query": query})
return result["result"]
# --- Gradio UI ---
with gr.Blocks(title="RAG PDF & Text Chatbot") as demo:
gr.Markdown("## 🧠 RAG-powered Q&A Chatbot (Groq + LangChain)")
gr.Markdown("Upload a .pdf or .txt file and ask questions based on its content.")
file_input = gr.File(label="Upload PDF or Text File", file_types=[".pdf", ".txt"])
upload_status = gr.Textbox(label="Status", interactive=False)
file_input.change(fn=process_file, inputs=file_input, outputs=upload_status)
question_box = gr.Textbox(label="Ask your question")
answer_box = gr.Textbox(label="Answer", interactive=False)
submit_btn = gr.Button("Get Answer")
submit_btn.click(fn=ask_question, inputs=question_box, outputs=answer_box)
demo.launch()
|