updated

app.py

@@ -1,77 +1,59 @@
 import os
 from getpass import getpass
-import csv
-from langchain_core.documents import Document
 from langchain_text_splitters import RecursiveCharacterTextSplitter
-#from langchain.schema import Document
 from langchain_huggingface import HuggingFaceEmbeddings
 import torch
 from langchain_huggingface import HuggingFaceEndpoint
-from langchain_community.cache import InMemoryCache
-from langchain.globals import set_llm_cache
+from langchain_core.caches import InMemoryCache
+from langchain_core.globals import set_llm_cache
 from langchain_chroma import Chroma
 from langchain.chains import RetrievalQA
-import numpy as np
 import gradio
-import sqlite3
-from dotenv import load_dotenv
-
-# Load environment variables
-load_dotenv()
-
-#hfapi_key = getpass("Enter you HuggingFace access token:")
-hfapi_key = os.getenv("Mytoken")
-
-if not hfapi_key:
-    raise ValueError("HUGGINGFACE_API_KEY not found in environment variables")
+import PyPDF2
+import json
+import re
+import time
+import threading
+from langchain_core.runnables import RunnableConfig, RunnablePassthrough
+from langchain_core.output_parsers import StrOutputParser
+from langchain_core.prompts import PromptTemplate
+from langchain_core.runnables import RunnableLambda
 
+hfapi_key = getpass("Enter you HuggingFace access token:")
 os.environ["HF_TOKEN"] = hfapi_key
 os.environ["HUGGINGFACEHUB_API_TOKEN"] = hfapi_key
 
-set_llm_cache(InMemoryCache())
+set_llm_cache(InMemoryCache())  # Set cache globally
 
 persist_directory = 'docs/chroma/'
+pdf_path = 'AIML.pdf'
 
-####################################
-def load_file_as_JSON():
-    print("$$$$$ ENTER INTO load_file_as_JSON $$$$$")
-    rows = []
-    with open("mini-llama-articles.csv", mode="r", encoding="utf-8") as file:
-        csv_reader = csv.reader(file)
-        for idx, row in enumerate(csv_reader):
-            if idx == 0:
-                continue
-                # Skip header row
-            rows.append(row)
-
-    print("@@@@@@ EXIT FROM load_file_as_JSON @@@@@")
-    return rows
 ####################################
 def get_documents():
     print("$$$$$ ENTER INTO get_documents $$$$$")
-    documents = [
-        Document(
-            page_content=row[1], metadata={"title": row[0], "url": row[2], "source_name": row[3]}
-        )
-        for row in load_file_as_JSON()
-    ]
-    print("documents lenght is ", len(documents))
-    print("first entry from documents ", documents[0])
-    print("document metadata ", documents[0].metadata)
+    
+    with open(pdf_path, 'rb') as file:
+        # Create a PDF reader object
+        pdf_reader = PyPDF2.PdfReader(file)
+        
+        # Extract text from all pages
+        full_text = ""
+        for page in pdf_reader.pages:
+            full_text += page.extract_text() + "\n"
+
     print("@@@@@@ EXIT FROM get_documents @@@@@")
-    return documents
+    return full_text
 ####################################
-def getDocSplitter():
+def getTextSplits():
     print("$$$$$ ENTER INTO getDocSplitter $$$$$")
     text_splitter = RecursiveCharacterTextSplitter(
         chunk_size = 512,
         chunk_overlap = 128
     )
-    splits = text_splitter.split_documents(get_documents())
-    print("Split length ", len(splits))
-    print("Page content ", splits[0].page_content)
+    texts = text_splitter.split_text(get_documents())
+    #print("Page content ", texts)
     print("@@@@@@ EXIT FROM getDocSplitter @@@@@")
-    return splits
+    return texts
 ####################################
 def getEmbeddings():
     print("$$$$$ ENTER INTO getEmbeddings $$$$$")
@@ -90,133 +72,216 @@ def getEmbeddings():
         encode_kwargs=encode_kwargs # Pass the encoding options
     )
     
-    print("Embedding ", embedding)
     print("@@@@@@ EXIT FROM getEmbeddings @@@@@")
     return embedding
 ####################################
 def getLLM():
     print("$$$$$ ENTER INTO getLLM $$$$$")
+
+    model_kwargs = {
+        'device': "cuda" if torch.cuda.is_available() else "cpu",
+        'stream': True  # Ensure streaming is enabled
+    }
+
     llm = HuggingFaceEndpoint(
         repo_id="HuggingFaceH4/zephyr-7b-beta",
-        #repo_id="chsubhasis/ai-tutor-towardsai",
         task="text-generation",
-        max_new_tokens = 512,   
-        top_k = 10,
-        temperature = 0.1,
-        repetition_penalty = 1.03,
+        max_new_tokens= 512,
+        do_sample= True,
+        temperature = 0.7,
+        repetition_penalty= 1.2,
+        top_k = 10
+        #model_kwargs=model_kwargs  # Pass the model configuration options
     )
-    print("llm ", llm)
-    print("Who is the CEO of Apple? ", llm.invoke("Who is the CEO of Apple?")) #test
     print("@@@@@@ EXIT FROM getLLM @@@@@")
     return llm
 ####################################
 def is_chroma_db_present(directory: str):
-    """
-    Check if the directory exists and contains any files.
-    """
+
+    #Check if the directory exists and contains any files.
     return os.path.exists(directory) and len(os.listdir(directory)) > 0
 ####################################
-def getRetiriver():
+def getRetiriver(query, metadata_filter:None):
     print("$$$$$ ENTER INTO getRetiriver $$$$$")
+    
+    # Classify query
+    query_type = classify_query(query)
+    print("Query classification", query_type)
+
+    k_default = 2
+    fetch_k_default = 5
+    search_type_default = "mmr"
+  
+    # Routing logic
+    if query_type == 'concept':
+        # For conceptual queries, prioritize comprehensive context
+        k_default = 5
+        fetch_k_default = 10
+        search_type_default = "mmr"
+    elif query_type == 'example':
+        # For example queries, focus on more specific, relevant contexts
+        search_type_default = "similarity"
+    elif query_type == 'code':
+        # For code-related queries, use a more targeted retrieval
+        search_type_default = "similarity"
+
     if is_chroma_db_present(persist_directory):
         print(f"Chroma vector DB found in '{persist_directory}' and will be loaded.")
         # Load vector store from the local directory
-        #vectordb = Chroma(persist_directory=persist_directory)
         vectordb = Chroma(
             persist_directory=persist_directory, 
             embedding_function=getEmbeddings(),
             collection_name="ai_tutor")
     else:
-        vectordb = Chroma.from_documents(
+        vectordb = Chroma.from_texts(
             collection_name="ai_tutor",
-            documents=getDocSplitter(), # splits we created earlier
+            texts=getTextSplits(),
             embedding=getEmbeddings(),
             persist_directory=persist_directory, # save the directory
         )
-    print("vectordb collection count ", vectordb._collection.count())
     
-    docs = vectordb.search("What is Artificial Intelligence", search_type="mmr", k=5)
-    for i in range(len(docs)):
-     print(docs[i].page_content)
+    print("metadata_filter", metadata_filter)
+    if(metadata_filter):
+        metadata_filter_dict = {
+        "result": metadata_filter  # ChromaDB will perform a substring search
+        }
+        print("@@@@@@ EXIT FROM getRetiriver with metadata_filter @@@@@")
+
+        if search_type_default == "similarity":
+            return vectordb.as_retriever(search_type=search_type_default, search_kwargs={"k": k_default, "filter": metadata_filter_dict})
+        
+        return vectordb.as_retriever(search_type=search_type_default, search_kwargs={"k": k_default, "fetch_k":fetch_k_default, "filter": metadata_filter_dict})
+
+    print("@@@@@@ EXIT FROM getRetiriver without metadata_filter @@@@@")
+    if search_type_default == "similarity":
+        return vectordb.as_retriever(search_type=search_type_default, search_kwargs={"k": k_default})
     
-    metadata_filter = {
-        "result": "llama"  # ChromaDB will perform a substring search
-    }
+    return vectordb.as_retriever(search_type=search_type_default, search_kwargs={"k": k_default, "fetch_k":fetch_k_default})
+####################################
+def classify_query(query):
+    """
+    Classify the type of query to determine routing strategy.
+    
+    Query Types:
+    - 'concept': Theoretical or conceptual questions
+    - 'example': Requests for practical examples
+    - 'code': Coding or implementation-related queries
+    - 'general': Default catch-all category
+    """
+    query = query.lower()
     
-    retriever = vectordb.as_retriever(search_type="mmr", search_kwargs={"k": 3, "fetch_k":5, "filter": metadata_filter})
-    print("retriever ", retriever)
-    print("@@@@@@ EXIT FROM getRetiriver @@@@@")
-    return retriever
+    # Concept detection patterns
+    concept_patterns = [
+        r'what is',
+        r'define',
+        r'explain',
+        r'describe',
+        r'theory of',
+        r'concept of'
+    ]
+    
+    # Example detection patterns
+    example_patterns = [
+        r'give an example',
+        r'show me an example',
+        r'demonstrate',
+        r'illustrate'
+    ]
+    
+    # Code-related detection patterns
+    code_patterns = [
+        r'how to implement',
+        r'code for',
+        r'python code',
+        r'algorithm implementation',
+        r'write a program'
+    ]
+    
+    # Check patterns
+    for pattern in concept_patterns:
+        if re.search(pattern, query):
+            return 'concept'
+    
+    for pattern in example_patterns:
+        if re.search(pattern, query):
+            return 'example'
+    
+    for pattern in code_patterns:
+        if re.search(pattern, query):
+            return 'code'
+    
+    return 'general'
 ####################################
-def get_rag_response(query):
-  print("$$$$$ ENTER INTO get_rag_response $$$$$")  
-  qa_chain = RetrievalQA.from_chain_type(
-    llm=getLLM(),
-    chain_type="stuff",
-    retriever=getRetiriver(),
-    return_source_documents=True
-  )
-  
-  #RAG Evaluation
-  # Sample dataset of questions and expected answers
-  dataset = [
-    {"question": "Who is the CEO of Meta?", "expected_answer": "Mark Zuckerberg"},
-    {"question": "Who is the CEO of Apple?", "expected_answer": "Tiiiiiim Coooooook"},
-  ]
+def get_rag_response(query, metadata_filter=None):
+    print("$$$$$ ENTER INTO get_rag_response $$$$$")
   
-  hit_rate, mrr = evaluate_rag(qa_chain, dataset)
-  print(f"Hit Rate: {hit_rate:.2f}, Mean Reciprocal Rank (MRR): {mrr:.2f}")
-  
-  result = qa_chain({"query": query})
-  print("Result ",result)
-  print("@@@@@@ EXIT FROM get_rag_response @@@@@")
-  return result["result"]
+    # Create the retriever
+    retriever = getRetiriver(query, metadata_filter)
+
+    # Get the LLM
+    llm = getLLM()
+
+    # Create a prompt template
+    template = """Use the following pieces of context to answer the question at the end. 
+    If you don't know the answer, just say that you don't know, don't try to make up an answer.
+
+    Context: {context}
+
+    Question: {question}
+
+    Helpful Answer:"""
+    
+    prompt = PromptTemplate.from_template(template)
+
+    # Function to prepare input for the chain
+    def prepare_inputs(inputs):
+        retrieved_docs = retriever.invoke(inputs["question"])
+        context = format_docs(retrieved_docs)
+        return {
+            "context": context,
+            "question": inputs["question"]
+        }
+
+    # Construct the RAG chain with streaming
+    rag_chain = (
+        RunnablePassthrough()
+        | RunnableLambda(prepare_inputs)
+        | prompt
+        | llm
+        | StrOutputParser()
+    )
+
+    # Stream the response
+    full_response = ""
+    for chunk in rag_chain.stream({"question": query}):
+        full_response += chunk
+        # Add a small delay to create a streaming effect
+        time.sleep(0.05)  # 50 milliseconds between chunk updates
+        yield full_response
+
 ####################################
-def evaluate_rag(qa, dataset):
-    print("$$$$$ ENTER INTO evaluate_rag $$$$$")
-    hits = 0
-    reciprocal_ranks = []
-
-    for entry in dataset:
-        question = entry["question"]
-        expected_answer = entry["expected_answer"]
-        
-        # Get the answer from the RAG system
-        response = qa({"query": question}) 
-        answer = response["result"] 
-        
-        # Check if the answer matches the expected answer
-        if expected_answer.lower() in answer.lower():
-            hits += 1
-            reciprocal_ranks.append(1)  # Hit at rank 1
-        else:
-            reciprocal_ranks.append(0)
-
-    # Calculate Hit Rate and MRR
-    hit_rate = hits / len(dataset)
-    mrr = np.mean(reciprocal_ranks)
-
-    print("@@@@@@ EXIT FROM evaluate_rag @@@@@")
-    return hit_rate, mrr
+# Utility function to format documents
+def format_docs(docs):
+    return "\n\n".join(doc.page_content for doc in docs)
 ####################################
-def launch_ui():
-    print("$$$$$ ENTER INTO launch_ui $$$$$")
-    # Input from user
-    in_question = gradio.Textbox(lines=10, placeholder=None, value="query", label='Enter your query')
+# Input from user
+in_question = gradio.Textbox(lines=10, placeholder=None, value="What are Artificial Intelligence and Machine Learning?", label='Ask a question to your AI Tutor')
 
-    # Output prediction
-    out_response = gradio.Textbox(type="text", label='RAG Response')
+# Optional metadata filter input
+in_metadata_filter = gradio.Textbox(lines=2, placeholder=None, label='Optionally add a filter word')
 
-    # Gradio interface to generate UI
-    iface = gradio.Interface(fn = get_rag_response,
-                            inputs = [in_question],
-                            outputs = [out_response],
-                            title = "RAG Response",
-                            description = "Write the query and get the response from the RAG system",
-                            allow_flagging = 'never')
+# Output prediction
+out_response = gradio.Textbox(label='Response', interactive=False, show_copy_button=True)
 
-    iface.launch(share = True)
+# Gradio interface to generate UI
+iface = gradio.Interface(
+    fn = get_rag_response,
+    inputs=[in_question, in_metadata_filter],
+    outputs=out_response,
+    title="Your AI Tutor",
+    description="Ask a question, optionally add metadata filters.",
+    allow_flagging='never',
+    stream_every=0.5
+    )
 
-####################################
-if __name__ == "__main__":
-    launch_ui()
+iface.launch(share = True)