Spaces:
Runtime error
Runtime error
from fastapi import FastAPI, UploadFile, File | |
from fastapi.responses import HTMLResponse | |
from fastapi.staticfiles import StaticFiles | |
import os | |
from dotenv import load_dotenv | |
from PyPDF2 import PdfReader | |
from langchain.text_splitter import RecursiveCharacterTextSplitter | |
from langchain_google_genai import GoogleGenerativeAIEmbeddings | |
from langchain_community.vectorstores import FAISS | |
from langchain_google_genai import ChatGoogleGenerativeAI | |
from langchain.chains.question_answering import load_qa_chain | |
from langchain.prompts import PromptTemplate | |
import logging | |
app = FastAPI() | |
# Set up logging | |
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s: %(message)s') | |
load_dotenv() | |
genai.configure(api_key=os.getenv("AIzaSyD0GxR2J1JxGic807Cc89Jq6MB4aDJYgDc")) | |
def get_pdf_text(pdf_docs): | |
text = "" | |
for pdf in pdf_docs: | |
try: | |
pdf_reader = PdfReader(pdf.file) | |
for page in pdf_reader.pages: | |
text += page.extract_text() | |
except Exception as e: | |
logging.error(f"Error processing PDF file: {e}") | |
return text | |
def get_text_chunks(text): | |
text_splitter = RecursiveCharacterTextSplitter( | |
chunk_size=10000, | |
chunk_overlap=1000 | |
) | |
chunks = text_splitter.split_text(text) | |
return chunks | |
def get_vector_store(text_chunks): | |
logging.info("Starting vector store creation") | |
embeddings = GoogleGenerativeAIEmbeddings(model="models/embedding-001") | |
vector_store = FAISS.from_texts(texts=text_chunks, embedding=embeddings) | |
logging.info("FAISS vector store created") | |
faiss_index_dir = os.path.join(os.path.dirname(__file__), "faiss_index") | |
os.makedirs(faiss_index_dir, exist_ok=True) | |
vector_store.save_local(faiss_index_dir) | |
logging.info("FAISS vector store saved successfully.") | |
def get_conversation_chain(): | |
prompt_template = """ | |
Answer the question clear and precise. If not provided the context return the result as | |
"Sorry I dont know the answer", don't provide the wrong answer. | |
Context:\n {context}?\n | |
Question:\n{question}\n | |
Answer: | |
""" | |
model = ChatGoogleGenerativeAI(model="gemini-pro", temperature=0.3) | |
prompt = PromptTemplate(template=prompt_template, input_variables=['context', 'question']) | |
chain = load_qa_chain(model, chain_type='stuff', prompt=prompt) | |
return chain | |
def user_input(user_question): | |
logging.info("Processing user input") | |
faiss_index_dir = os.path.join(os.path.dirname(__file__), "faiss_index") | |
if not os.path.exists(faiss_index_dir): | |
return "Please upload and process PDF files before asking questions." | |
try: | |
new_db = FAISS.load_local(faiss_index_dir, GoogleGenerativeAIEmbeddings(model='models/embedding-001'), allow_dangerous_deserialization=True) | |
logging.info("FAISS vector store loaded successfully") | |
docs = new_db.similarity_search(user_question) | |
chain = get_conversation_chain() | |
response = chain({"input_documents": docs, "question": user_question}, return_only_outputs=True) | |
return response["output_text"] | |
except Exception as e: | |
logging.error(f"Error processing user input: {e}") | |
return f"Sorry, there was an error processing your request: {str(e)}. Please try again later." | |
async def upload_pdf(pdf_docs: list[UploadFile] = File(...)): | |
raw_text = get_pdf_text(pdf_docs) | |
text_chunks = get_text_chunks(raw_text) | |
get_vector_store(text_chunks) | |
return {"message": "PDFs processed successfully. You can now ask questions."} | |
async def ask_question(user_question: str): | |
response = user_input(user_question) | |
return {"response": response} | |
async def read_root(): | |
return """ | |
<html> | |
<head> | |
<title>Chat with PDFs</title> | |
</head> | |
<body> | |
<h1>Welcome to Chat with PDFs API</h1> | |
<p>Use POST /upload_pdf/ to upload PDF files.</p> | |
<p>Use GET /ask_question/ to ask questions from the PDFs you uploaded.</p> | |
</body> | |
</html> | |
""" |