Spaces:

Talha812
/

AIPaperPilot

Sleeping

File size: 19,045 Bytes

# # import streamlit as st
# # import torch
# # from transformers import GPTNeoXForCausalLM, AutoTokenizer
# # from sentence_transformers import SentenceTransformer
# # import faiss
# # import fitz  # PyMuPDF
# # from langchain_text_splitters import RecursiveCharacterTextSplitter

# # # 1. Set page config FIRST
# # st.set_page_config(page_title="📚 Smart Book Analyst", layout="wide")

# # # Configuration
# # MODEL_NAME = "ibm-granite/granite-3.1-1b-a400m-instruct"
# # EMBED_MODEL = "sentence-transformers/all-mpnet-base-v2"
# # DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
# # CHUNK_SIZE = 512
# # CHUNK_OVERLAP = 50

# # @st.cache_resource
# # def load_models():
# #     try:
# #         # Load Granite model
# #         tokenizer = AutoTokenizer.from_pretrained(
# #             MODEL_NAME,
# #             trust_remote_code=True
# #         )
        
# #         model = GPTNeoXForCausalLM.from_pretrained(
# #             MODEL_NAME,
# #             device_map="auto" if DEVICE == "cuda" else None,
# #             torch_dtype=torch.float16 if DEVICE == "cuda" else torch.float32,
# #             trust_remote_code=True
# #         ).eval()
        
# #         # Load sentence transformer for embeddings
# #         embedder = SentenceTransformer(EMBED_MODEL, device=DEVICE)
        
# #         return tokenizer, model, embedder
    
# #     except Exception as e:
# #         st.error(f"Model loading failed: {str(e)}")
# #         st.stop()

# # tokenizer, model, embedder = load_models()

# # # Text processing
# # def process_text(text):
# #     splitter = RecursiveCharacterTextSplitter(
# #         chunk_size=CHUNK_SIZE,
# #         chunk_overlap=CHUNK_OVERLAP,
# #         length_function=len
# #     )
# #     return splitter.split_text(text)

# # # PDF extraction
# # def extract_pdf_text(uploaded_file):
# #     try:
# #         doc = fitz.open(stream=uploaded_file.read(), filetype="pdf")
# #         return "\n".join([page.get_text() for page in doc])
# #     except Exception as e:
# #         st.error(f"PDF extraction error: {str(e)}")
# #         return ""

# # # Summarization function
# # def generate_summary(text):
# #     chunks = process_text(text)[:10]
# #     summaries = []
    
# #     for chunk in chunks:
# #         prompt = f"""<|user|>
# #         Summarize this text section focusing on key themes, characters, and plot points:
# #         {chunk[:2000]}
# #         <|assistant|>
# #         """
        
# #         inputs = tokenizer(prompt, return_tensors="pt").to(DEVICE)
# #         outputs = model.generate(**inputs, max_new_tokens=300, temperature=0.3)
# #         summaries.append(tokenizer.decode(outputs[0], skip_special_tokens=True))
    
# #     combined = "\n".join(summaries)
# #     final_prompt = f"""<|user|>
# #     Combine these section summaries into a coherent book summary:
# #     {combined}
# #     <|assistant|>
# #     The comprehensive summary is:"""
    
# #     inputs = tokenizer(final_prompt, return_tensors="pt").to(DEVICE)
# #     outputs = model.generate(**inputs, max_new_tokens=500, temperature=0.5)
# #     return tokenizer.decode(outputs[0], skip_special_tokens=True).split(":")[-1].strip()

# # # FAISS index creation
# # def build_faiss_index(texts):
# #     embeddings = embedder.encode(texts, show_progress_bar=False)
# #     dimension = embeddings.shape[1]
# #     index = faiss.IndexFlatIP(dimension)
# #     faiss.normalize_L2(embeddings)
# #     index.add(embeddings)
# #     return index

# # # Answer generation
# # def generate_answer(query, context):
# #     prompt = f"""<|user|>
# #     Using this context: {context}
# #     Answer the question precisely and truthfully. If unsure, say "I don't know".
# #     Question: {query}
# #     <|assistant|>
# #     """
    
# #     inputs = tokenizer(prompt, return_tensors="pt", max_length=1024, truncation=True).to(DEVICE)
# #     outputs = model.generate(
# #         **inputs,
# #         max_new_tokens=300,
# #         temperature=0.4,
# #         top_p=0.9,
# #         repetition_penalty=1.2,
# #         do_sample=True
# #     )
# #     return tokenizer.decode(outputs[0], skip_special_tokens=True).split("<|assistant|>")[-1].strip()

# # # Streamlit UI
# # st.title("📚 AI-Powered Book Analysis System")

# # uploaded_file = st.file_uploader("Upload book (PDF or TXT)", type=["pdf", "txt"])

# # if uploaded_file:
# #     with st.spinner("📖 Analyzing book content..."):
# #         try:
# #             if uploaded_file.type == "application/pdf":
# #                 text = extract_pdf_text(uploaded_file)
# #             else:
# #                 text = uploaded_file.read().decode()
            
# #             chunks = process_text(text)
# #             st.session_state.docs = chunks
# #             st.session_state.index = build_faiss_index(chunks)
            
# #             with st.expander("📝 Book Summary", expanded=True):
# #                 summary = generate_summary(text)
# #                 st.write(summary)
                
# #         except Exception as e:
# #             st.error(f"Processing failed: {str(e)}")

# # if 'index' in st.session_state and st.session_state.index:
# #     query = st.text_input("Ask about the book:")
# #     if query:
# #         with st.spinner("🔍 Searching for answers..."):
# #             try:
# #                 query_embed = embedder.encode([query])
# #                 faiss.normalize_L2(query_embed)
# #                 distances, indices = st.session_state.index.search(query_embed, k=3)
                
# #                 context = "\n".join([st.session_state.docs[i] for i in indices[0]])
# #                 answer = generate_answer(query, context)
                
# #                 st.subheader("Answer")
# #                 st.markdown(f"```\n{answer}\n```")
# #                 st.caption("Retrieved context confidence: {:.2f}".format(distances[0][0]))
                
# #             except Exception as e:
# #                 st.error(f"Query failed: {str(e)}")


# import streamlit as st
# import torch
# from transformers import GPTNeoXForCausalLM, AutoTokenizer
# from sentence_transformers import SentenceTransformer
# import faiss
# import fitz
# from langchain_text_splitters import RecursiveCharacterTextSplitter

# # Set page config FIRST
# st.set_page_config(page_title="📚 Smart Book Analyst", layout="wide")

# # Configuration
# MODEL_NAME = "ibm-granite/granite-3.1-1b-a400m-instruct"
# EMBED_MODEL = "sentence-transformers/all-mpnet-base-v2"
# DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
# CHUNK_SIZE = 1024  # Increased chunk size for better performance
# CHUNK_OVERLAP = 100
# MAX_SUMMARY_CHUNKS = 5  # Reduced from 10 to 5 for faster processing

# @st.cache_resource
# def load_models():
#     try:
#         # Load model with optimized settings
#         tokenizer = AutoTokenizer.from_pretrained(
#             MODEL_NAME,
#             trust_remote_code=True
#         )
        
#         model = GPTNeoXForCausalLM.from_pretrained(
#             MODEL_NAME,
#             device_map="auto",
#             torch_dtype=torch.float16 if DEVICE == "cuda" else torch.float32,
#             trust_remote_code=True,
#             low_cpu_mem_usage=True
#         ).eval()
        
#         # Load embedder with faster model
#         embedder = SentenceTransformer(EMBED_MODEL, device=DEVICE)
#         embedder.max_seq_length = 256  # Reduce embedding dimension
        
#         return tokenizer, model, embedder
    
#     except Exception as e:
#         st.error(f"Model loading failed: {str(e)}")
#         st.stop()

# tokenizer, model, embedder = load_models()

# def process_text(text):
#     splitter = RecursiveCharacterTextSplitter(
#         chunk_size=CHUNK_SIZE,
#         chunk_overlap=CHUNK_OVERLAP,
#         length_function=len
#     )
#     return splitter.split_text(text)

# def extract_pdf_text(uploaded_file):
#     try:
#         doc = fitz.open(stream=uploaded_file.read(), filetype="pdf")
#         return "\n".join(page.get_text() for page in doc)
#     except Exception as e:
#         st.error(f"PDF extraction error: {str(e)}")
#         return ""

# def generate_summary(text):
#     chunks = process_text(text)[:MAX_SUMMARY_CHUNKS]
#     if not chunks:
#         return "No meaningful content found."
    
#     progress_bar = st.progress(0)
#     summaries = []
    
#     for i, chunk in enumerate(chunks):
#         progress_bar.progress((i+1)/len(chunks), text=f"Processing chunk {i+1}/{len(chunks)}...")
#         prompt = f"""<|user|>
#         Summarize key points in 2 sentences:
#         {chunk[:1500]}
#         <|assistant|>
#         """
        
#         inputs = tokenizer(prompt, return_tensors="pt").to(DEVICE)
#         outputs = model.generate(
#             **inputs,
#             max_new_tokens=150,
#             temperature=0.2,
#             do_sample=False  # Disable sampling for faster generation
#         )
#         summaries.append(tokenizer.decode(outputs[0], skip_special_tokens=True))
    
#     combined = "\n".join(summaries)
#     final_prompt = f"""<|user|>
#     Combine these into a concise summary (3-5 paragraphs):
#     {combined}
#     <|assistant|>
#     Summary:"""
    
#     inputs = tokenizer(final_prompt, return_tensors="pt").to(DEVICE)
#     outputs = model.generate(
#         **inputs,
#         max_new_tokens=300,
#         temperature=0.3,
#         do_sample=False
#     )
#     return tokenizer.decode(outputs[0], skip_special_tokens=True).split("Summary:")[-1].strip()

# def build_faiss_index(texts):
#     embeddings = embedder.encode(texts, show_progress_bar=False, batch_size=32)
#     dimension = embeddings.shape[1]
#     index = faiss.IndexFlatIP(dimension)
#     faiss.normalize_L2(embeddings)
#     index.add(embeddings)
#     return index

# def generate_answer(query, context):
#     prompt = f"""<|user|>
#     Context: {context[:2000]}
#     Q: {query}
#     A:"""
    
#     inputs = tokenizer(prompt, return_tensors="pt", max_length=1024, truncation=True).to(DEVICE)
#     outputs = model.generate(
#         **inputs,
#         max_new_tokens=200,
#         temperature=0.3,
#         top_p=0.85,
#         repetition_penalty=1.1,
#         do_sample=True
#     )
#     return tokenizer.decode(outputs[0], skip_special_tokens=True).split("A:")[-1].strip()

# # Streamlit UI
# st.title("📚 AI-Powered Book Analysis System")

# uploaded_file = st.file_uploader("Upload book (PDF or TXT)", type=["pdf", "txt"])

# if uploaded_file:
#     with st.spinner("📖 Analyzing book content..."):
#         try:
#             if uploaded_file.type == "application/pdf":
#                 text = extract_pdf_text(uploaded_file)
#             else:
#                 text = uploaded_file.read().decode()
            
#             if not text.strip():
#                 st.error("Uploaded file appears to be empty")
#                 st.stop()
                
#             chunks = process_text(text)
#             st.session_state.docs = chunks
#             st.session_state.index = build_faiss_index(chunks)
            
#             with st.expander("📝 Book Summary", expanded=True):
#                 summary = generate_summary(text)
#                 st.write(summary)
                
#         except Exception as e:
#             st.error(f"Processing failed: {str(e)}")

# if 'index' in st.session_state and st.session_state.index:
#     query = st.text_input("Ask about the book:")
#     if query:
#         with st.spinner("🔍 Searching for answers..."):
#             try:
#                 query_embed = embedder.encode([query])
#                 faiss.normalize_L2(query_embed)
#                 distances, indices = st.session_state.index.search(query_embed, k=2)
                
#                 context = "\n".join([st.session_state.docs[i] for i in indices[0]])
#                 answer = generate_answer(query, context)
                
#                 st.subheader("Answer")
#                 st.markdown(f"```\n{answer}\n```")
#                 st.caption(f"Confidence: {distances[0][0]:.2f}")
                
#             except Exception as e:
#                 st.error(f"Query failed: {str(e)}")


import streamlit as st
import torch
from transformers import GPTNeoXForCausalLM, AutoTokenizer
from sentence_transformers import SentenceTransformer
import faiss
import fitz
from langchain_text_splitters import RecursiveCharacterTextSplitter

# Set page config first
st.set_page_config(page_title="📚 Smart Book Analyst", layout="wide")

# Configuration
MODEL_NAME = "ibm-granite/granite-3.1-1b-a400m-instruct"
EMBED_MODEL = "sentence-transformers/all-mpnet-base-v2"
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
CHUNK_SIZE = 1024
CHUNK_OVERLAP = 100
MAX_SUMMARY_CHUNKS = 5

@st.cache_resource
def load_models():
    try:
        # Load model with correct tokenizer mapping
        tokenizer = AutoTokenizer.from_pretrained(
            MODEL_NAME,
            trust_remote_code=True,
            padding_side="left"  # Crucial for generation quality
        )
        tokenizer.pad_token = tokenizer.eos_token
        
        model = GPTNeoXForCausalLM.from_pretrained(
            MODEL_NAME,
            device_map="auto",
            torch_dtype=torch.float16 if DEVICE == "cuda" else torch.float32,
            trust_remote_code=True,
            low_cpu_mem_usage=True
        ).eval()
        
        # Configure embedder properly
        embedder = SentenceTransformer(EMBED_MODEL, device=DEVICE)
        embedder.max_seq_length = 512
        
        return tokenizer, model, embedder
    
    except Exception as e:
        st.error(f"Model loading failed: {str(e)}")
        st.stop()

tokenizer, model, embedder = load_models()

def process_text(text):
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=CHUNK_SIZE,
        chunk_overlap=CHUNK_OVERLAP,
        length_function=len
    )
    return splitter.split_text(text)

def extract_pdf_text(uploaded_file):
    try:
        doc = fitz.open(stream=uploaded_file.read(), filetype="pdf")
        return "\n".join(page.get_text() for page in doc)
    except Exception as e:
        st.error(f"PDF extraction error: {str(e)}")
        return ""

def generate_summary(text):
    chunks = process_text(text)[:MAX_SUMMARY_CHUNKS]
    if not chunks:
        return "No meaningful content found."
    
    progress_bar = st.progress(0)
    summaries = []
    
    for i, chunk in enumerate(chunks):
        # Proper progress text formatting
        progress_bar.progress((i+1)/len(chunks), 
                            text=f"Processing section {i+1}/{len(chunks)}...")
        
        prompt = f"""<|user|>
        Summarize the key points from this text section in 3 bullet points:
        {chunk[:1500]}
        <|assistant|>
        """
        
        inputs = tokenizer(
            prompt,
            return_tensors="pt",
            max_length=1024,
            truncation=True
        ).to(DEVICE)
        
        outputs = model.generate(
            **inputs,
            max_new_tokens=200,
            temperature=0.3,
            top_p=0.9,
            repetition_penalty=1.1,
            do_sample=True,
            pad_token_id=tokenizer.eos_token_id  # Critical fix
        )
        
        decoded = tokenizer.decode(
            outputs[0],
            skip_special_tokens=True
        ).split("<|assistant|>")[-1].strip()
        
        summaries.append(decoded)
    
    combined = "\n\n".join(summaries)
    final_prompt = f"""<|user|>
    Combine these bullet points into a coherent 3-paragraph summary:
    {combined}
    <|assistant|>
    Here is the comprehensive summary:"""
    
    inputs = tokenizer(final_prompt, return_tensors="pt").to(DEVICE)
    outputs = model.generate(
        **inputs,
        max_new_tokens=400,
        temperature=0.5,
        top_p=0.9,
        repetition_penalty=1.1,
        do_sample=True,
        pad_token_id=tokenizer.eos_token_id
    )
    
    return tokenizer.decode(
        outputs[0],
        skip_special_tokens=True
    ).split("Here is the comprehensive summary:")[-1].strip()

def build_faiss_index(texts):
    embeddings = embedder.encode(
        texts,
        show_progress_bar=False,
        batch_size=16,
        convert_to_tensor=True
    ).cpu().numpy()
    
    dimension = embeddings.shape[1]
    index = faiss.IndexFlatIP(dimension)
    faiss.normalize_L2(embeddings)
    index.add(embeddings)
    return index

def generate_answer(query, context):
    prompt = f"""<|user|>
    Based on this context:
    {context[:2000]}
    
    Answer this question concisely: {query}
    <|assistant|>
    """
    
    inputs = tokenizer(
        prompt,
        return_tensors="pt",
        max_length=1024,
        truncation=True
    ).to(DEVICE)
    
    outputs = model.generate(
        **inputs,
        max_new_tokens=300,
        temperature=0.4,
        top_p=0.95,
        repetition_penalty=1.15,
        do_sample=True,
        pad_token_id=tokenizer.eos_token_id,
        no_repeat_ngram_size=3  # Prevent repetition
    )
    
    return tokenizer.decode(
        outputs[0],
        skip_special_tokens=True
    ).split("<|assistant|>")[-1].strip()

# Streamlit UI
st.title("📚 AI-Powered Book Analysis System")

uploaded_file = st.file_uploader("Upload book (PDF or TXT)", type=["pdf", "txt"])

if uploaded_file:
    with st.spinner("📖 Analyzing book content..."):
        try:
            if uploaded_file.type == "application/pdf":
                text = extract_pdf_text(uploaded_file)
            else:
                text = uploaded_file.read().decode()
            
            if not text.strip():
                st.error("Uploaded file is empty")
                st.stop()
                
            chunks = process_text(text)
            st.session_state.docs = chunks
            st.session_state.index = build_faiss_index(chunks)
            
            with st.expander("📝 Book Summary", expanded=True):
                summary = generate_summary(text)
                st.write(summary)
                
        except Exception as e:
            st.error(f"Processing failed: {str(e)}")

if 'index' in st.session_state and st.session_state.index:
    query = st.text_input("Ask about the book:")
    if query:
        with st.spinner("🔍 Searching for answers..."):
            try:
                query_embed = embedder.encode([query])
                faiss.normalize_L2(query_embed)
                distances, indices = st.session_state.index.search(query_embed, k=3)
                
                context = "\n".join([st.session_state.docs[i] for i in indices[0]])
                answer = generate_answer(query, context)
                
                st.subheader("Answer")
                st.markdown(f"```\n{answer}\n```")
                st.caption(f"Confidence score: {distances[0][0]:.2f}")
                
            except Exception as e:
                st.error(f"Query failed: {str(e)}")