import os import sys import time import base64 import warnings warnings.filterwarnings('ignore') user_query = sys.argv[1] pdf_path = sys.argv[2] vector_db = sys.argv[3] embeddings_flag = sys.argv[4] f = open("demofile.txt", "a") f.write(user_query) f.write(pdf_path) f.write(vector_db) f.write(embeddings_flag) if embeddings_flag == "yes": from langchain_community.document_loaders import UnstructuredPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.embeddings import HuggingFaceBgeEmbeddings from langchain_community.vectorstores import Qdrant class EmbeddingsManager: def __init__(self, model_name, device, encode_kwargs, qdrant_url, collection_name): """ Initializes the EmbeddingsManager with the specified model and Qdrant settings. Args: model_name (str): The HuggingFace model name for embeddings. device (str): The device to run the model on ('cpu' or 'cuda'). encode_kwargs (dict): Additional keyword arguments for encoding. qdrant_url (str): The URL for the Qdrant instance. collection_name (str): The name of the Qdrant collection. """ self.model_name = model_name self.device = device self.encode_kwargs = encode_kwargs self.qdrant_url = qdrant_url self.collection_name = collection_name self.embeddings = HuggingFaceBgeEmbeddings( model_name=self.model_name, model_kwargs={"device": self.device}, encode_kwargs=self.encode_kwargs, ) def create_embeddings(self, pdf_path: str): """ Processes the PDF, creates embeddings, and stores them in Qdrant. Args: pdf_path (str): The file path to the PDF document. Returns: str: Success message upon completion. """ if not os.path.exists(pdf_path): raise FileNotFoundError(f"The file {pdf_path} does not exist.") # Load and preprocess the document loader = UnstructuredPDFLoader(pdf_path) docs = loader.load() if not docs: raise ValueError("No documents were loaded from the PDF.") text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=250 ) splits = text_splitter.split_documents(docs) if not splits: raise ValueError("No text chunks were created from the documents.") # Create and store embeddings in Qdrant try: qdrant = Qdrant.from_documents( splits, self.embeddings, url=self.qdrant_url, prefer_grpc=False, collection_name=self.collection_name, ) except Exception as e: raise ConnectionError(f"Failed to connect to Qdrant: {e}") return "✅ Vector DB Successfully Created and Stored in Qdrant!" from langchain_community.embeddings import HuggingFaceBgeEmbeddings from langchain_community.vectorstores import Qdrant from langchain_ollama import ChatOllama from qdrant_client import QdrantClient from langchain import PromptTemplate from langchain.chains import RetrievalQA class ChatbotManager: def __init__(self, model_name, device, encode_kwargs, llm_model, llm_temperature, qdrant_url, collection_name): """ Initializes the ChatbotManager with embedding models, LLM, and vector store. Args: model_name (str): The HuggingFace model name for embeddings. device (str): The device to run the model on ('cpu' or 'cuda'). encode_kwargs (dict): Additional keyword arguments for encoding. llm_model (str): The local LLM model name for ChatOllama. llm_temperature (float): Temperature setting for the LLM. qdrant_url (str): The URL for the Qdrant instance. collection_name (str): The name of the Qdrant collection. """ self.model_name = model_name self.device = device self.encode_kwargs = encode_kwargs self.llm_model = llm_model self.llm_temperature = llm_temperature self.qdrant_url = qdrant_url self.collection_name = collection_name # Initialize Embeddings self.embeddings = HuggingFaceBgeEmbeddings( model_name=self.model_name, model_kwargs={"device": self.device}, encode_kwargs=self.encode_kwargs, ) # Initialize Local LLM self.llm = ChatOllama( model=self.llm_model, temperature=self.llm_temperature, max_tokens=1024, # Increase output token limit if responses are cut off top_p=0.1, # Nucleus sampling for balanced randomness frequency_penalty=0, # Avoid penalizing common words presence_penalty=0 # Add other parameters if needed ) # Define the prompt template self.prompt_template = """Use the following pieces of information to answer the user's question. 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} Only return the helpful answer. Answer must be detailed and well explained. Helpful answer: """ # Initialize Qdrant client self.client = QdrantClient( url=self.qdrant_url, prefer_grpc=False ) # Initialize the Qdrant vector store self.db = Qdrant( client=self.client, embeddings=self.embeddings, collection_name=self.collection_name ) # Initialize the prompt self.prompt = PromptTemplate( template=self.prompt_template, input_variables=['context', 'question'] ) # Initialize the retriever self.retriever = self.db.as_retriever(search_kwargs={"k": 1}) # Define chain type kwargs self.chain_type_kwargs = {"prompt": self.prompt} # Initialize the RetrievalQA chain with return_source_documents=False self.qa = RetrievalQA.from_chain_type( llm=self.llm, chain_type="stuff", retriever=self.retriever, return_source_documents=False, # Set to False to return only 'result' chain_type_kwargs=self.chain_type_kwargs, verbose=False ) def get_response(self, query: str) -> str: """ Processes the user's query and returns the chatbot's response. Args: query (str): The user's input question. Returns: str: The chatbot's response. """ try: response = self.qa.run(query) return response # 'response' is now a string containing only the 'result' except Exception as e: return "⚠️ Sorry, I couldn't process your request at the moment." if embeddings_flag == "yes": embeddings_manager = EmbeddingsManager( model_name="BAAI/bge-small-en", device="cpu", encode_kwargs={"normalize_embeddings": True}, qdrant_url="http://localhost:6333", collection_name=vector_db ) result = embeddings_manager.create_embeddings(pdf_path) chatbot_manager = ChatbotManager( model_name="BAAI/bge-small-en", device="cpu", encode_kwargs={"normalize_embeddings": True}, llm_model="llama3.2:3b", llm_temperature=0.2, qdrant_url="http://localhost:6333", collection_name=vector_db ) answer = chatbot_manager.get_response(user_query) f.write(answer) #f.write(result) f.close() print(answer)