TopEdu_Demo / utils /rag_system.py
yyzsna's picture
Upload folder using huggingface_hub
102c695 verified
import os
import uuid
import tempfile
from typing import List, Optional, Dict, Any
from pathlib import Path
import PyPDF2
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain_community.document_loaders import PyPDFLoader
from langchain.schema import Document
from dotenv import load_dotenv
from datetime import datetime
import json
import base64
from openai import OpenAI
import re
from semantic_chunking import SemanticChunker
# Load environment variables
load_dotenv()
class AlternativeEmbeddings:
"""Alternative embeddings using Sentence Transformers when OpenAI is not available"""
def __init__(self):
self.model = None
self.embedding_size = 384
try:
from sentence_transformers import SentenceTransformer
# Try smaller models in order of preference for better cloud compatibility
model_options = [
("all-MiniLM-L6-v2", 384), # Very small and reliable
("paraphrase-MiniLM-L3-v2", 384), # Even smaller
("BAAI/bge-small-en-v1.5", 384) # Original choice
]
for model_name, embed_size in model_options:
try:
print(f"πŸ”„ Trying to load model: {model_name}")
self.model = SentenceTransformer(model_name)
self.embedding_size = embed_size
print(f"βœ… Successfully loaded: {model_name}")
break
except Exception as e:
print(f"⚠️ Failed to load {model_name}: {str(e)}")
continue
if not self.model:
raise Exception("All embedding models failed to load")
except ImportError:
print("❌ sentence-transformers not available. Please install it or provide OpenAI API key.")
raise ImportError("sentence-transformers not available")
def embed_documents(self, texts):
if not self.model:
raise Exception("No embedding model available")
try:
return self.model.encode(texts, convert_to_numpy=True).tolist()
except Exception as e:
print(f"Error encoding documents: {e}")
raise
def embed_query(self, text):
if not self.model:
raise Exception("No embedding model available")
try:
return self.model.encode([text], convert_to_numpy=True)[0].tolist()
except Exception as e:
print(f"Error encoding query: {e}")
raise
class SEALionLLM:
"""Custom LLM class for SEA-LION models"""
def __init__(self):
self.client = OpenAI(
api_key=os.getenv("SEA_LION_API_KEY"),
base_url=os.getenv("SEA_LION_BASE_URL", "https://api.sea-lion.ai/v1")
)
# Model configurations
self.instruct_model = "aisingapore/Gemma-SEA-LION-v3-9B-IT"
self.reasoning_model = "aisingapore/Llama-SEA-LION-v3.5-8B-R"
def _is_complex_query(self, query: str) -> bool:
"""Determine if query requires reasoning model or simple instruct model"""
# Keywords that indicate complex university search queries
complex_keywords = [
"university", "admission", "requirement", "tuition", "fee", "program", "course",
"degree", "master", "bachelor", "phd", "scholarship", "deadline", "application",
"budget", "under", "less than", "below", "compare", "recommend", "suggest",
"which", "what are the", "show me", "find me", "search for",
# Chinese keywords
"ε€§ε­¦", "ε­¦θ΄Ή", "δΈ“δΈš", "瑕士", "学士", "博士", "η”³θ―·", "要求", "ε₯–学金",
# Malay keywords
"universiti", "yuran", "program", "ijazah", "syarat", "permohonan",
# Thai keywords
"ΰΈ‘ΰΈ«ΰΈ²ΰΈ§ΰΈ΄ΰΈ—ΰΈ’ΰΈ²ΰΈ₯ΰΈ±ΰΈ’", "ΰΈ„ΰΉˆΰΈ²ΰΉ€ΰΈ₯ΰΉˆΰΈ²ΰΉ€ΰΈ£ΰΈ΅ΰΈ’ΰΈ™", "ΰΈ«ΰΈ₯ักΰΈͺΰΈΉΰΈ•ΰΈ£", "ปริญญา", "ΰΉ€ΰΈ‡ΰΈ·ΰΉˆΰΈ­ΰΈ™ΰΉ„ΰΈ‚",
# Indonesian keywords
"universitas", "biaya", "kuliah", "program", "sarjana", "persyaratan"
]
# Check for multiple criteria (indicates complex search)
criteria_count = 0
query_lower = query.lower()
for keyword in complex_keywords:
if keyword.lower() in query_lower:
criteria_count += 1
# Also check for comparison words, numbers, conditions
comparison_patterns = [
r"under \$?\d+", r"less than \$?\d+", r"below \$?\d+", r"between \$?\d+ and \$?\d+",
r"不袅过.*ε…ƒ", r"低于.*ε…ƒ", r"ε°‘δΊŽ.*ε…ƒ", # Chinese
r"kurang dari", r"di bawah", # Malay/Indonesian
r"ΰΈ™ΰΉ‰ΰΈ­ΰΈ’ΰΈΰΈ§ΰΉˆΰΈ²", r"ΰΈ•ΰΉˆΰΈ³ΰΈΰΈ§ΰΉˆΰΈ²" # Thai
]
for pattern in comparison_patterns:
if re.search(pattern, query_lower):
criteria_count += 2
# Complex query if multiple keywords or comparison patterns found
return criteria_count >= 2
def _is_translation_query(self, query: str) -> bool:
"""Check if query is primarily for translation"""
translation_keywords = [
"translate", "translation", "แปΰΈ₯", "ηΏ»θ―‘", "terjemah", "traduire"
]
query_lower = query.lower()
return any(keyword in query_lower for keyword in translation_keywords)
def generate_response(self, query: str, context: str = "", language: str = "English") -> str:
"""Generate response using appropriate SEA-LION model"""
# Choose model based on query complexity
if self._is_translation_query(query) or not self._is_complex_query(query):
model = self.instruct_model
use_reasoning = False
else:
model = self.reasoning_model
use_reasoning = True
# Prepare messages
system_prompt = f"""You are a helpful assistant specializing in ASEAN university admissions.
Respond in {language} unless specifically asked otherwise.
If provided with context from university documents, use that information to give accurate, specific answers.
Always cite your sources when using provided context.
For complex university search queries, provide:
1. Direct answers to the question
2. Relevant admission requirements
3. Tuition fees (if available)
4. Application deadlines (if available)
5. Source citations from the documents
Context: {context}"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": query}
]
try:
if use_reasoning:
# Use reasoning model with thinking mode
response = self.client.chat.completions.create(
model=model,
messages=messages,
max_tokens=2000,
temperature=0.1,
extra_body={"thinking_mode": True}
)
else:
# Use instruct model for simpler queries
response = self.client.chat.completions.create(
model=model,
messages=messages,
max_tokens=1500,
temperature=0.3
)
# Strip out reasoning steps from the response
response_text = response.choices[0].message.content
if "</think>" in response_text:
response_text = response_text.split("</think>")[-1].strip()
return response_text
except Exception as e:
print(f"Error with SEA-LION model: {str(e)}")
return f"I apologize, but I encountered an error processing your query. Please try rephrasing your question. Error: {str(e)}"
def extract_metadata(self, document_text: str) -> Dict[str, str]:
"""Extract metadata from document text using LLM"""
system_prompt = """You are an expert at extracting metadata from university documents.
Analyze the provided document text and extract the following information:
1. University name (full official name)
2. Country (where the university is located)
3. Document type (choose from: admission_requirements, tuition_fees, program_information, scholarship_info, application_deadlines, general_info)
4. Language (choose from: English, Chinese, Malay, Thai, Indonesian, Vietnamese, Filipino)
Return your response as a JSON object with these exact keys:
{
"university_name": "extracted university name or \'Unknown\' if not found",
"country": "extracted country or \'Unknown\' if not found",
"document_type": "most appropriate document type from the list above",
"language": "detected language of the document"
}
Guidelines:
- For university_name: Look for official university names, avoid abbreviations when possible
- For country: Look for country names, city names that indicate country, or domain extensions
- For document_type: Analyze the content to determine what type of information it contains
- For language: Determine the primary language of the document.
- If information is unclear, use "Unknown" for university_name and country
- Always choose one of the specified document_type options and language options
"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Extract metadata from this document text:\n\n{document_text}"}
]
try:
response = self.client.chat.completions.create(
model=self.instruct_model,
messages=messages,
max_tokens=500,
temperature=0.1
)
response_text = response.choices[0].message.content.strip()
print("--- DEBUG: LLM Metadata Extraction Details ---")
print(f"**Input Text for LLM (first 2 pages):**\n```\n{document_text[:1000]}...\n```") # Show first 1000 chars of input
print(f"**Raw LLM Response:**\n```json\n{response_text}\n```")
json_match = re.search(r'\{.*?\}', response_text, re.DOTALL)
if json_match:
json_str = json_match.group(0)
try:
metadata = json.loads(json_str)
print(f"**Parsed JSON Metadata:**\n```json\n{json.dumps(metadata, indent=2)}\n```")
required_keys = ["university_name", "country", "document_type", "language"]
if all(key in metadata for key in required_keys):
print("DEBUG: Successfully extracted and parsed metadata from LLM.")
return metadata
else:
print("DEBUG: LLM response missing required keys, attempting fallback or using defaults.")
return self._get_default_metadata()
except json.JSONDecodeError as e:
print(f"DEBUG: JSON Parsing Failed: {e}")
print(f"DEBUG: Attempting fallback text extraction from raw response.")
return self._extract_from_text_response(response_text)
else:
print("DEBUG: No JSON object found in LLM response.")
return self._extract_from_text_response(response_text)
except Exception as e:
print(f"DEBUG: Error during LLM Metadata Extraction: {str(e)}")
return self._get_default_metadata()
def _extract_from_text_response(self, response_text: str) -> Dict[str, str]:
"""Fallback method to extract metadata from non-JSON LLM response"""
metadata = self._get_default_metadata()
lines = response_text.split("\n")
for line in lines:
line = line.strip()
if "university" in line.lower() and ":" in line:
value = line.split(":", 1)[1].strip().strip('",')
metadata["university_name"] = value
elif "country" in line.lower() and ":" in line:
value = line.split(":", 1)[1].strip().strip('",')
metadata["country"] = value
elif "document_type" in line.lower() and ":" in line:
value = line.split(":", 1)[1].strip().strip('",')
metadata["document_type"] = value
elif "language" in line.lower() and ":" in line:
value = line.split(":", 1)[1].strip().strip('",')
metadata["language"] = value
print(f"DEBUG: Fallback text extraction result: {metadata}")
return metadata
def _get_default_metadata(self) -> Dict[str, str]:
"""Return default metadata when extraction fails"""
return {
"university_name": "Unknown",
"country": "Unknown",
"document_type": "general_info",
"language": "Unknown"
}
def classify_query_type(query: str) -> str:
"""Public function to classify query type for UI display"""
# Create a temporary SEALionLLM instance just for classification
temp_llm = SEALionLLM()
if temp_llm._is_translation_query(query) or not temp_llm._is_complex_query(query):
return "simple"
else:
return "complex"
class DocumentIngestion:
def __init__(self):
# Initialize SEA-LION LLM for metadata extraction
self.sea_lion_llm = SEALionLLM()
# Use BGE embeddings by default for better performance
try:
self.embeddings = AlternativeEmbeddings()
self.embedding_type = "BGE-small-en"
if not self.embeddings.model:
raise Exception("BGE model not available")
except Exception:
# Fallback to OpenAI if BGE not available
openai_key = os.getenv("OPENAI_API_KEY")
if openai_key and openai_key != "placeholder_for_embeddings" and openai_key != "your_openai_api_key_here":
try:
self.embeddings = OpenAIEmbeddings()
self.embedding_type = "OpenAI"
except Exception as e:
print("Both BGE and OpenAI embeddings failed. Please check your setup.")
raise e
else:
print("No embedding model available. Please install sentence-transformers or provide OpenAI API key.")
raise Exception("No embedding model available")
self.text_splitter = SemanticChunker(
embeddings_model=self.embeddings,
chunk_size=4, # 4 sentences per base chunk
overlap=1, # 1 sentence overlap
similarity_threshold=0.75, # Semantic similarity threshold
min_chunk_size=150, # Minimum 150 characters
max_chunk_size=1500, # Maximum 1500 characters
debug=True # Show statistics in Streamlit
)
# st.info(f"🧠 Using semantic chunking with {self.embedding_type} embeddings") # Commented out as it\'s a Streamlit call
self.persist_directory = os.getenv("CHROMA_PERSIST_DIRECTORY", "./chroma_db")
os.makedirs(self.persist_directory, exist_ok=True)
def extract_text_from_pdf(self, pdf_file_path) -> List[str]:
"""Extract text from PDF file path with multiple fallback methods."""
try:
# Method 1: Try with PyPDF2 (handles most PDFs including encrypted ones with PyCryptodome)
with open(pdf_file_path, 'rb') as pdf_file:
pdf_reader = PyPDF2.PdfReader(pdf_file)
# Check if PDF is encrypted
if pdf_reader.is_encrypted:
# Try to decrypt with empty password (common for protected but not password-protected PDFs)
try:
pdf_reader.decrypt("")
except Exception:
print(f"PDF {os.path.basename(pdf_file_path)} is password-protected. Please provide an unprotected version.")
return [] # Return empty list for password-protected PDFs
text_per_page = []
for page_num, page in enumerate(pdf_reader.pages):
try:
page_text = page.extract_text()
text_per_page.append(page_text)
except Exception as e:
print(f"Could not extract text from page {page_num + 1} of {os.path.basename(pdf_file_path)}: {str(e)}")
text_per_page.append("") # Append empty string for failed pages
if any(text.strip() for text in text_per_page):
return text_per_page
else:
print(f"No extractable text found in {os.path.basename(pdf_file_path)}. This might be a scanned PDF or image-based document.")
return []
except Exception as e:
error_msg = str(e)
if "PyCryptodome" in error_msg:
print(f"Encryption error with {os.path.basename(pdf_file_path)}: {error_msg}")
print("πŸ’‘ The PDF uses encryption. PyCryptodome has been installed to handle this.")
elif "password" in error_msg.lower():
print(f"Password-protected PDF: {os.path.basename(pdf_file_path)}")
print("πŸ’‘ Please provide an unprotected version of this PDF.")
else:
print(f"Error extracting text from {os.path.basename(pdf_file_path)}: {error_msg}")
return []
def process_documents(self, pdf_file_paths) -> List[Document]:
"""Process PDF file paths and convert to documents with automatic metadata extraction."""
documents = []
processed_count = 0
failed_count = 0
print(f"πŸ“„ Processing {len(pdf_file_paths)} document(s) with automatic metadata detection...") # Changed to print
for pdf_file_path in pdf_file_paths:
if pdf_file_path.endswith('.pdf'):
filename = os.path.basename(pdf_file_path)
print(f"πŸ” Extracting text from: **{filename}**") # Changed to print
# Extract text per page
text_per_page = self.extract_text_from_pdf(pdf_file_path)
print(f"DEBUG: Extracted {len(text_per_page)} pages from {filename}")
if text_per_page:
# Combine first two pages for metadata extraction
text_for_metadata = "\n".join(text_per_page[:2])
print(f"DEBUG: Text for metadata extraction (first 500 chars): {text_for_metadata[:500]}")
# Extract metadata using LLM
print(f"πŸ€– Detecting metadata for: **{filename}**") # Changed to print
extracted_metadata = self.sea_lion_llm.extract_metadata(text_for_metadata)
# Create metadata
metadata = {
"source": filename,
"university": extracted_metadata.get("university_name", "Unknown"),
"country": extracted_metadata.get("country", "Unknown"),
"document_type": extracted_metadata.get("document_type", "general_info"),
"language": extracted_metadata.get("language", "Unknown"), # Added language
"upload_timestamp": datetime.now().isoformat(),
"file_id": str(uuid.uuid4())
}
# Create document
doc = Document(
page_content="\n".join(text_per_page), # Use all pages for document content
metadata=metadata
)
documents.append(doc)
processed_count += 1
print(f"βœ… Successfully processed: **{filename}** ({len(doc.page_content)} characters)") # Changed to print
else:
failed_count += 1
print(f"⚠️ Could not extract text from **{filename}**") # Changed to print
else:
failed_count += 1
filename = os.path.basename(pdf_file_path)
print(f"❌ Unsupported file type for {filename} (expected .pdf)") # Changed to print
# Summary
if processed_count > 0:
print(f"πŸŽ‰ Successfully processed **{processed_count}** document(s)") # Changed to print
if failed_count > 0:
print(f"⚠️ Failed to process **{failed_count}** document(s)") # Changed to print
return documents
def create_vector_store(self, documents: List[Document]) -> Chroma:
"""Create and persist vector store from documents."""
if not documents:
print("No documents to process") # Changed to print
return None
# Split documents into chunks
texts = self.text_splitter.split_documents(documents)
# Create vector store
vectorstore = Chroma.from_documents(
documents=texts,
embedding=self.embeddings,
persist_directory=self.persist_directory
)
return vectorstore
def load_existing_vectorstore(self) -> Optional[Chroma]:
"""Load existing vector store if it exists."""
try:
vectorstore = Chroma(
persist_directory=self.persist_directory,
embedding_function=self.embeddings
)
return vectorstore
except Exception as e:
print(f"Could not load existing vector store: {str(e)}") # Changed to print
return None
class RAGSystem:
def __init__(self):
# Initialize embeddings - try BGE first, fallback to OpenAI
try:
self.embeddings = AlternativeEmbeddings()
if not self.embeddings.model:
# Fallback to OpenAI if BGE not available
self.embeddings = OpenAIEmbeddings()
except Exception:
# If both fail, use OpenAI as last resort
self.embeddings = OpenAIEmbeddings()
self.sea_lion_llm = SEALionLLM()
self.persist_directory = os.getenv("CHROMA_PERSIST_DIRECTORY", "./chroma_db")
def get_vectorstore(self) -> Optional[Chroma]:
"""Get the vector store."""
try:
vectorstore = Chroma(
persist_directory=self.persist_directory,
embedding_function=self.embeddings
)
return vectorstore
except Exception as e:
print(f"Error loading vector store: {str(e)}")
return None
def query(self, question: str, language: str = "English") -> Dict[str, Any]:
"""Query the RAG system using SEA-LION models."""
vectorstore = self.get_vectorstore()
# if not vectorstore:
# return {
# "answer": "No documents have been ingested yet. Please upload some PDF documents first.",
# "source_documents": [],
# "query_id": None
# }
try:
# Retrieve relevant documents
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
relevant_docs = retriever.get_relevant_documents(question)
# Prepare context from retrieved documents
context_parts = []
for i, doc in enumerate(relevant_docs, 1):
source_info = doc.metadata.get('source', 'Unknown')
university = doc.metadata.get('university', 'Unknown')
country = doc.metadata.get('country', 'Unknown')
context_parts.append(f"""
Document {i} (Source: {source_info}, University: {university}, Country: {country}):
{doc.page_content[:500]}...
""")
context = "\n".join(context_parts)
# Generate response using SEA-LION model
answer = self.sea_lion_llm.generate_response(
query=question,
context=context,
language=language
)
# Generate query ID for sharing
query_id = str(uuid.uuid4())
return {
"answer": answer,
"source_documents": relevant_docs,
"query_id": query_id,
"original_question": question,
"language": language,
"model_used": "SEA-LION" + (" Reasoning" if self.sea_lion_llm._is_complex_query(question) else " Instruct")
}
except Exception as e:
print(f"Error querying system: {str(e)}")
return {
"answer": f"Error processing your question: {str(e)}",
"source_documents": [],
"query_id": None
}
def save_query_result(query_result: Dict[str, Any]):
"""Save query result for sharing."""
if query_result.get("query_id"):
results_dir = "query_results"
os.makedirs(results_dir, exist_ok=True)
result_file = f"{results_dir}/{query_result['query_id']}.json"
# Prepare data for saving (remove non-serializable objects)
save_data = {
"query_id": query_result["query_id"],
"question": query_result.get("original_question", ""),
"answer": query_result["answer"],
"language": query_result.get("language", "English"),
"timestamp": datetime.now().isoformat(),
"sources": [
{
"source": doc.metadata.get("source", "Unknown"),
"university": doc.metadata.get("university", "Unknown"),
"country": doc.metadata.get("country", "Unknown"),
"content_preview": doc.page_content[:200] + "..." if len(doc.page_content) > 200 else doc.page_content
}
for doc in query_result.get("source_documents", [])
]
}
try:
with open(result_file, 'w', encoding='utf-8') as f:
json.dump(save_data, f, indent=2, ensure_ascii=False)
return True
except Exception as e:
print(f"Error saving query result: {str(e)}")
return False
return False
def load_shared_query(query_id: str) -> Optional[Dict[str, Any]]:
"""Load a shared query result."""
result_file = f"query_results/{query_id}.json"
if os.path.exists(result_file):
try:
with open(result_file, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
print(f"Error loading shared query: {str(e)}")
return None