updated

.gitignore

@@ -0,0 +1,149 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+pip-wheel-metadata/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+.python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# Machine Learning and Speech Libraries
+# TensorFlow
+*.ckpt*
+*.pbtxt
+*.tfevents*
+# PyTorch
+*.pt
+# Keras
+*.h5
+# Scikit-learn
+*.pkl
+# Speech Recognition
+*.wav
+*.mp3
+.idea/
+logs
+images
+resources
+experiments

app.py

@@ -0,0 +1,27 @@
+
+import uvicorn
+from fastapi import FastAPI, Depends
+from fastapi.middleware.cors import CORSMiddleware
+from core import logging as logger
+from core.api.chatai import chatai_api_router
+from core.api.jwt_bearer import access_check_bearer
+from core.api.user_management_api import user_management_api_router
+from core.utils.utils import load_ini_config
+config = load_ini_config("config.ini")
+app = FastAPI(docs_url=config.get('fastapi_config', 'docs_url'), redoc_url=config.get('fastapi_config', 'redoc_url'),
+              openapi_url=config.get('fastapi_config', 'openapi_url')
+              )
+
+PROTECTED = [Depends(access_check_bearer)]
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"], 
+    allow_credentials=True,
+    allow_methods=["*"], 
+    allow_headers=["*"],  
+)
+app.include_router(user_management_api_router, prefix="/chatai")
+app.include_router(chatai_api_router, prefix="/chatai")
+if __name__ == '__main__':
+    uvicorn.run(app, port=config.get("fastapi_config", "port"), host=config.get('fastapi_config', 'host'),
+                timeout_keep_alive=300, timeout_graceful_shutdown=600)

config.ini

@@ -0,0 +1,13 @@
+
+
+
+[fastapi_config]
+host = 0.0.0.0
+port = 7860
+docs_url = /docs
+redoc_url = /redoc
+openapi_url = /openapi.json
+
+
+[oauth]
+redirect_to : https://google.com/

core/__init__.py

@@ -0,0 +1,16 @@
+
+
+import logging.config
+import yaml
+import os
+
+if os.path.exists("logs"):
+    pass
+else:
+    os.makedirs("logs")
+
+log_config_path = os.path.join(os.getcwd(), "logging_config.yaml")
+with open(log_config_path, 'r') as file:
+    config = yaml.safe_load(file.read())
+
+logging.config.dictConfig(config)

core/api/chatai.py

@@ -0,0 +1,497 @@
+import json
+import os
+import string
+import tempfile
+from urllib.parse import urlparse
+import requests
+from fastapi import UploadFile, File, Form, HTTPException
+from fastapi.requests import Request
+from fastapi.routing import APIRouter
+from supabase import create_client
+from core import logging as logger
+from core.api.user_management_api import user_management
+from core.api.user_management_api import user_management as user_management_pipeline
+from core.models.apis_models import *
+from core.pipeline.chataipipeline import ChatAIPipeline
+from core.services.supabase.user_management.token_limit import token_limit_check
+from core.utils.error_handling import create_error_response, create_success_response, raise_http_exception
+from core.utils.utils import get_ip_info, encode_to_base64, clean_text, decode_base64
+from core.services.supabase.limit.limit_check import LimitChecker
+from PyPDF2 import PdfReader
+from dotenv import load_dotenv
+load_dotenv()
+import io
+
+chatai_api_router = APIRouter(tags=["ChatAI"])
+supabase_client = create_client(os.getenv("SUPABASE_URL"), os.getenv("SUPABASE_KEY"))
+supabase_client_ = supabase_client
+ChatAI_pipeline = ChatAIPipeline()
+url_limit,pdf_limit,ocr_limit=LimitChecker(supabase_client)
+
+@chatai_api_router.post("/add_text")
+async def add_text(request: AddTextRequest):
+    logger.info(f">>>AddText API Triggered By {request.vectorstore}<<<")
+    try:
+        vectorstore, text = request.vectorstore, request.text
+        username, chat_bot_name = request.vectorstore.split("$")[1], request.vectorstore.split("$")[2]
+        cleaned_text = " ".join(text.split())
+        num_token = len(cleaned_text)
+        lim = token_limit_check(supabase_client=supabase_client, username=username, chatbot_name=chat_bot_name,
+                                len_text=num_token)
+        text = clean_text(text)
+        if lim:
+            dct = {
+                "output": {"text": text},
+                "source": "Text",
+            }
+            cleaned_text = " ".join(text.split())  # handles unnencessary spaces
+            # Count characters
+            num_token = len(cleaned_text)
+            logger.info(f"Number of token {num_token}")
+            dct = json.dumps(dct, indent=1).encode("utf-8", errors="replace")
+            file_name = user_management_pipeline.create_data_source_name(source_name="text", username=username)
+            supabase_client.storage.from_("ChatAI").upload(file=dct, path=f"{file_name}_data.json")
+
+            supa = supabase_client.table("ChatAI_ChatbotDataSources").insert(
+                {"username": username, "chatbotName": chat_bot_name, "dataSourceName": file_name,
+                 "numTokens": num_token, "sourceEndpoint": "/add_text",
+                 "sourceContentURL": os.path.join(os.environ["SUPABASE_PUBLIC_BASE_URL"],
+                                                  f"{file_name}_data.json")}).execute()
+
+            response = create_success_response(200, {"message": "Successfully added the text."})
+            logger.info(f">>>Text added successfully for {request.vectorstore}.<<<")
+
+            return response
+        else:
+            response = create_error_response(400,
+                                             "Exceeding limits, please try with a smaller chunks of information or subscribe to our premium plan.")
+            return response
+
+    except Exception as e:
+        logger.error(f">>>Error in add_text: {e} for {request.vectorstore}.<<<")
+        raise_http_exception(500, "Internal Server Error")
+
+
+@chatai_api_router.post("/answer_query")
+async def answer_query(request: AnswerQueryRequest, req: Request):
+    logger.info(f">>>answer_query API Triggered By {request.vectorstore}<<<")
+    try:
+        username, chatbot_name = request.vectorstore.split("$")[1], request.vectorstore.split("$")[2]
+        ip_address = req.client.host
+        city = get_ip_info(ip_address)
+        output, followup_questions, source = ChatAI_pipeline.answer_query_(query=request.query,
+                                                                              vectorstore=request.vectorstore,
+                                                                              llm_model=request.llm_model)
+        supa = supabase_client.table("ChatAI_ChatHistory").insert(
+            {"username": username, "chatbotName": chatbot_name, "llmModel": request.llm_model,
+             "question": request.query, "response": output, "IpAddress": ip_address, "ResponseTokenCount": len(output),
+             "vectorstore": request.vectorstore, "City": city}).execute()
+
+        response = create_success_response(200, data={"output": output, "follow_up_questions": followup_questions,
+                                                      "source": source})
+        logger.info(f">>>Query answered successfully for {request.vectorstore}.<<<")
+        return response
+
+    except Exception as e:
+        logger.error(f">>>Error in answer_query: {e} for {request.vectorstore}.<<<")
+        raise e
+
+
+@chatai_api_router.post("/get_links")
+async def get_links(request: GetLinksRequest):
+    logger.info(f">>>get_links API Triggered By {request.url}<<<")
+    try:
+        response = ChatAI_pipeline.get_links_(url=request.url, timeout=30)
+        response = create_success_response(200, {"urls": response, "source": urlparse(request.url).netloc})
+        logger.info(f">>>Links fetched successfully for {request.url}.<<<")
+        return response
+
+    except Exception as e:
+        logger.error(f">>>Error in get_links: {e} for {request.url}.<<<")
+        raise_http_exception(500, "Internal Server Error")
+
+
+@chatai_api_router.post("/image_pdf_text_extraction")
+async def image_pdf_text_extraction(vectorstore: str = Form(...)
+                                    , pdf: UploadFile = File(...)):
+    logger.info(f">>>image_pdf_text_extraction API Triggered By {pdf.filename}<<<")
+    try:
+        username, chatbot_name = vectorstore.split("$")[1], vectorstore.split("$")[2]
+        pdf_bytes = await pdf.read()
+        source = pdf.filename
+        pdf_reader = PdfReader(io.BytesIO(pdf_bytes))
+        doc_len = len(pdf_reader.pages)
+        if doc_len<ocr_limit:
+            response = ChatAI_pipeline.image_pdf_text_extraction_(image_pdf=pdf_bytes)
+
+            num_tokens = 0
+            try:
+                num_tokens = len(" ".join([response[x] for x in response]))
+            except (KeyError, TypeError, AttributeError):
+                pass
+            lim = token_limit_check(supabase_client=supabase_client, username=username, chatbot_name=chatbot_name,
+                                    len_text=num_tokens)
+            logger.info(f"this is the {lim}")
+            if lim:
+                dct = {
+                    "output": response,
+                    "source": source
+                }
+
+                dct = json.dumps(dct, indent=1).encode("utf-8", errors="replace")
+                file_name = user_management_pipeline.create_data_source_name(source_name=source, username=username)
+                num_tokens = 0
+                try:
+                    valid_responses = [response[x] for x in response if response[x] is not None]
+                    num_tokens = len(" ".join(valid_responses))
+                except Exception as e:
+                    num_tokens = 0
+
+                response = supabase_client.storage.from_("ChatAI").upload(file=dct, path=f"{file_name}_data.json")
+                supa = supabase_client.table("ChatAI_ChatbotDataSources").insert(
+                    {"username": username,
+                     "chatbotName": chatbot_name,
+                     "dataSourceName": file_name,
+                     "numTokens": num_tokens,
+                     "sourceEndpoint": "/image_pdf_text_extraction",
+                     "sourceContentURL": os.path.join(os.environ["SUPABASE_PUBLIC_BASE_URL"],
+                                                      f"{file_name}_data.json")}).execute()
+
+                response = create_success_response(200,
+                                                   {"source": pdf.filename, "message": "Successfully extracted the text."})
+                logger.info(f">>>Text extracted successfully for {pdf.filename}.<<<")
+                return response
+            else:
+                response = create_error_response(402,
+                                                 "Exceeding limits, please try with a smaller chunks of PDF or subscribe to our premium plan.")
+                return response
+        else:
+            response = create_error_response(402,
+                                             "Exceeding limits, please try with a PDF having less than 20 pages for pdf .")
+            return response
+    except Exception as e:
+        raise e
+
+
+@chatai_api_router.post("/text_pdf_extraction")
+async def text_pdf_extraction(vectorstore: str = Form(...)
+                              , pdf: UploadFile = File(...)):
+    logger.info(f">>>text_pdf_extraction API Triggered By {pdf.filename}<<<")
+    try:
+        username, chatbot_name = vectorstore.split("$")[1], vectorstore.split("$")[2]
+        content = await pdf.read()
+        pdf_reader = PdfReader(io.BytesIO(content))
+        doc_len = len(pdf_reader.pages)
+
+        if doc_len < pdf_limit :
+            source = pdf.filename
+            with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as temp_file:
+                temp_file.write(content)
+                temp_file_path = temp_file.name
+
+            response = ChatAI_pipeline.text_pdf_extraction_(pdf=temp_file_path)
+            numTokens = len(" ".join([response[x] for x in response]))
+            lim = token_limit_check(supabase_client=supabase_client, username=username, chatbot_name=chatbot_name,
+                                    len_text=numTokens)
+            os.remove(temp_file_path)
+            if lim:
+                dct = {
+                    "output": response,
+                    "source": source
+                }
+                numTokens = len(" ".join([response[x] for x in response]))
+                logger.info(f"Num of tokens {numTokens} text_pdf_extraction")
+                dct = json.dumps(dct, indent=1).encode("utf-8", errors="replace")
+                file_name = user_management_pipeline.create_data_source_name(source_name=source, username=username)
+                response = supabase_client.storage.from_("ChatAI").upload(file=dct, path=f"{file_name}_data.json")
+                response = (
+                    supabase_client.table("ChatAI_ChatbotDataSources")
+                    .insert({"username": username,
+                             "chatbotName": chatbot_name,
+                             "dataSourceName": file_name,
+                             "numTokens": numTokens,
+                             "sourceEndpoint": "/text_pdf_extraction",
+                             "sourceContentURL": os.path.join(os.environ["SUPABASE_PUBLIC_BASE_URL"],
+                                                              f"{file_name}_data.json")})
+                    .execute()
+                )
+                response = create_success_response(200, {"source": source, "message": "Successfully extracted the text."})
+                logger.info(f">>>Text extracted successfully for {source}.<<<")
+                return response
+            else:
+                response = create_error_response(402,
+                                                 "Exceeding limits, please try with a smaller chunks of PDF or subscribe to our premium plan.")
+                return response
+        else:
+            response = create_error_response(402,
+                                             "Exceeding limits, please try with a pdf having pages less than 200.")
+            return response
+
+    except Exception as e:
+        logger.error(f">>>Error in text_pdf_extraction: {e} for {vectorstore}.<<<")
+        raise_http_exception(500, "Internal Server Error")
+
+
+
+
+@chatai_api_router.post("/website_url_text_extraction")
+async def add_website(request: AddWebsiteRequest):
+    vectorstore, website_urls, source = request.vectorstore, request.website_urls, request.source
+
+    logger.info(f">>>website_url_text_extraction API Triggered By {request.website_urls}<<<")
+    try:
+        username, chatbot_name = vectorstore.split("$")[1], vectorstore.split("$")[2]
+        total_requested_urls=len(website_urls)
+
+        if total_requested_urls < url_limit :
+            text = ChatAI_pipeline.website_url_text_extraction_list_(urls=website_urls)
+            num_token = len(" ".join([text[x] for x in text]))
+
+            logger.info(f">>>website_url_text_extraction len{num_token}<<<")
+            lim = token_limit_check(supabase_client=supabase_client, username=username, chatbot_name=chatbot_name,
+                                    len_text=num_token)
+            if not lim:
+
+                response = create_error_response(402,
+                                                 "Exceeding limits, please try with a smaller chunks of information or subscribe to our premium plan.")
+                return response
+            else:
+                dct = {
+                    "output": text,
+                    "source": source
+                }
+
+                dct = json.dumps(dct, indent=1).encode("utf-8", errors="replace")
+                file_name = user_management_pipeline.create_data_source_name(source_name=urlparse(source).netloc,
+                                                                             username=username)
+                supabase_client.storage.from_("ChatAI").upload(file=dct, path=f"{file_name}_data.json")
+                (
+                    supabase_client.table("ChatAI_ChatbotDataSources")
+                    .insert({"username": username,
+                             "chatbotName": chatbot_name,
+                             "dataSourceName": file_name,
+                             "numTokens": num_token,
+                             "sourceEndpoint": "/fetch_text/urls",
+                             "sourceContentURL": os.path.join(os.environ["SUPABASE_PUBLIC_BASE_URL"],
+                                                              f"{file_name}_data.json")})
+                    .execute()
+                )
+                response = create_success_response(200, {"message": "Successfully fetched the website text."})
+                logger.info(f">>>Website text extracted successfully for {request.website_urls}.<<<")
+                return response
+        else:
+            response = create_error_response(402,
+                                             "Please select urls less than 50")
+            return response
+    except Exception as e:
+        logger.error(f">>>Error in website_url_text_extraction: {e} for {request.website_urls}.<<<")
+        raise HTTPException(status_code=500, detail="Internal Server Error")
+
+
+@chatai_api_router.get("/get_current_count")
+async def get_count(vectorstore: str):
+    logger.info(f">>>get_current_count API Triggered By {vectorstore}<<<")
+    try:
+        username, chatbot_name = vectorstore.split("$")[1], vectorstore.split("$")[2]
+        current_count = user_management_pipeline.get_current_count_(username)
+
+        response = create_success_response(200, {"current_count": current_count})
+        logger.info(f">>>Current count fetched successfully for {vectorstore}.<<<")
+        return response
+
+    except Exception as e:
+        logger.error(f">>>Error in get_current_count: {e} for {vectorstore}.<<<")
+        raise_http_exception(500, "Internal Server Error")
+
+
+@chatai_api_router.post("/list_chatbots")
+async def list_chatbots(request: ListChatbotsRequest):
+    logger.info(f">>>list_chatbots API Triggered By {request.username}<<<")
+    try:
+        chatbots = user_management.list_tables(username=request.username)
+        response = create_success_response(200, {"chatbots": chatbots})
+        logger.info(f">>>Chatbots listed successfully for {request.username}.<<<")
+        return response
+
+    except Exception as e:
+        logger.error(f">>>Error in list_chatbots: {e} for {request.username}.<<<")
+        raise_http_exception(500, "Internal Server Error")
+
+
+@chatai_api_router.post("/get_chat_history")
+async def chat_history(request: GetChatHistoryRequest):
+    logger.info(f">>>get_chat_history API Triggered By {request.vectorstore}<<<")
+    try:
+        _, username, chatbotName = request.vectorstore.split("$", 2)
+
+        history = supabase_client.table("ChatAI_ChatHistory").select(
+            "timestamp", "question", "response"
+        ).eq("username", username).eq("chatbotName", chatbotName).execute().data
+
+        response = create_success_response(200, {"history": history})
+        logger.info(f">>>Chat history fetched successfully for {request.vectorstore}.<<<")
+        return response
+
+
+    except IndexError:
+        logger.warning(f"Chat history not found for {request.vectorstore}")
+        return create_error_response(404, "Chat history not found for the given chatbot.")
+
+    except Exception as e:
+        logger.error(f">>>Error in get_chat_history: {e} for {request.vectorstore}.<<<")
+        raise_http_exception(500, "Internal Server Error")
+
+
+@chatai_api_router.post("/delete_chatbot")
+async def delete_chatbot(request: DeleteChatbotRequest):
+    logger.info(f">>>delete_chatbot API Triggered By {request.vectorstore}<<<")
+    try:
+        username, chatbot_name = request.vectorstore.split("$")[1], request.vectorstore.split("$")[2]
+        supabase_client.table('ChatAI_ChatbotInfo').delete().eq('user_id', username).eq('chatbotname',
+                                                                                           chatbot_name).execute()
+        all_sources = supabase_client.table("ChatAI_ChatbotDataSources").select("*").eq("username", username).eq(
+            "chatbotName", chatbot_name).execute().data
+        all_sources = [x["sourceContentURL"].split("/")[-1] for x in all_sources]
+        supabase_client.table("ChatAI_ChatbotDataSources").delete().eq("username", username).eq("chatbotName",
+                                                                                                   chatbot_name).execute()
+        for source in all_sources:
+            supabase_client.table("ChatAI_Chatbot")
+            supabase_client.storage.from_("ChatAI").remove(source)
+        user_management.delete_table(table_name=chatbot_name)
+        user_management.delete_qdrant_cluster(vectorstorename=request.vectorstore)
+        response = create_success_response(200, {"message": "Chatbot deleted successfully"})
+        logger.info(f">>>Chatbot deleted successfully for {request.vectorstore}.<<<")
+        return response
+    except Exception as e:
+        logger.error(f">>>Error in delete_chatbot: {e} for {request.vectorstore}.<<<")
+        raise_http_exception(500, "Internal Server Error")
+
+
+
+
+
+
+@chatai_api_router.get("/list_chatbot_sources")
+async def list_chatbot_sources(vectorstore: str):
+    try:
+        logger.info(f">>>list_chatbot_sources API Triggered By {vectorstore}<<<")
+
+        username, chatbot_name = vectorstore.split("$")[1], vectorstore.split("$")[2]
+        result = supabase_client.table("ChatAI_ChatbotDataSources").select("*").eq("username", username).eq(
+            "chatbotName",
+            chatbot_name).execute().data
+
+        response = create_success_response(200, {"output": result})
+        logger.info(f">>>Chatbot listed successfully for {vectorstore}.<<<")
+        return response
+
+    except Exception as e:
+        logger.error(f">>>Error in list_chatbot_sources: {e} for {vectorstore}.<<<")
+        raise_http_exception(500, "Internal Server Error")
+
+
+@chatai_api_router.get("/get_data_source")
+async def get_data_source(vectorstore: str, source_url: str):
+    try:
+        logger.info(f">>>get_data_source API Triggered By {vectorstore}<<<")
+
+        r = requests.get(source_url)
+        res = encode_to_base64(eval(r.content.decode("utf-8", errors="replace")))
+
+        response = create_success_response(200, {"output": res})
+
+        return response
+
+    except Exception as e:
+        logger.error(f">>>Error in get_data_source: {e} for {vectorstore}.<<<")
+        raise_http_exception(500, "Internal Server Error")
+
+
+@chatai_api_router.post("/delete_chatbot_source")
+async def delete_chatbot_source(request: DeleteChatbotSourceRequest):
+    vectorstore, data_source_name = request.vectorstore, request.data_source_name
+    try:
+
+        response = supabase_client.table("ChatAI_ChatbotDataSources").delete().eq("dataSourceName",
+                                                                                     data_source_name).execute()
+        response = supabase_client.storage.from_('ChatAI').remove(f"{data_source_name}_data.json")
+
+        response = create_success_response(200, {"output": f"Successfully deleted the {data_source_name} data source."})
+
+        logger.info(f">>>Data source deleted successfully for {vectorstore}.<<<")
+        return response
+
+
+    except Exception as e:
+        logger.error(f">>>Error in delete_chatbot_source: {e} for {vectorstore}.<<<")
+        raise_http_exception(500, "Internal Server Error")
+
+
+@chatai_api_router.post("/train_chatbot")
+async def train_chatbot(request: TrainChatbotRequest):
+    vectorstore, url_sources = request.vectorstore, request.urls
+    logger.info(f">>>train_chatbot API Triggered By {vectorstore}<<<")
+    try:
+        texts = []
+        sources = []
+        fileTypes = [
+            supabase_client.table("ChatAI_ChatbotDataSources").select("sourceEndpoint").eq("sourceContentURL",
+                                                                                              x).execute().data[0][
+                "sourceEndpoint"] for x in url_sources]
+        for source, fileType in zip(url_sources, fileTypes):
+            if ((fileType == "/text_pdf_extraction") | (fileType == "/image_pdf_text_extraction")):
+                logger.info(f"Source is {source}")
+                r = requests.get(source)
+                file = eval(r.content.decode("utf-8", errors="replace"))
+                content = file["output"]
+                fileSource = file["source"]
+                texts.append(".".join(
+                    [content[key] for key in content.keys()]).replace(
+                    "\n", " "))
+
+                sources.append(fileSource)
+            elif fileType == "/add_text" or fileType == "/add_qa_pair":
+                r = requests.get(source)
+                file = eval(r.content.decode("utf-8", errors="replace"))
+                content = file["output"]["text"]
+                fileSource = file["source"]
+                texts.append(content.replace("\n", " "))
+                sources.append(fileSource)
+            elif ((fileType == "/fetch_text/urls") | (fileType == "/youtube_transcript")):
+                r = requests.get(source)
+                file = eval(r.content.decode("utf-8", errors="replace"))
+                content = file["output"]
+                fileSource = file["source"]
+                texts.append(".".join(
+                    [content[key] for key in content.keys()]).replace(
+                    "\n", " "))
+                sources.append(fileSource)
+            else:
+                pass
+        texts = [(text, source) for text, source in zip(texts, sources)]
+        ChatAI_pipeline.add_document_(texts, vectorstore)
+        response = create_success_response(200, {"message": "Chatbot trained successfully."})
+        logger.info(f">>>Chatbot trained successfully for {vectorstore}.<<<")
+
+        return response
+
+
+    except Exception as e:
+        logger.error(f">>>Error in train_chatbot: {e} for {vectorstore}.<<<")
+        raise e
+
+
+
+
+
+@chatai_api_router.post("/new_chatbot")
+async def new_chatbot(request: NewChatbotRequest):
+    logger.info(f">>> new_chatbot API Triggered <<<")
+    try:
+        response = user_management.new_chatbot_(chatbot_name=request.chatbot_name, username=request.username)
+        logger.info(f">>> Chatbot created successfully for {request.username}.<<<")
+        return response
+
+    except Exception as e:
+        logger.error(f">>>Error in new_chatbot: {e} for {request.username}.<<<")
+        raise_http_exception(500, "Internal Server Error")

core/api/jwt_bearer.py

@@ -0,0 +1,27 @@
+import os
+from fastapi import Depends
+from supabase import create_client
+from core import logging as logger
+from core.utils.error_handling import create_error_response, raise_http_exception
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from dotenv import load_dotenv
+load_dotenv()
+
+security = HTTPBearer()
+
+supabase_client = create_client(
+    os.getenv("SUPABASE_URL"),
+    os.getenv("SUPABASE_KEY")
+)
+
+
+async def access_check_bearer(credentials: HTTPAuthorizationCredentials = Depends(security)):
+    access_token = credentials.credentials
+    try:
+        supabase_client.auth.get_user(access_token)
+
+    except Exception as e:
+        logger.info(f">>> Invalid access token {e}<<<")
+        raise_http_exception(code=401,
+                                   message="Invalid Access Token", details=[
+                {"info": "Invalid access token or access token expired please login again"}])

core/api/user_management_api.py

@@ -0,0 +1,152 @@
+from fastapi import Depends
+from core import logging as logger
+from core.models.apis_models import *
+from fastapi.routing import APIRouter
+from core.api.jwt_bearer import access_check_bearer, supabase_client
+from core.pipeline.user_management_pipeline import SupabaseUserManagementPipeline
+
+from core.utils.error_handling import raise_http_exception, create_success_response, create_error_response
+
+user_management_api_router = APIRouter(tags=["User Management"])
+
+user_management = SupabaseUserManagementPipeline()
+
+
+@user_management_api_router.post("/user_signup")
+async def user_signup(request: UserSignupRequest):
+    logger.info(f">>>user_signup API Triggered <<<")
+    response = user_management.user_signup_(username=request.username, password=request.password,email=request.email)
+    logger.info(f">>>user_signup API Success<<<")
+
+    return response
+
+
+@user_management_api_router.post("/user_signin")
+async def user_signin(request: UserSigninRequest):
+    logger.info(f">>>user_signin API Triggered <<<")
+
+    response = user_management.user_signin_(username=request.username, password=request.password,email=request.email)
+    if response != None:
+        logger.info(f">>>user_signin API Success.<<<")
+        return response
+    else:
+        logger.info(f">>> Username or password is incorrect please try again.<<<")
+        response = create_error_response(400, "Username or password is incorrect please try again.")
+        return response
+
+
+@user_management_api_router.post("/get_user_data")
+async def get_user_data(request: GetUserDataRequest):
+    logger.info(f">>>get_user_data API Triggered <<<")
+    response = user_management.get_user_data_(access_token=request.access_token)
+    return response
+
+
+@user_management_api_router.post("/login_with_access_token")
+async def login_with_access_token(request: LoginWithAccessTokenRequest):
+    logger.info(f">>>login_with_access_token API Triggered <<<")
+
+    response = user_management.login_with_access_token_(access_token=request.access_token,
+                                                        refresh_token=request.refresh_token)
+    logger.info(f">>>login_with_access_token API Success<<<")
+    return response
+
+
+@user_management_api_router.post("/set_session_data")
+async def set_session_data(request: SetSessionDataRequest):
+    logger.info(f">>> set_session_data API Triggered <<<")
+
+    response = user_management.set_session_data_(access_token=request.access_token, refresh_token=request.refresh_token,
+                                                 user_id=request.user_id)
+    return response
+
+
+@user_management_api_router.post("/sign_out")
+async def sign_out():
+    logger.info(f">>> sign_out API Triggered <<<")
+
+    response = user_management.sign_out_()
+    logger.info(f">>>sign_out API Success<<<")
+    return response
+
+
+@user_management_api_router.post("/oauth_signin")
+async def oauth_signin():
+    logger.info(f">>> oauth_signin API Triggered <<<")
+    response = user_management.oauth_signin_()
+    logger.info(f">>>oauth_signin API Success<<<")
+    return response
+
+
+@user_management_api_router.post("/check_session")
+async def check_session():
+    logger.info(f">>>check_session API Triggered <<<")
+
+    response = user_management.check_session_()
+    return response
+
+
+@user_management_api_router.get("/get_public_chatbot")
+async def get_public_chatbots():
+    logger.info(f">>>get_public_chatbot API Triggered<<<")
+    try:
+        response = supabase_client.table("ChatAI_ChatbotInfo").select("*").eq("isPrivate", False).execute().data
+        logger.info(f">>>Public chatbots fetched successfully.<<<")
+        return response
+    except Exception as e:
+        logger.error(f">>>Error in get_public_chatbot: {e}<<<")
+        raise_http_exception(500, "Internal Server Error")
+
+
+@user_management_api_router.post("/public_private_check")
+async def public_or_private(request: PublicPrivateCheckRequest):
+    vectorstore, mode = request.vectorstore, request.mode
+    logger.info(f">>>public_private_check API Triggered for {vectorstore}.<<<")
+    try:
+        
+        username, chatbot_name = vectorstore.split("$")[1], vectorstore.split("$")[2]
+        if len(mode) == 0:
+            value = (
+                supabase_client.table("ChatAI_ChatbotInfo")
+                .select("isPrivate")
+                .eq("user_id", username)
+                .eq("chatbotname", chatbot_name)
+                .execute()
+            )
+            value = value.data[0]["isPrivate"]
+            response = create_success_response(200, {"output": value})
+        else:
+            response = (
+                supabase_client.table("ChatAI_ChatbotInfo")
+                .update({"isPrivate": mode})
+                .eq("user_id", username)
+                .eq("chatbotname", chatbot_name)
+                .execute()
+            )
+            response = create_success_response(200, {"output": response})
+        logger.info(f">>>Public/Private check successful for {vectorstore}.<<<")
+        return response
+
+
+    except Exception as e:
+        logger.error(f">>>Error in public_private_check: {e} for {vectorstore}.<<<")
+        raise_http_exception(500, "Internal Server Error")
+
+
+@user_management_api_router.post("/refresh_session", dependencies=[Depends(access_check_bearer)])
+async def refresh_session(request: RefreshSessionRequest):
+    logger.info(f">>>refresh_session API Triggered <<<")
+    response = user_management.refresh_session__(refresh_token=request.refresh_token)
+    logger.info(f">>>refresh token fetched successfully.<<<")
+
+    return response
+
+
+@user_management_api_router.post("/username_creation_oauth", dependencies=[Depends(access_check_bearer)])
+async def username_creation_oauth(request: UsernameCreationOauthRequest):
+    logger.info(f">>> username_creation_oauth API Triggered <<<")
+
+    response = user_management.username_creation_oauth_(username=request.username,email=request.email, user_id=request.user_id)
+
+    logger.info(f">>>username creation successful.<<<")
+    return response

core/models/apis_models.py

@@ -0,0 +1,137 @@
+from pydantic import BaseModel,EmailStr
+from typing import List, Optional
+
+
+
+
+## ---------------------------------- Chatbot API Models ----------------------------------
+
+class AddTextRequest(BaseModel):
+    vectorstore: str
+    text: str
+
+
+class AddWebsiteRequest(BaseModel):
+    website_urls: List[str]
+    vectorstore: str
+    source: str
+
+
+class AnswerQueryRequest(BaseModel):
+    query: str
+    vectorstore: str
+    llm_model: str = "llama3-70b-8192"
+
+class GetLinksRequest(BaseModel):
+    url: str
+
+
+
+class WebsiteUrlTextExtractionRequest(BaseModel):
+    url: str
+
+
+class WebsiteUrlTextExtractionListRequest(BaseModel):
+    urls: List[str]
+
+
+class GetCurrentCountRequest(BaseModel):
+    vectorstore: str
+
+
+class ListChatbotsRequest(BaseModel):
+    username: str
+
+
+class GetChatHistoryRequest(BaseModel):
+    vectorstore: str
+
+
+class ChatHistoryItem(BaseModel):
+    timestamp: str
+    question: str
+    response: str
+
+
+class DeleteChatbotRequest(BaseModel):
+    vectorstore: str
+
+
+class TrainChatbotRequest(BaseModel):
+    vectorstore: str
+    urls: list[str]
+
+
+class LoadPDFRequest(BaseModel):
+    vectorstore: str
+
+
+class LoadEditedJson(BaseModel):
+    vectorstore: str
+    data_source_name: str
+    source_endpoint: str
+    json_data: dict
+
+
+class PublicPrivateCheckRequest(BaseModel):
+    vectorstore: str
+    mode: str | None = None
+
+
+class DeleteChatbotSourceRequest(BaseModel):
+    vectorstore: str
+    data_source_name: str
+
+
+## ---------------------------------- User Management API Models ----------------------------------
+
+class UserSignupRequest(BaseModel):
+    username: str
+    email:EmailStr
+    password: str
+
+
+class UserSigninRequest(BaseModel):
+    username:str
+    email:EmailStr
+
+    password: str
+
+
+class CheckSessionRequest(BaseModel):
+    user_id: str
+
+
+class GetUserDataRequest(BaseModel):
+    access_token: str
+
+
+class RefreshSessionRequest(BaseModel):
+    refresh_token: str
+
+
+class LoginWithAccessTokenRequest(BaseModel):
+    access_token: str
+    refresh_token: str
+
+
+class UsernameCreationOauthRequest(BaseModel):
+    username: str
+    email: str
+    user_id: str
+
+
+
+class SetSessionDataRequest(BaseModel):
+    access_token: str
+    refresh_token: str
+    user_id: str
+
+
+class SignOutRequest(BaseModel):
+    user_id: str
+
+
+class NewChatbotRequest(BaseModel):
+    chatbot_name: str
+    username: str

core/models/response_handling_models.py

@@ -0,0 +1,19 @@
+from typing import Any, Dict
+from pydantic import BaseModel
+
+class SuccessResponse(BaseModel):
+    status: str
+    code: int
+    data: Dict[str, Any]
+
+
+class ErrorResponse(BaseModel):
+    status: str
+    error: Dict[str, Any]
+
+
+class SuccessResponseUsermanagement(BaseModel):
+    status: str
+    message: str
+    code: int
+    data: Dict[str, Any]

core/models/utls.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel, Field
+
+
+class FollowUps(BaseModel):
+    q1: str = Field(description="First Follow-up Question")
+    q2: str = Field(description="Second Follow-up Question")
+    q3: str = Field(description="Third Follow-up Question")
+

core/pipeline/chataipipeline.py

@@ -0,0 +1,61 @@
+
+from langchain_core.prompts import ChatPromptTemplate
+from langchain_core.prompts import PromptTemplate
+from core.prompts.custom_prompts import _custom_prompts
+from core.services.answer_query.answerquery import AnswerQuery
+from core.services.document.add_document import AddDocument
+from core.services.embeddings.Qdrant_BM25_embedding import qdrant_bm25_embedding
+from core.services.embeddings.jina_embeddings import jina_embedding
+from core.services.get_links.web_scraper import WebScraper
+from core.services.ocr.replicate_ocr.replicate_ocr import ReplicateOCR as OCRService
+from core.services.pdf_extraction.image_pdf.image_pdf_text_extraction import get_text_from_image_pdf
+from core.services.pdf_extraction.text_pdf.text_pdf_extraction import extract_text_from_pdf
+from core.services.website_url.text_extraction_urlsnew import WebScrapertext
+from core.utils.utils import json_parser
+
+
+class ChatAIPipeline:
+    def __init__(self):
+        prompt_template = _custom_prompts["RAG_ANSWER_PROMPT"]
+        follow_up_prompt_template = _custom_prompts["FOLLOW_UP_PROMPT"]
+        prompt = ChatPromptTemplate.from_template(prompt_template)
+        json_parser_ = json_parser()
+        follow_up_prompt = PromptTemplate(
+            template=follow_up_prompt_template,
+            input_variables=["context"],
+            partial_variables={"format_instructions": json_parser_.get_format_instructions()},
+        )
+        self.vector_embedding = jina_embedding()
+        self.sparse_embedding = qdrant_bm25_embedding()
+        self.add_document_service = AddDocument(self.vector_embedding, self.sparse_embedding)
+
+        self.answer_query_service = AnswerQuery(vector_embedding=self.vector_embedding,
+                                                sparse_embedding=self.sparse_embedding, prompt=prompt,
+                                                follow_up_prompt=follow_up_prompt, json_parser=json_parser_)
+        self.get_website_links = WebScraper()
+        self.ocr_service = OCRService()
+        self.web_text_extractor = WebScrapertext()
+
+    def add_document_(self, texts: list[tuple[str]], vectorstore: str):
+        return self.add_document_service.add_documents(texts=texts, vectorstore=vectorstore)
+
+    def answer_query_(self, query: str, vectorstore: str, llm_model: str = "llama-3.3-70b-versatile"):
+        output, follow_up_questions, source = self.answer_query_service.answer_query(query=query,
+                                                                                     vectorstore=vectorstore,
+                                                                                     llmModel=llm_model)
+        return output, follow_up_questions, source
+
+    def get_links_(self, url: str, timeout: int):
+        return self.get_website_links.get_links(url=url, timeout=timeout)
+
+    def image_pdf_text_extraction_(self, image_pdf: bytes):
+        return get_text_from_image_pdf(pdf_bytes=image_pdf)
+
+    def text_pdf_extraction_(self, pdf: str):
+        return extract_text_from_pdf(pdf_path=pdf)
+
+    def website_url_text_extraction_(self, url: str):
+        return self.web_text_extractor.extract_text_from_url(url=url)
+
+    def website_url_text_extraction_list_(self, urls: list):
+        return self.web_text_extractor.extract_text_from_urls(urls=urls)

core/pipeline/user_management_pipeline.py

@@ -0,0 +1,75 @@
+
+from core.services.supabase.user_management.user_service import UserManagement
+from core.services.supabase.user_management.chatbot_management import SupabaseChatoBotManagement
+from core.services.supabase.chat_ai_setup.chataiusermanagement import ChatAIUserDBSetup
+from core.services.supabase.user_management.chat_history import get_chat_history
+from supabase.client import create_client
+from qdrant_client import QdrantClient
+import os
+
+SUPABASE_URL = os.getenv("SUPABASE_URL")
+SUPABASE_KEY = os.getenv("SUPABASE_KEY")
+
+QDRANT_URL = os.getenv("QDRANT_URL")
+QDRANT_API_KEY = os.getenv("QDRANT_API_KEY")
+
+
+class SupabaseUserManagementPipeline:
+    def __init__(self):
+        self.supabase_client = create_client(SUPABASE_URL, SUPABASE_KEY)
+        self.qdrant_client = QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY)
+        self.user_management = UserManagement(supabase_client=self.supabase_client)
+        self.chatbot_management = SupabaseChatoBotManagement(supabase_client=self.supabase_client,
+                                                             qdrant_client=self.qdrant_client)
+        self.user_db_setup = ChatAIUserDBSetup(supabase_client=self.supabase_client)
+
+    def user_signup_(self, username: str,password: str,email:str) -> dict:
+        return self.user_management.user_signup(username=username, password=password, email=email)
+
+    def user_signin_(self, username: str, password: str,email:str) -> dict:
+        return self.user_management.user_signin(username=username, password=password,email=email)
+
+    def check_session_(self) -> dict:
+        return self.user_management.check_session()
+
+    def get_user_data_(self, access_token: str) -> dict:
+        return self.user_management.get_user_data(access_token=access_token)
+
+    def refresh_session__(self, refresh_token: str) -> dict:
+        return self.user_management.refresh_session_(refresh_token=refresh_token)
+
+    def login_with_access_token_(self, access_token: str, refresh_token: str) -> dict:
+        return self.user_management.login_with_access_token(access_token=access_token, refresh_token=refresh_token)
+
+    def username_creation_oauth_(self, username: str, user_id: str):
+        return self.user_management.user_name_creation_oauth(user_id=user_id, username=username)
+
+    def set_session_data_(self, access_token: str, refresh_token: str, user_id: str):
+        return self.user_management.set_session_data(access_token=access_token, refresh_token=refresh_token,
+                                                     user_id=user_id)
+
+    def sign_out_(self):
+        return self.user_management.sign_out_()
+
+    def oauth_signin_(self) -> dict:
+        return self.user_management.oauth()
+
+    def new_chatbot_(self, chatbot_name: str, username: str):
+        return self.chatbot_management.new_chatbot(chatbot_name=chatbot_name, username=username)
+
+    def get_chat_history_(self, vectorstore: str):
+        return get_chat_history(vectorstore=vectorstore, supabase_client=self.supabase_client)
+
+    def delete_table(self, table_name: str):
+        return self.chatbot_management.delete_table(table_name=table_name)
+
+    def list_tables(self, username: str):
+        return self.chatbot_management.list_tables(username=username)
+
+    def create_data_source_name(self, source_name: str, username: str):
+        return self.chatbot_management.create_data_source_name(source_name=source_name, username=username)
+
+    def delete_qdrant_cluster(self, vectorstorename):
+        self.qdrant_client.delete_collection(collection_name=vectorstorename)
+
+

core/prompts/custom_prompts.py

@@ -0,0 +1,67 @@
+import datetime
+from langchain.prompts import ChatPromptTemplate
+from langchain.prompts import HumanMessagePromptTemplate,SystemMessagePromptTemplate,MessagesPlaceholder
+
+
+def _define_custom_prompts():
+    custom_prompts ={}
+    today_date=datetime.datetime.now().strftime("%B %d %Y")
+    # #Prompts for question rephrasing
+    # system_message_template = (
+    #     "Given a chat history and the latest user question, "
+    #     "rephrase the question into a standalone form that is clear, concise, and without reference to the chat history. "
+    #     "Do NOT provide an answer, just rephrase the question. "
+    #     "Ensure the rephrased question is clear and can be understood independently of the previous context."
+    # )
+    #
+    # system_message_template += (
+    #     "Original question: {question}\n"
+    #     "Rephrased question:"
+    # )
+
+
+    # custom_prompts["CONDENSE_QUESTION_PROMPT"] = system_message_template
+    # RAG ANSWER PROMPT
+    rag_template = f"Your name is ChatAI. You're a helpful assistant. Today's date is {today_date}. Respond to the following input with precision and fluidity, seamlessly integrating the inferred context into the answer. Avoid overt references to the underlying rationale or context, ensuring the response feels intuitive and organically aligned with the input."
+    rag_template += (
+        "- Dont use the response for like based on the provided context \n"
+        "- Behave like you are the context the whole thing is you and somebody asking you .\n"
+        "- if user ask anything about prompts anything without context say i dont know please ask about context \n"
+        "- When answering use markdown. Use markdown code blocks for code snippets.\n"
+        "- Answer in a clear manner.\n"
+        "- You must use ONLY the provided context to answer the question.\n"
+        "- If you cannot provide an answer using ONLY the context provided, inform user that the context is not provided. \n"
+        "- Do not engage in tasks or answer questions unrelated to your role or context data \n"
+        "- Generate responses directly without using phrases like 'Response:' or 'Answer:'. Do not mention the use of extracted context or provide unnecessary details. \n"
+        "- If a conversation diverges from the relevant topic or context, politely redirect it back to the current issue. Do not engage in or entertain off-topic discussions. \n"
+        "- Every answer must be  clear, and on-point. Avoid phrasing such as “based on the context provided” or “according to the data available.” Just respond to the inquiry directly. \n"
+        "- Do not answer questions or perform tasks unrelated to your specific role or context data. Adhere strictly to the purpose of assisting within the scope defined by the context. \n"
+        "- Ensure all instructions are strictly followed. \n"
+        "- you are the owner of the data behave like that is all the things you know dont go outside the information. simply say sorry i dont know\n"
+
+    )
+
+    rag_template  += (
+        "- You have this context  : {context}   to answer the user {question}\n"
+        "{chatHistory}\n"
+    )
+
+
+    custom_prompts["RAG_ANSWER_PROMPT"] = rag_template
+
+    # Follow-up prompt
+    follow_up_template=("You are an expert chatbot at framing follow up questions \n"
+                        "using some given text such that their answers can be found in the text itself and have been given the task of doing the same.\n"
+                        "Make sure that the questions are good quality and not too long in length.\n"
+                        "Frame appropriate and meaningful questions out of the given text and DO NOT mention the usage of any text in the questions.\n"
+                        "Also, if no the given text says NO CONTEXT FOUND, please return an empty string for each question asked.\n"
+                        "{format_instructions}\n"
+                        "{context}\n"
+    )
+
+    custom_prompts["FOLLOW_UP_PROMPT"]=follow_up_template
+
+
+    return custom_prompts
+
+_custom_prompts =_define_custom_prompts()

core/services/answer_query/answerquery.py

@@ -0,0 +1,167 @@
+import os
+
+from langchain.memory import ChatMessageHistory
+from langchain.retrievers import ContextualCompressionRetriever
+from langchain_community.document_compressors import JinaRerank
+from langchain_core.chat_history import BaseChatMessageHistory
+from langchain_core.output_parsers import StrOutputParser
+from langchain_core.runnables import RunnablePassthrough, RunnableLambda
+from langchain_core.runnables.history import RunnableWithMessageHistory
+from langchain_groq import ChatGroq
+
+from core.services.vector_db.qdrent.upload_document import answer_query_from_existing_collection
+
+os.environ["JINA_API_KEY"] = os.getenv("JINA_API")
+
+
+class AnswerQuery:
+    def __init__(self, prompt, vector_embedding, sparse_embedding, follow_up_prompt, json_parser):
+        self.chat_history_store = {}
+        self.compressor = JinaRerank(model="jina-reranker-v2-base-multilingual")
+        self.vector_embed = vector_embedding
+        self.sparse_embed = sparse_embedding
+        self.prompt = prompt
+        self.follow_up_prompt = follow_up_prompt
+        self.json_parser = json_parser
+
+    def format_docs(self, docs: str):
+        global sources
+        global temp_context
+        sources = []
+        context = ""
+        for doc in docs:
+            context += f"{doc.page_content}\n\n\n"
+            source = doc.metadata
+            source = source["source"]
+            sources.append(source)
+        if context == "":
+            context = "No context found"
+        else:
+            pass
+        sources = list(set(sources))
+        temp_context = context
+        return context
+
+    def answer_query(self, query: str, vectorstore: str, llmModel: str = "llama-3.3-70b-versatile"):
+        global sources
+        global temp_context
+        vector_store_name = vectorstore
+        vector_store = answer_query_from_existing_collection(vector_embed=self.vector_embed,
+                                                             sparse_embed=self.sparse_embed,
+                                                             vectorstore=vectorstore)
+
+        retriever = vector_store.as_retriever(search_type="mmr", search_kwargs={"k": 10, "fetch_k": 20})
+        compression_retriever = ContextualCompressionRetriever(
+            base_compressor=self.compressor, base_retriever=retriever
+        )
+        brain_chain = (
+                {"context": RunnableLambda(lambda x: x["question"]) | compression_retriever | RunnableLambda(
+                    self.format_docs),
+                 "question": RunnableLambda(lambda x: x["question"]),
+                 "chatHistory": RunnableLambda(lambda x: x["chatHistory"])}
+                | self.prompt
+                | ChatGroq(model=llmModel, temperature=0.75, max_tokens=512)
+                | StrOutputParser()
+        )
+        message_chain = RunnableWithMessageHistory(
+            brain_chain,
+            self.get_session_history,
+            input_messages_key="question",
+            history_messages_key="chatHistory"
+        )
+        chain = RunnablePassthrough.assign(messages_trimmed=self.trim_messages) | message_chain
+        follow_up_chain = self.follow_up_prompt | ChatGroq(model_name="llama-3.3-70b-versatile",
+                                                           temperature=0) | self.json_parser
+
+        output = chain.invoke(
+            {"question": query},
+            {"configurable": {"session_id": vector_store_name}}
+        )
+        follow_up_questions = follow_up_chain.invoke({"context": temp_context})
+
+        return output, follow_up_questions, sources
+
+    async def answer_query_stream(self, query: str, vectorstore: str, llmModel):
+        global sources
+        global temp_context
+
+        vector_store_name = vectorstore
+        vector_store = answer_query_from_existing_collection(
+            vector_embed=self.vector_embed,
+            sparse_embed=self.sparse_embed,
+            vectorstore=vectorstore
+        )
+
+        retriever = vector_store.as_retriever(search_type="mmr", search_kwargs={"k": 10, "fetch_k": 20})
+        compression_retriever = ContextualCompressionRetriever(
+            base_compressor=self.compressor,
+            base_retriever=retriever
+        )
+
+        brain_chain = (
+                {
+                    "context": RunnableLambda(lambda x: x["question"]) | compression_retriever | RunnableLambda(
+                        self.format_docs),
+                    "question": RunnableLambda(lambda x: x["question"]),
+                    "chatHistory": RunnableLambda(lambda x: x["chatHistory"])
+                }
+                | self.prompt
+                | ChatGroq(
+            model=llmModel,
+            temperature=0.75,
+            max_tokens=512,
+            streaming=True
+        )
+                | StrOutputParser()
+        )
+
+        message_chain = RunnableWithMessageHistory(
+            brain_chain,
+            self.get_session_history,
+            input_messages_key="question",
+            history_messages_key="chatHistory"
+        )
+
+        chain = RunnablePassthrough.assign(messages_trimmed=self.trim_messages) | message_chain
+
+        async for chunk in chain.astream(
+                {"question": query},
+                {"configurable": {"session_id": vector_store_name}}
+        ):
+            yield {
+                "type": "main_response",
+                "content": chunk
+            }
+
+        follow_up_chain = self.follow_up_prompt | ChatGroq(
+            model_name="llama-3.3-70b-versatile",
+            temperature=0
+        ) | self.json_parser
+
+        follow_up_questions = await follow_up_chain.ainvoke({"context": temp_context})
+
+        yield {
+            "type": "follow_up_questions",
+            "content": follow_up_questions
+        }
+
+        yield {
+            "type": "sources",
+            "content": sources
+        }
+
+    def trim_messages(self, chain_input):
+        for store_name in self.chat_history_store:
+            messages = self.chat_history_store[store_name].messages
+            if len(messages) <= 1:
+                pass
+            else:
+                self.chat_history_store[store_name].clear()
+                for message in messages[-1:]:
+                    self.chat_history_store[store_name].add_message(message)
+        return True
+
+    def get_session_history(self, session_id: str) -> BaseChatMessageHistory:
+        if session_id not in self.chat_history_store:
+            self.chat_history_store[session_id] = ChatMessageHistory()
+        return self.chat_history_store[session_id]

core/services/document/add_document.py

@@ -0,0 +1,37 @@
+
+import string
+from langchain.docstore.document import Document
+from langchain_text_splitters import RecursiveCharacterTextSplitter
+from core.services.vector_db.qdrent.upload_document import upload_document_existing_collection
+
+
+class AddDocument:
+    def __init__(self, vector_embedding, sparse_embedding):
+        self.vector_embed = vector_embedding
+        self.sparse_embed = sparse_embedding
+
+    def clean_text(self,text:str)->str:
+        text=text.replace("\n", "")
+        text = text.translate(str.maketrans('', '', string.punctuation.replace(".", "")))
+        text = text.encode('utf-8', errors='ignore').decode('utf-8')
+        return text
+    def add_documents(self, texts: list[tuple[str]], vectorstore: str):
+        splitter = RecursiveCharacterTextSplitter(
+            chunk_size=400,
+            chunk_overlap=100,
+            add_start_index=True
+        )
+        contents, sources = zip(*texts)
+        cleaned_texts = [self.clean_text(text) for text in contents]
+
+        # Create Document objects
+        docs = [
+            Document(page_content=text, metadata={"source": source})
+            for text, source in zip(cleaned_texts, sources)
+        ]
+
+        # Split documents and upload
+        documents = splitter.split_documents(docs)
+        upload_document_existing_collection(vector_embed=self.vector_embed,
+                                            sparse_embed=self.sparse_embed,
+                                            vectorstore=vectorstore, documents=documents)

core/services/embeddings/Qdrant_BM25_embedding.py

@@ -0,0 +1,8 @@
+
+from langchain_qdrant import FastEmbedSparse
+
+
+def qdrant_bm25_embedding():
+    instance = FastEmbedSparse(model="Qdrant/BM25", threads=20, parallel=0)
+
+    return instance

core/services/embeddings/jina_embeddings.py

@@ -0,0 +1,8 @@
+from langchain_community.embeddings import JinaEmbeddings
+import os
+
+def jina_embedding():
+    text_embeddings=JinaEmbeddings(
+    jina_api_key=os.getenv('JINA_API'),
+    model_name="jina-embeddings-v3")
+    return text_embeddings

core/services/get_links/web_scraper.py

@@ -0,0 +1,42 @@
+import time
+import requests
+from bs4 import BeautifulSoup
+from urllib.parse import urlparse, urljoin
+from concurrent.futures import ThreadPoolExecutor
+
+class WebScraper:
+    def __init__(self):
+        pass
+
+    def get_links(self,url: str, timeout=4):
+        start = time.time()
+
+        def get_links_from_page(url: str) -> list:
+            response = requests.get(url)
+            soup = BeautifulSoup(response.content, "lxml")
+            anchors = soup.find_all("a")
+            links = []
+            for anchor in anchors:
+                if "href" in anchor.attrs:
+                    if urlparse(anchor.attrs["href"]).netloc == urlparse(url).netloc:
+                        links.append(anchor.attrs["href"])
+                    elif not anchor.attrs["href"].startswith(("//", "file", "javascript", "tel", "mailto", "http")):
+                        links.append(urljoin(url + "/", anchor.attrs["href"]))
+                    else:
+                        pass
+                    links = [link for link in links if "#" not in link]
+                    links = list(set(links))
+                else:
+                    continue
+            return links
+
+        links = get_links_from_page(url)
+        unique_links = set()
+        for link in links:
+            now = time.time()
+            if now - start > timeout:
+                break
+            else:
+                unique_links = unique_links.union(set(get_links_from_page(link)))
+        return list(set([x[:len(x) - 1] if x[-1] == "/" else x for x in unique_links]))
+

core/services/ocr/replicate_ocr/replicate_ocr.py

@@ -0,0 +1,33 @@
+
+import base64
+import cv2
+import replicate
+
+
+def replicate_run(input):
+    output = replicate.run(
+        "abiruyt/text-extract-ocr:a524caeaa23495bc9edc805ab08ab5fe943afd3febed884a4f3747aa32e9cd61",
+        input=input
+    )
+    return output
+
+
+class ReplicateOCR:
+    def __init__(self):
+        pass
+
+    def read_text(self, image_path):
+        _, buffer = cv2.imencode('.png', image_path)
+        base_64_image = base64.b64encode(buffer).decode('utf-8')
+
+        input_image_uri = f"data:image/png;base64,{base_64_image}"
+        input = {
+            "image": input_image_uri,
+        }
+        output = replicate_run(input)
+        if output:
+            output = output.replace("\n", " ").replace("\t", " ")
+            return output
+
+        else:
+            return None

core/services/pdf_extraction/image_pdf/image_pdf_text_extraction.py

@@ -0,0 +1,38 @@
+
+import concurrent.futures
+import numpy as np
+from pdf2image import convert_from_bytes
+
+from core import logging as logger
+from core.services.ocr.replicate_ocr.replicate_ocr import ReplicateOCR
+
+
+def get_text_from_image_pdf(pdf_bytes: bytes):
+    easy_ocr = ReplicateOCR()
+
+    def process_image(args: tuple) -> str:
+        index, image = args
+        gray_image = image.convert('L')
+        np_image = np.array(gray_image)
+        text = easy_ocr.read_text(np_image)
+        logger.debug(f"Processed page {index + 1}")
+        return text
+
+    all_images = convert_from_bytes(
+        pdf_bytes,
+        thread_count=1,
+        grayscale=True,
+        size=(1700, None)
+    )
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
+        image_data = list(enumerate(all_images))
+        texts = list(executor.map(process_image, image_data))
+
+    logger.info("OCR processing completed")
+    try:
+        return {i + 1: text for i, text in enumerate(texts)}
+
+    except Exception as e:
+        logger.error(f"Error in OCR processing: {e}")
+        return None

core/services/pdf_extraction/text_pdf/text_pdf_extraction.py

@@ -0,0 +1,19 @@
+
+import base64
+import pymupdf
+from concurrent.futures import ThreadPoolExecutor
+
+
+def extract_text_from_page(page):
+    text = page.get_text()
+    return text
+
+
+def extract_text_from_pdf(pdf_path: str):
+    doc = pymupdf.open(pdf_path)
+    pages = [doc.load_page(i) for i in range(len(doc))]
+    with ThreadPoolExecutor() as executor:
+        texts = list(executor.map(extract_text_from_page, pages))
+
+    doc.close()
+    return {x + 1: y for x, y in enumerate(texts)}

core/services/supabase/chat_ai_setup/chataiusermanagement.py

@@ -0,0 +1,32 @@
+
+class ChatAIUserDBSetup:
+    def __init__(self, supabase_client):
+        self.client = supabase_client
+
+    def setup_new_user(self, user_id: str, username: str,email:str) -> dict:
+        userData = self.client.table("ChatAI_UserInfo").select("*").execute().data
+        if username not in [userData[x]["username"] for x in range(len(userData))]:
+            try:
+                self.client.table("ChatAI_UserInfo").insert(
+                    {"user_id": user_id, "username": username,"email":email}).execute()
+
+                self.client.table("ChatAI_UserConfig").insert({"user_id": username}).execute()
+
+                res = {
+                    "code": 200,
+                    "message": "User Setup Successful"
+                }
+
+            except Exception as e:
+                res = {
+                    "code": 409,
+                    "message": "Email already exists",
+                }
+
+            return res
+
+        else:
+            return {
+                "code": 409,
+                "message": "Username already exists"
+            }

core/services/supabase/limit/limit_check.py

@@ -0,0 +1,14 @@
+class LimitChecker:
+    def __init__(self, supabase_client):
+        limits = self.fetch_limits(supabase_client)
+        self.url_limit = limits.get("total_website_url")
+        self.pdf_limit = limits.get("text_pdf_limit")
+        self.image_pdf_limit = limits.get("image_pdf_limit")
+
+    def fetch_limits(self, supabase_client):
+        table = supabase_client.table("ChatAI_limit_check")
+        results = table.select("*").execute()
+        return results.data[0]
+
+    def __iter__(self):
+        return iter((self.url_limit, self.pdf_limit, self.image_pdf_limit))

core/services/supabase/user_management/__init__.py

@@ -0,0 +1 @@
+

core/services/supabase/user_management/chat_history.py

@@ -0,0 +1,8 @@
+
+def get_chat_history(supabase_client, vectorstore: str):
+    username, chatbotName = vectorstore.split("$")[1], vectorstore.split("$")[2]
+    response = supabase_client.table("ChatAI_ChatHistory").select("timestamp", "question", "response").eq(
+        "username",
+        username).eq(
+        "chatbotName", chatbotName).execute().data
+    return response

core/services/supabase/user_management/chatbot_management.py

@@ -0,0 +1,62 @@
+
+import random
+import re
+import string
+
+from core.services.supabase.chat_ai_setup.chataiusermanagement import ChatAIUserDBSetup
+from core.services.vector_db.qdrent.qdrentvector_db import QdrantVectorStoreDB
+from core.utils.error_handling import create_error_response
+from core.utils.utils import load_ini_config
+from core.services.embeddings.Qdrant_BM25_embedding import qdrant_bm25_embedding
+from core.services.embeddings.jina_embeddings import jina_embedding
+
+
+
+class SupabaseChatoBotManagement:
+    def __init__(self, supabase_client, qdrant_client):
+        vector_embeddings = jina_embedding()
+        sparse_embeddings = qdrant_bm25_embedding()
+        self.config = load_ini_config("config.ini")
+        self.qdrant_vector_db = QdrantVectorStoreDB(qdrant_client=qdrant_client, vector_embeddings=vector_embeddings,
+                                                    sparse_embeddings=sparse_embeddings)
+        self.supabase_client = supabase_client
+        self.user_db_setup = ChatAIUserDBSetup(self.supabase_client)
+
+    def new_chatbot(self, chatbot_name: str, username: str):
+        currentBotCount = len(self.qdrant_vector_db.list_tables(username=username))
+        limit = self.supabase_client.table("ChatAI_UserConfig").select("chatbotLimit").eq("user_id",
+                                                                                             username).execute().data[
+            0][
+            "chatbotLimit"]
+        if currentBotCount >= int(limit):
+            return create_error_response(code=400, message="You have reached the limit of chatbots you can create")
+        self.supabase_client.table("ChatAI_ChatbotInfo").insert(
+            {"user_id": username, "chatbotname": chatbot_name}).execute()
+        chat_bot_name = f"chatai${username}${chatbot_name}"
+        return self.qdrant_vector_db.create_table(table_name=chat_bot_name)
+
+    def delete_table(self, table_name: str):
+        return self.qdrant_vector_db.delete_table(table_name=table_name)
+
+    def list_tables(self, username: str):
+        return self.qdrant_vector_db.list_tables(username=username)
+
+    def create_data_source_name(self, source_name, username):
+        sources = [x["dataSourceName"] for x in
+                   self.supabase_client.table("ChatAI_ChatbotDataSources").select("dataSourceName").execute().data]
+        n_sources = re.sub(r'[^a-zA-Z0-9]', '_', source_name)
+        n_sources = n_sources.lower()
+        new_source_name = f"chatai_{username}_{n_sources}_{''.join(random.choices(string.ascii_letters + string.digits, k=60))}"
+        if new_source_name not in sources:
+            return new_source_name
+        else:
+            return self.create_data_source_name(source_name, username)
+
+    def get_current_count(self, username):
+        data_ = self.supabase_client.table("ChatAI_ChatbotInfo").select("*").filter("user_id", "eq",
+                                                                                       username).execute().data
+        current_count = 0
+        for data in data_:
+            current_count += int(data["charactercount"])
+
+        return current_count

core/services/supabase/user_management/token_limit.py

@@ -0,0 +1,45 @@
+
+import pandas as pd
+from core import logging as logger
+
+
+def token_limit_check(supabase_client, username, chatbot_name, len_text):
+    logger.info(f">>>token_limit_check API Triggered_{username}_{chatbot_name} <<<")
+
+    try:
+        chatbot_response = supabase_client.table("ChatAI_ChatbotInfo") \
+            .select("charactercount") \
+            .eq("user_id", username) \
+            .eq("chatbotname", chatbot_name) \
+            .execute()
+
+        if not chatbot_response.data:
+            logger.error(f"Error checking username and chatbotname{username},{chatbot_name}")
+            return False
+        currentCount = int(chatbot_response.data[0]["charactercount"])
+        newCount = currentCount + len_text
+        logger.info(f"{newCount}")
+
+        limit_response = supabase_client.table("ChatAI_UserConfig") \
+            .select("tokenLimit") \
+            .eq("user_id", username) \
+            .execute()
+        
+        limit = int(limit_response.data[0]["tokenLimit"])
+
+        if newCount < limit:
+            supabase_client.table("ChatAI_ChatbotInfo") \
+                .update({"charactercount": str(newCount)}) \
+                .eq("user_id", username) \
+                .eq("chatbotname", chatbot_name) \
+                .execute()
+
+            logger.info(f">>>token_limit_check API Success for user: {username} chatbot: {chatbot_name}<<<")
+            return True
+        else:
+            logger.error(f">>>token_limit_check exceeded for user: {username}{chatbot_name}<<<")
+            return False
+
+    except Exception as e:
+        raise e
+        return False

core/services/supabase/user_management/user_service.py

@@ -0,0 +1,155 @@
+
+import jwt
+from core.services.supabase.chat_ai_setup.chataiusermanagement import ChatAIUserDBSetup
+from jwt import ExpiredSignatureError, InvalidTokenError
+from fastapi import HTTPException
+from core.utils.error_handling import raise_http_exception, create_success_response, create_error_response, \
+    success_response_user_management
+from core.utils.utils import config_read
+
+config = config_read()
+
+
+class UserManagement:
+    def __init__(self, supabase_client):
+        self.supabase_client = supabase_client
+        self.user_db_setup = ChatAIUserDBSetup(self.supabase_client)
+
+    def user_signup(self, username: str, password: str,email:str) -> dict:
+        res, _ = self.supabase_client.auth.sign_up(
+            {"email": email, "password": password, "role": "user"}
+        )
+
+        user_id = res[1].id
+        r_ = self.user_db_setup.setup_new_user(user_id=user_id, username=username,email=email)
+        if r_.get('code') == 409:
+            response = create_error_response(code=409, message=r_.get('message'))
+        elif r_.get('code') == 200:
+            response = success_response_user_management(code=200, message="Signup successful", data={
+                "info": "Please check you email address for email verification"})
+        else:
+            response = raise_http_exception(code=400, message="Failed to sign up please try again later")
+        return response
+
+    def user_signin(self, username: str, password: str,email:str):
+        try:
+            res = self.supabase_client.auth.sign_in_with_password(
+                {"email": email, "password": password}
+            )
+
+            if res != None:
+                user_id = res.user.id
+                access_token = res.session.access_token
+                refresh_token = res.session.refresh_token
+
+                userData = self.supabase_client.table("ChatAI_UserInfo").select("*").filter("user_id", "eq",
+                                                                                               user_id).execute().data
+                username = userData[0]["username"]
+
+                message = {
+                    "username": username,
+                    "user_id": user_id,
+                    "access_token": access_token,
+                    "refresh_token": refresh_token,
+                }
+                response = success_response_user_management(code=200, message="Signin successful", data=message)
+
+                return response
+            else:
+                response = create_error_response(code=400, message="Email or password is incorrect please try again",)
+                return response
+
+
+
+
+        except Exception as e:
+            create_error_response(code=400, message="Failed to sign in please try again later",
+                                  details=[{"error": str(e)}])
+
+    def check_session(self):
+        res = self.supabase_client.auth.get_session()
+        if res == None:
+            try:
+                self.supabase_client.auth.sign_out()
+
+                return create_error_response(code=401, message="Session expired please login again")
+            except Exception as e:
+                create_error_response(code=400, message="Failed to check session")
+
+        else:
+            response = success_response_user_management(code=200, message="Session active", data=dict(res))
+            return response
+
+    def get_user_data(self, access_token: str) -> dict:
+        res = self.supabase_client.auth.get_user(jwt=access_token)
+        response = success_response_user_management(code=200, message="User data fetched successfully", data=res)
+        return response
+
+    def refresh_session_(self, refresh_token: str) -> dict:
+        res = self.supabase_client.auth.refresh_session(refresh_token)
+        response = success_response_user_management(code=200, message="Session refreshed successfully", data=res)
+
+        return response
+
+    def login_with_access_token(self, access_token: str, refresh_token: str):
+        try:
+            decoded_token = jwt.decode(access_token, options={"verify_signature": False})
+            user_id_oauth = decoded_token.get("username")
+            try:
+                self.supabase_client.table("ChatAI_UserInfo").select("*").filter("user_id", "eq",
+                                                                                    user_id_oauth).execute()
+                user_id = self.supabase_client.table("ChatAI_UserInfo").select("*").filter("email", "eq",
+                                                                                              user_id_oauth).execute()
+                if len(user_id.data) == 0:
+                    user_name = ""
+                else:
+                    user_name = user_id.data[0]["username"]
+
+                json = {
+                    "user_id": decoded_token.get("sub"),
+                    "email": decoded_token.get("email"),
+                    "access_token": access_token,
+                    "refresh_token": refresh_token,
+                    "issued_at": decoded_token.get("iat"),
+                    "expires_at": decoded_token.get("exp"),
+                    "username": user_name
+
+                }
+                response = success_response_user_management(code=200, message="Login with access token successful",
+                                                            data=json)
+                return response
+            except:
+
+                response = create_error_response(code=400, message="Failed to login with access token",
+                                                 details=[{"error": "User not found"}])
+
+                return response
+
+        except (ExpiredSignatureError, InvalidTokenError) as e:
+            create_error_response(code=400, message="Failed to login with access token", details=[{"error": str(e)}])
+
+    def user_name_creation_oauth(self, username: str, user_id: str,email:str):
+        r_ = self.user_db_setup.setup_new_user(user_id=user_id, username=username,email=email)
+        response = success_response_user_management(code=200, message="Username creation successful", data=r_)
+
+        return response
+
+    def set_session_data(self, access_token, refresh_token, user_id):
+        res = self.supabase_client.auth.set_session(access_token, refresh_token)
+        response = success_response_user_management(code=200, message="Session data set successfully", data=dict(res))
+        return response
+
+    def sign_out_(self):
+        try:
+
+            res = self.supabase_client.auth.sign_out()
+            response = success_response_user_management(code=200, message="Signout successful")
+            return response
+        except Exception as e:
+            raise HTTPException(status_code=400, detail=str(e))
+
+    def oauth(self):
+        res = self.supabase_client.auth.sign_in_with_oauth(
+            {"provider": "google", "options": {"redirect_to": f"{config.get('oauth', 'redirect_to')}"}})
+        response = success_response_user_management(code=200, message="OAuth successful", data=dict(res))
+        return response

core/services/vector_db/__init__.py

@@ -0,0 +1 @@
+

core/services/vector_db/qdrent/qdrentvector_db.py

@@ -0,0 +1,34 @@
+
+from core.services.vector_db.qdrent.upload_document import upload_document_new_collection
+from core.utils.error_handling import create_success_response, success_response_user_management, raise_http_exception
+
+
+class QdrantVectorStoreDB:
+    def __init__(self, qdrant_client, vector_embeddings, sparse_embeddings):
+        self.vector_embeddings = vector_embeddings
+        self.sparse_embeddings = sparse_embeddings
+        self.qdrant_client = qdrant_client
+
+    def create_table(self, table_name: str):
+        upload_document_new_collection(vector_embed=self.vector_embeddings, sparse_embed=self.sparse_embeddings,
+                                       table_name=table_name)
+        response = success_response_user_management(code=200, message="Chatbot Creation Successful")
+        return response
+
+
+    def delete_table(self, table_name: str):
+        try:
+            self.qdrant_client.delete_collection(collection_name=table_name)
+            return success_response_user_management(code=200, message="Chatbot Deletion Successful")
+        except Exception as e:
+            raise_http_exception(code=400, message="Failed to delete chatbot")
+
+
+    def list_tables(self, username: str):
+        try:
+            qdrant_collections = self.qdrant_client.get_collections()
+            return list(filter(lambda x: True if x.split("$")[1] == username else False,
+                               [x.name for x in qdrant_collections.collections]))
+
+        except Exception as e:
+            return e

core/services/vector_db/qdrent/upload_document.py

@@ -0,0 +1,47 @@
+
+import os
+
+from langchain_qdrant import QdrantVectorStore, RetrievalMode
+
+
+def answer_query_from_existing_collection(vector_embed, sparse_embed, vectorstore):
+    vector_store = QdrantVectorStore.from_existing_collection(
+        embedding=vector_embed,
+        sparse_embedding=sparse_embed,
+        collection_name=vectorstore,
+        url=os.environ["QDRANT_URL"],
+        api_key=os.environ["QDRANT_API_KEY"],
+        retrieval_mode=RetrievalMode.HYBRID
+    )
+    return vector_store
+
+
+def upload_document_existing_collection(documents, vector_embed, sparse_embed, vectorstore):
+    vector_store = QdrantVectorStore.from_documents(
+        documents=documents,
+        embedding=vector_embed,
+        sparse_embedding=sparse_embed,
+        prefer_grpc=True,
+        collection_name=vectorstore,
+        url=os.environ["QDRANT_URL"],
+        api_key=os.environ["QDRANT_API_KEY"],
+        force_recreate=True,
+        retrieval_mode=RetrievalMode.HYBRID
+    )
+
+
+def upload_document_new_collection(vector_embed, sparse_embed, table_name):
+    try:
+        qdrant = QdrantVectorStore.from_documents(
+            documents=[],
+            embedding=vector_embed,
+            sparse_embedding=sparse_embed,
+            url=os.environ["QDRANT_URL"],
+            prefer_grpc=True,
+            api_key=os.environ["QDRANT_API_KEY"],
+            collection_name=table_name,
+            force_recreate=True,
+            retrieval_mode=RetrievalMode.HYBRID
+        )
+    except Exception as e:
+        return str(e)