Initial commit for deploying the project

.gitattributes

@@ -33,3 +33,4 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+src/chroma_rag/chroma.sqlite3 filter=lfs diff=lfs merge=lfs -text

src/.gitignore

@@ -0,0 +1,18 @@
+# Ignore all MP3 files
+*.mp3
+
+# Ignore all PDF files
+*.pdf
+
+# Ignore all MP4 files
+*.mp4
+
+# Ignore all TXT files
+*.txt
+
+# Ignore environment files
+.env
+
+# Ignore Python cache files
+__pycache__/
+*.pyc

src/Dockerfile

@@ -0,0 +1,32 @@
+# Use Python 3.10 as base image
+FROM python:3.10-slim
+
+# Set working directory in container
+WORKDIR /app/src
+
+# Copy requirements file
+COPY requirements.txt .
+
+# Install dependencies
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy the entire project
+COPY . .
+
+# Expose ports for API and Streamlit
+EXPOSE 8000 8501
+
+# Create script to run both services
+RUN echo '#!/bin/bash\n\
+python response_api.py &\n\
+sleep 5\n\
+streamlit run app.py' > ./start.sh
+
+# Make the script executable
+RUN chmod +x ./start.sh
+
+# Copy .env file and set environment variables
+ENV $(cat .env | xargs)
+
+# Run the start script
+CMD ["./start.sh"]

src/README.md

@@ -0,0 +1,2 @@
+# Gemini_RAG
+Create multi-functionality LLM application which support RAG functionality with multiple agents.

src/agents_rag/agents.py

@@ -0,0 +1,78 @@
+from crewai import Agent
+from dotenv import load_dotenv
+
+from agents_rag.tools import tool
+
+load_dotenv()
+import os
+
+from langchain_google_genai import ChatGoogleGenerativeAI
+
+## call the gemini models
+llm = ChatGoogleGenerativeAI(
+    model="gemini-2.0-flash-exp",
+    verbose=True,
+    temperature=0.5,
+    google_api_key=os.getenv("GEMINI_API_KEY"),
+)
+
+# Creating a senior researcher agent with memory and verbose mode
+
+news_researcher = Agent(
+    role="Senior Researcher",
+    goal="Unccover ground breaking technologies in {topic}",
+    verbose=True,
+    memory=True,
+    backstory=(
+        "Driven by curiosity, you're at the forefront of"
+        "innovation, eager to explore and share knowledge that could change"
+        "the world."
+    ),
+    tools=[tool],
+    llm=llm,
+    allow_delegation=True,
+)
+
+## creating a write agent with custom tools responsible in writing news blog
+
+news_writer = Agent(
+    role="Writer",
+    goal="Narrate compelling tech stories about {topic}",
+    verbose=True,
+    memory=True,
+    backstory=(
+        "With a flair for simplifying complex topics, you craft"
+        "engaging narratives that captivate and educate, bringing new"
+        "discoveries to light in an accessible manner."
+    ),
+    tools=[tool],
+    llm=llm,
+    allow_delegation=False,
+)
+
+writer_rag = Agent(
+    role="Query Answerer using Web Search",
+    goal="Create comprehensive and well-structured answers to given queries based on web search content.",
+    backstory=(
+        "Specialized in synthesizing information from multiple sources into "
+        "coherent and engaging content while maintaining accuracy."
+    ),
+    tools=[tool],  # Ensure 'tool' is suitable for web search tasks
+    llm=llm,  # Ensure 'llm' is configured for this agent's needs
+    verbose=True,
+    memory=True,
+    allow_delegation=False,
+)
+
+rag_agent = Agent(
+    role="RAG agent that answer of given query in quick manner along with citation using provided context only don't search on only of you don't found relvatent context then denie to give answer and also answer give in given format",
+    goal="Create comprehensive and well-structured answers to given queries with citation based on given context only and also give the answer in given format.",
+    backstory=(
+        "Specialized in synthesizing information from multiple sources into "
+        "coherent and engaging content while maintaining accuracy."
+    ),  # Ensure 'tool' is suitable for web search tasks
+    llm=llm,  # Ensure 'llm' is configured for this agent's needs
+    verbose=True,
+    memory=True,
+    allow_delegation=False,
+)

src/agents_rag/crew.py

@@ -0,0 +1,52 @@
+from crewai import Crew, Process
+
+from agents_rag.agents import rag_agent, writer_rag
+from agents_rag.tasks import rag_task, write_rag_task
+
+
+def get_crew_response(query=None, context=None, url=None):
+    if url != "None":
+        print("in crwe url")
+        ## Forming the tech focused crew with some enhanced configuration
+        crew_search = Crew(
+            agents=[writer_rag],
+            tasks=[write_rag_task],
+            process=Process.sequential,
+        )
+
+        ## starting the task execution process wiht enhanced feedback
+        result_web = crew_search.kickoff(
+            inputs={
+                "query": f"{query}",
+                "url": f"{url}",
+            }
+        )
+        print(result_web)
+
+        result_web = result_web.removeprefix("```json")
+        result_web = result_web.removesuffix("```")
+        result_web = result_web.replace("```", "")
+
+        return result_web
+    else:
+        print("In crew else")
+        ## Forming the tech focused crew with some enhanced configuration
+        crew_rag = Crew(
+            agents=[rag_agent],
+            tasks=[rag_task],
+            process=Process.sequential,
+        )
+
+        ## starting the task execution process wiht enhanced feedback
+        result_rag = crew_rag.kickoff(
+            inputs={
+                "query": f"{query}",
+                "context": f"{context}",
+            }
+        )
+
+        result_rag = result_rag.removeprefix("```json")
+        result_rag = result_rag.removesuffix("```")
+        result_rag = result_rag.replace("```", "")
+
+        return result_rag

src/agents_rag/tasks.py

@@ -0,0 +1,60 @@
+from crewai import Task
+
+from agents_rag.agents import news_researcher, news_writer, rag_agent, writer_rag
+from agents_rag.tools import tool
+
+# Research task
+research_task = Task(
+    description=(
+        "Identify the next big trend in {topic}."
+        "Focus on identifying pros and cons and the overall narrative."
+        "Your final report should clearly articulate the key points,"
+        "its market opportunities, and potential risks."
+    ),
+    expected_output="A comprehensive 3 paragraphs long report on the latest AI trends.",
+    tools=[tool],
+    agent=news_researcher,
+)
+
+# Writing task with language model configuration
+write_task = Task(
+    description=(
+        "Compose an insightful article on {topic}."
+        "Focus on the latest trends and how it's impacting the industry."
+        "This article should be easy to understand, engaging, and positive."
+    ),
+    expected_output="A 4 paragraph article on {topic} advancements formatted as markdown.",
+    tools=[tool],
+    agent=news_writer,
+    async_execution=False,
+    # output_file="new-blog-post.md",  # Example of output customization
+)
+
+write_rag_task = Task(
+    description=(
+        "Answer the Give query: {query} like RAG application from given url: {url} only. if user manention url then use only that url to answer the query otherwise you can do web-search to answer the qurey. "
+        "Focus on the latest trends."
+        "The answer should be easy to understand, engaging, and positive."
+        "Also Show source of content That use to generate generate answer."
+        "All source are also show in citation and each citation have number and in end there is map of citation number and source."
+    ),
+    expected_output="A Give normal length answer if user asked something specific like small or big or more this some type of word then give answer according to that length on {query} advancements formatted as only one JSON format. in which it has two field first is Answer in which generated answer is written and second field is context in which the context value that are used to for generation of answer and third is citations. If required, then give in table format.",
+    agent=writer_rag,
+    tools=[tool],  # Ensure 'tool' is appropriate for the task
+    async_execution=False,
+    output_file="new-blog-post.md",  # Specify the desired output file path
+)
+
+rag_task = Task(
+    description=(
+        "Answer the Give query: {query} like RAG application of using given context: {context} "
+        "Focus on the latest trends."
+        "The answer should be easy to understand, engaging, and positive."
+        "Also Show source of content That use to generate generate answer."
+        "All source are also show in citation and each citation have number and in end there is map of citation number and source."
+    ),
+    expected_output="A Give normal length answer if user asked something specific like small or big or or more this some type of word then give answer according to that length on {query} advancements formatted as JSON only in which it has three field first is Answer in which generated answer is written and second field is context in which the context value that are used to for generation of answer and third is citations. If required, then give in table format.",
+    agent=rag_agent,
+    async_execution=False,
+    # output_file="new-blog-post.md",  # Specify the desired output file path
+)

src/agents_rag/tools.py

@@ -0,0 +1,14 @@
+## https://serper.dev/
+
+from dotenv import load_dotenv
+
+load_dotenv()
+import os
+
+os.environ["SERPER_API_KEY"] = os.getenv("SERPER_API_KEY")
+
+
+from crewai_tools import SerperDevTool
+
+# Initialize the tool for internet searching capabilities
+tool = SerperDevTool()

src/app.py

@@ -0,0 +1,476 @@
+# import json
+# import os
+# import re
+# import time
+# from pathlib import Path
+# from typing import Dict, List
+
+# import requests
+# import streamlit as st
+
+
+# def extract_and_verify_url(string):
+#     """Extract the URL from the string and verify if it points to content."""
+#     url_pattern = re.compile(
+#         r"http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+"
+#     )
+#     match = url_pattern.search(string)
+#     if match:
+#         url = match.group()
+#         try:
+#             response = requests.head(url, allow_redirects=True, timeout=5)
+#             if response.status_code == 200:
+#                 return {
+#                     "url": url,
+#                     "status": "Valid and content exists",
+#                     "status_code": 200,
+#                 }
+#             else:
+#                 return {
+#                     "url": url,
+#                     "status": f"Invalid (HTTP {response.status_code})",
+#                     "status_code": response.status_code,
+#                 }
+#         except requests.RequestException as e:
+#             return {"url": url, "status": f"Error: {e}"}
+#     return {"url": "", "status": "No URL found"}
+
+
+# def stream_data(data_val):
+#     for word in data_val.split(" "):
+#         yield word + " "
+#         time.sleep(0.02)
+
+
+# def list_files_in_directory(directory):
+#     """List all files in the given directory."""
+#     try:
+#         return os.listdir(directory)
+#     except FileNotFoundError:
+#         return []
+
+
+# def save_uploaded_file(uploaded_file, directory):
+#     """Save the uploaded file to the specified directory."""
+#     with open(os.path.join(directory, uploaded_file.name), "wb") as f:
+#         f.write(uploaded_file.getbuffer())
+
+#         return os.path.join(directory, uploaded_file.name)
+
+
+# def call_rag_api(
+#     query: str,
+#     url: str,
+#     is_uploaded: bool = False,
+# ) -> Dict:
+#     """Call the RAG API and get a response."""
+#     endpoint = f"http://127.0.0.1:8000/get-response"
+#     payload = {"query": query, "is_uploaded": is_uploaded, "url": url}
+
+#     try:
+#         response = requests.post(endpoint, json=payload)
+#         response.raise_for_status()
+#         result = response.json()
+#         print(type(result))
+#         print(result)
+#         return {
+#             "status": "success",
+#             "response": result["response"],
+#             "context": result["context"],
+#             "citations": result["citations"],
+#         }
+#     except requests.exceptions.RequestException as e:
+#         return {"status": "error", "message": str(e)}
+
+
+# # Main Streamlit app
+# def main():
+#     st.title("🤖 RAG Chat Assistant")
+
+#     # Sidebar inputs and actions
+#     with st.sidebar:
+#         st.header("📚 Document Control")
+
+#         # Input directory
+#         directory = st.text_input("Enter the directory path:", value="data")
+
+#         # Ensure directory exists
+#         Path(directory).mkdir(parents=True, exist_ok=True)
+
+#         # Display files in the directory
+#         st.subheader("Files in Directory")
+#         files = list_files_in_directory(directory)
+#         if files:
+#             st.write(files)
+#         else:
+#             st.write("No files found.")
+
+#         # Upload file
+#         st.subheader("Upload a File")
+#         uploaded_file = st.file_uploader(
+#             "Choose a file", type=["txt", "pdf", "doc", "docx", "mp3", "mp4"]
+#         )
+#         if uploaded_file:
+#             file_path = save_uploaded_file(uploaded_file, directory)
+#             with st.spinner:
+#                 endpoint = f"http://127.0.0.1:8000/process-file"
+#                 payload = {"file_path": file_path}
+#                 result = requests.post(endpoint, json=payload)
+#                 st.success(result["response"])
+#             st.success(f"File '{uploaded_file.name}' uploaded successfully!")
+
+#         # Delete a file
+#         st.subheader("Delete a File")
+#         if files:
+#             file_to_delete = st.selectbox("Select a file to delete:", options=files)
+#             if st.button("Delete File"):
+#                 try:
+#                     os.remove(os.path.join(directory, file_to_delete))
+#                     st.success(f"File '{file_to_delete}' deleted successfully!")
+#                 except Exception as e:
+#                     st.error(f"Error deleting file: {e}")
+
+#         # Chat system status
+#         st.divider()
+#         st.markdown("### System Status")
+#         if uploaded_file:
+#             st.success("Document loaded")
+#         else:
+#             st.info("No document uploaded")
+
+#     # Initialize chat history
+#     if "messages" not in st.session_state:
+#         st.session_state.messages = []
+
+#     # Display chat messages in streaming manner
+#     chat_placeholder = st.container()
+
+#     for message in st.session_state.messages:
+#         with chat_placeholder.container():
+#             with st.chat_message(message["role"]):
+#                 st.markdown(message["content"])
+#                 if message["role"] == "assistant" and "context" in message:
+#                     with st.expander("View source context"):
+#                         st.info(message["context"])
+
+#     # Chat input
+#     if prompt := st.chat_input("Ask me anything about your documents..."):
+#         res = extract_and_verify_url(prompt)
+#         print(res)
+#         if res["url"] != None:
+#             print(res["url"])
+#         st.session_state.messages.append({"role": "user", "content": prompt})
+#         with chat_placeholder.container():
+#             with st.chat_message("user"):
+#                 st.markdown(prompt)
+
+#         with chat_placeholder.container():
+#             with st.chat_message("assistant"):
+#                 with st.spinner("Thinking..."):
+#                     result = call_rag_api(
+#                         url=res["url"],
+#                         query=prompt,
+#                         is_uploaded=uploaded_file is not None,
+#                     )
+
+#                     if result["status"] == "success":
+
+#                         response_content = result["response"]
+#                         context = result["context"]
+#                         citations = result["citations"]
+
+#                         # st.markdown(response_content)
+#                         st.write_stream(stream_data(response_content))
+#                         with st.expander("View source context"):
+#                             st.json(citations)
+
+#                         st.session_state.messages.append(
+#                             {
+#                                 "role": "assistant",
+#                                 "content": response_content,
+#                                 "context": context,
+#                                 "citations": citations,
+#                             }
+#                         )
+#                     else:
+#                         st.error(
+#                             f"Error: {result.get('message', 'Unknown error occurred')}"
+#                         )
+
+
+# if __name__ == "__main__":
+#     main()
+
+
+import json
+import os
+import re
+import time
+from pathlib import Path
+from typing import Dict, List
+
+import requests
+import streamlit as st
+from streamlit_js_eval import streamlit_js_eval
+
+
+def extract_and_verify_url(string):
+    """Extract the URL from the string and verify if it points to content."""
+    url_pattern = re.compile(
+        r"http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+"
+    )
+    match = url_pattern.search(string)
+    if match:
+        url = match.group()
+        try:
+            response = requests.head(url, allow_redirects=True, timeout=5)
+            if response.status_code == 200:
+                return {
+                    "url": url,
+                    "status": "Valid and content exists",
+                    "status_code": 200,
+                }
+            else:
+                return {
+                    "url": url,
+                    "status": f"Invalid (HTTP {response.status_code})",
+                    "status_code": response.status_code,
+                }
+        except requests.RequestException as e:
+            return {"url": url, "status": f"Error: {e}"}
+    return {"url": "", "status": "No URL found"}
+
+
+def stream_data(data_val):
+    for word in data_val.split(" "):
+        yield word + " "
+        time.sleep(0.02)
+
+
+def list_files_in_directory(directory):
+    """List all files in the given directory."""
+    try:
+        return os.listdir(directory)
+    except FileNotFoundError:
+        return []
+
+
+def save_uploaded_file(uploaded_file, directory):
+    """Save the uploaded file to the specified directory."""
+    with open(os.path.join(directory, uploaded_file.name), "wb") as f:
+        f.write(uploaded_file.getbuffer())
+        return os.path.join(directory, uploaded_file.name)
+
+
+def call_rag_api(query: str, url: str, is_uploaded: bool = False) -> Dict:
+    """Call the RAG API and get a response."""
+    endpoint = f"http://127.0.0.1:8000/get-response"
+    payload = {"query": query, "is_uploaded": is_uploaded, "url": url}
+
+    try:
+        response = requests.post(endpoint, json=payload)
+        response.raise_for_status()
+        result = response.json()
+        return {
+            "status": "success",
+            "response": result["response"],
+            "context": result["context"],
+            "citations": result["citations"],
+        }
+    except requests.exceptions.RequestException as e:
+        return {"status": "error", "message": str(e)}
+
+
+def call_llm_api(query: str) -> Dict:
+    """Call the LLM API for answering questions."""
+    endpoint = f"http://127.0.0.1:8000/llm-response"
+    payload = {"query": query}
+
+    try:
+        response = requests.post(endpoint, json=payload)
+        response.raise_for_status()
+        result = response.json()
+        return {"status": "success", "response": result["response"]}
+    except requests.exceptions.RequestException as e:
+        return {"status": "error", "message": str(e)}
+
+
+# Main Streamlit app
+def main():
+    st.title("🤖 Multi-Functional Chat Assistant")
+
+    # Sidebar inputs and actions
+    with st.sidebar:
+
+        # Chat functionality selection
+        mode = st.radio("Select Mode:", ["LLM Answering", "Web Search Agent", "RAG"])
+
+        st.header("📂 Document Control")
+
+        # Input directory
+        directory = st.text_input("Enter the directory path:", value="data")
+
+        # Ensure directory exists
+        Path(directory).mkdir(parents=True, exist_ok=True)
+
+        # Display files in the directory
+        st.subheader("Files in Directory")
+        files = list_files_in_directory(directory)
+        if files:
+            st.write(files)
+        else:
+            st.write("No files found.")
+
+        # Upload file
+        st.subheader("Upload a File")
+        uploaded_file = st.file_uploader(
+            "Choose a file", type=["txt", "pdf", "doc", "docx", "mp3", "mp4"]
+        )
+        if uploaded_file:
+            file_path = save_uploaded_file(uploaded_file, directory)
+            endpoint = f"http://127.0.0.1:8000/process-file"
+            payload = {"file_path": file_path}
+
+            with st.spinner("File is in process..."):
+                response = requests.post(endpoint, json=payload)
+
+            st.success(f"File '{uploaded_file.name}' uploaded successfully!")
+            time.sleep(3)
+            streamlit_js_eval(js_expressions="parent.window.location.reload()")
+
+        # Delete a file
+        st.subheader("Delete a File")
+        if files:
+            file_to_delete = st.selectbox("Select a file to delete:", options=files)
+            if st.button("Delete File"):
+                try:
+                    payload = {"file_path": file_to_delete}
+                    endpoint = f"http://127.0.0.1:8000/delete-file"
+
+                    with st.spinner("File is deleting..."):
+                        response = requests.post(endpoint, json=payload)
+                        os.remove(os.path.join(directory, file_to_delete))
+
+                    st.success(f"File '{file_to_delete}' deleted successfully!")
+                    time.sleep(3)
+                    streamlit_js_eval(js_expressions="parent.window.location.reload()")
+                except Exception as e:
+                    st.error(f"Error deleting file: {e}")
+
+        # Chat system status
+        st.divider()
+        st.markdown("### System Status")
+        if uploaded_file:
+            st.success("Document loaded")
+        else:
+            st.info("No document uploaded")
+
+    # Initialize chat history
+    if "messages" not in st.session_state:
+        st.session_state.messages = []
+
+    # Display chat messages in streaming manner
+    chat_placeholder = st.container()
+
+    for message in st.session_state.messages:
+        with chat_placeholder.container():
+            with st.chat_message(message["role"]):
+                st.markdown(message["content"])
+
+    # # Chat functionality selection
+    # mode = st.radio("Select Mode:", ["LLM Answering", "Web Search Agent", "RAG"])
+
+    # Chat input
+    if prompt := st.chat_input("Ask me anything..."):
+        st.session_state.messages.append({"role": "user", "content": prompt})
+        with chat_placeholder.container():
+            with st.chat_message("user"):
+                st.markdown(prompt)
+
+        with chat_placeholder.container():
+            with st.chat_message("assistant"):
+                with st.spinner("Thinking..."):
+                    citations = []
+                    if mode == "LLM Answering":
+                        result = call_llm_api(prompt)
+                        if result["status"] == "success":
+                            response_content = result.get("response", "")
+                            st.write_stream(stream_data(response_content))
+
+                            st.session_state.messages.append(
+                                {"role": "assistant", "content": response_content}
+                            )
+
+                        else:
+                            st.error(
+                                f"Error: {result.get('message', 'Unknown error occurred')}"
+                            )
+                    elif mode == "Web Search Agent":
+                        res = extract_and_verify_url(prompt)
+                        result = call_rag_api(
+                            url=res.get("url", ""),
+                            query=prompt,
+                            is_uploaded=uploaded_file is not None,
+                        )
+                        if result["status"] == "success":
+
+                            response_content = result["response"]
+                            context = result["context"]
+                            citations = result["citations"]
+
+                            # st.markdown(response_content)
+                            st.write_stream(stream_data(response_content))
+                            # with st.expander("View source context"):
+                            #     st.json(citations)
+
+                            st.session_state.messages.append(
+                                {
+                                    "role": "assistant",
+                                    "content": response_content,
+                                    "context": context,
+                                    "citations": citations,
+                                }
+                            )
+                        else:
+                            st.error(
+                                f"Error: {result.get('message', 'Unknown error occurred')}"
+                            )
+
+                    elif mode == "RAG":
+                        res = extract_and_verify_url(prompt)
+                        result = call_rag_api(
+                            url="None",
+                            query=prompt,
+                            is_uploaded=uploaded_file is not None,
+                        )
+
+                        if result["status"] == "success":
+
+                            response_content = result["response"]
+                            context = result["context"]
+                            citations = result["citations"]
+
+                            # st.markdown(response_content)
+                            st.write_stream(stream_data(response_content))
+                            # with st.expander("View source context"):
+                            #     st.json(citations)
+
+                            st.session_state.messages.append(
+                                {
+                                    "role": "assistant",
+                                    "content": response_content,
+                                    "context": context,
+                                    "citations": citations,
+                                }
+                            )
+                        else:
+                            st.error(
+                                f"Error: {result.get('message', 'Unknown error occurred')}"
+                            )
+
+                    with st.expander("View source context"):
+                        st.json(citations)
+
+
+if __name__ == "__main__":
+    main()

src/assistants/assistant_v1.py

@@ -0,0 +1,266 @@
+# import os
+# from typing import List
+
+# import google.generativeai as genai
+# from dotenv import load_dotenv
+# from langchain.prompts import PromptTemplate
+# from langchain.schema import StrOutputParser
+# from pydantic import BaseModel, Field
+
+# from .gemini_assistant import GeminiAssistant
+
+# load_dotenv()
+
+# api_key = os.getenv("GEMINI_API_KEY")
+# model_name = "gemini-2.0-flash-exp"
+
+
+# class GenerativeAnswer(BaseModel):
+#     answer: str = Field(
+#         ...,
+#         description="""The answer to the user question, which is based only on the given sources.\n
+#         example: Artificial neural networks were inspired by information processing and distributed communication nodes in biological systems. ANNs have various differences from biological brains. Specifically, artificial neural networks tend to be static and symbolic, while the biological brain of most living organisms is dynamic (plastic) and analog (vec-001). However, ANNs have demonstrated significant capabilities in tasks such as pattern recognition, classification, and prediction, making them valuable tools in various fields (vec-002). Despite their effectiveness in certain applications, ANNs still lack the complexity and adaptability of biological brains, leading some researchers to view them as low-quality models for brain function (vec-003). Furthermore, advancements in neuroscience continue to provide insights into the workings of the brain, which may further differentiate artificial neural networks from their biological counterparts (vec-004). However, ANNs remain a powerful computational tool, offering practical solutions to a wide range of problems, albeit with inherent limitations compared to biological systems (vec-005) and so on.....
+#         """,
+#     )
+#     source_id: List[str] = Field(
+#         ...,
+#         description="The ID of a SPECIFIC source which justifies the answer.",
+#     )
+#     quote: List[str] = Field(
+#         ...,
+#         description="The VERBATIM quote from the specified source that justifies the answer.",
+#     )
+#     related_questions: List[str] = Field(
+#         ...,
+#         description="Related questions based on the user question and generated answer.",
+#     )
+
+#     # [INST]<<SYS>>You are an Intelligent Assistant for question-answering tasks.
+#     # Your knowledge base consists of documents related to Convene and its resources.
+#     # Convene is a community of Christian business leaders who gather monthly in advisory teams worldwide to share insights, face challenges, and grow in faith through executive learning. Members can join or lead a Convene Team, participate in One2One™ Coaching, and access exclusive events. The organization, founded in 1996 by Brian Thatcher and others from Saddleback Church, aims to help CEOs integrate faith with business to build profitable, faith-based enterprises. Services include peer advisory, executive coaching, and business consulting, focusing on enhancing business and leadership skills while maintaining a faith foundation. Convene emphasizes balancing business success with spiritual growth and family leadership.
+
+#     # Your [Task] is to answer questions accurately and concisely based on the context provided. You will strictly discuss topics within the scope of the context provided and Convene's activities, history, services, and community impacts as outlined.
+#     # Identify if the User question aligns with the context or Convene's interests. If the information requested is outside of the context or Convene's provided data or knowledge base, You should politely decline the request and tell the user to ask related questions.
+#     # Only provide answers based on the information available in your context and the above topics.
+
+#     # Use the following steps and pieces of retrieved context to answer the question. The answer should be detailed and concise.
+
+
+# prompt = """
+#     You are a friendly and helpful assistant.
+#     Ensure your answers are complete, unless the user requests a more concise approach.
+#     When generating code, offer explanations for code segments as necessary and maintain good coding practices.
+#     When presented with inquiries seeking information, provide answers that reflect a deep understanding of the field, guaranteeing their correctness.
+#     For any non-english queries, respond in the same language as the prompt unless otherwise specified by the user.
+#     For prompts involving reasoning, provide a clear explanation of each step in the reasoning process before presenting the final answer.
+
+#     [Task]
+#     Your task is to answer questions accurately and concisely based on the context provided. You will strictly discuss topics within the scope of the provided context, including its activities, history, services, and impacts.
+#     Assess whether the user’s question aligns with the context or the specified areas of interest.
+#     If the information requested falls outside the provided context or scope, politely decline the request and advise the user to ask relevant questions within the given context.
+#     Only provide answers based on the information available in the given context and the specified areas of interest.
+
+#     The answer should look like:
+#     Example: Artificial neural networks were inspired by information processing and distributed communication nodes in biological systems. ANNs have various differences from biological brains. Specifically, artificial neural networks tend to be static and symbolic, while the biological brain of most living organisms is dynamic (plastic) and analog (vec-001). However, ANNs have demonstrated significant capabilities in tasks such as pattern recognition, classification, and prediction, making them valuable tools in various fields (vec-002). Despite their effectiveness in certain applications, ANNs still lack the complexity and adaptability of biological brains, leading some researchers to view them as low-quality models for brain function (vec-003). Furthermore, advancements in neuroscience continue to provide insights into the workings of the brain, which may further differentiate artificial neural networks from their biological counterparts (vec-004). However, ANNs remain a powerful computational tool, offering practical solutions to a wide range of problems, albeit with inherent limitations compared to biological systems (vec-005) and so on.....
+#     means your answer must contains source_id in given format.
+#     Output should follow the pattern defined in schema and the output should be in json format only so that it can be directly used with json.loads()
+#     <</SYS>>
+#     steps:
+#     1. use the provided data.
+#     2. Check if the question is related to Convene or related to the context provided. If the question is not related, ask them politely to ask related questions. Do not follow further steps but use the wrong_question_schema to output the answer.
+#     3. use one source and extract the sentences from it that justify the answer all the source sentences are in must be unique.
+#     4. use the sentences to generate the answer and append its source_id to the answer with prefix "vec-". You must append its source_id only if it exists. Do not hallucinate or add from your imagination. Only add source_id which you get from the context. This is a strict requirement.
+#     5. use other source and extract the sentences from it that justify the answer. NOTE: use atleast 3 sources to generate the answer.
+#     6. If the user has asked question related to the context or Convene, use the whole answer and user's question to generate 3 related questions. If the question is outside the context, do not generate related questions.
+
+#     NOTE: Make sure no out of bounds questions or questions that are not in the context or knowledge base are answered. This is mandatory that no out of context questions are answered."""
+
+# gemini_rag_assistant = GeminiAssistant(
+#     model=model_name,
+#     api_key=api_key,
+#     system_prompt=prompt,
+#     output_model=GenerativeAnswer,
+# )
+
+
+# import os
+# from typing import List
+
+# import google.generativeai as genai
+# from dotenv import load_dotenv
+# from langchain.prompts import PromptTemplate
+# from langchain.schema import StrOutputParser
+# from pydantic import BaseModel, Field
+
+# from .gemini_assistant import GeminiAssistant
+
+# # Load environment variables
+# load_dotenv()
+
+# # Load the API key from the environment
+# api_key = os.getenv("GEMINI_API_KEY")
+# model_name = "gemini-2.0-flash-exp"
+
+
+# # Define the output model schema using Pydantic
+# class GenerativeAnswer(BaseModel):
+#     answer: str = Field(
+#         ...,
+#         description="""The answer to the user question, which is based only on the given sources.
+#         Example:
+#         Artificial neural networks were inspired by information processing and distributed communication nodes in biological systems.
+#         ANNs have various differences from biological brains. Specifically, artificial neural networks tend to be static and symbolic,
+#         while the biological brain of most living organisms is dynamic (plastic) and analog (vec-001). However, ANNs have demonstrated
+#         significant capabilities in tasks such as pattern recognition, classification, and prediction, making them valuable tools in various fields
+#         (vec-002). Despite their effectiveness in certain applications, ANNs still lack the complexity and adaptability of biological brains,
+#         leading some researchers to view them as low-quality models for brain function (vec-003). Furthermore, advancements in neuroscience
+#         continue to provide insights into the workings of the brain, which may further differentiate artificial neural networks from their biological counterparts
+#         (vec-004). However, ANNs remain a powerful computational tool, offering practical solutions to a wide range of problems, albeit with inherent limitations
+#         compared to biological systems (vec-005).""",
+#     )
+#     source_id: List[str] = Field(
+#         ...,
+#         description="The ID of a SPECIFIC source which justifies the answer.",
+#     )
+#     quote: List[str] = Field(
+#         ...,
+#         description="The VERBATIM quote from the specified source that justifies the answer.",
+#     )
+#     related_questions: List[str] = Field(
+#         ...,
+#         description="Related questions based on the user question and generated answer.",
+#     )
+
+
+# # Define the system prompt
+# prompt = """
+#     You are a friendly and helpful assistant.
+#     Ensure your answers are complete, unless the user requests a more concise approach.
+#     When generating code, offer explanations for code segments as necessary and maintain good coding practices.
+#     When presented with inquiries seeking information, provide answers that reflect a deep understanding of the field, guaranteeing their correctness.
+#     For any non-English queries, respond in the same language as the prompt unless otherwise specified by the user.
+#     For prompts involving reasoning, provide a clear explanation of each step in the reasoning process before presenting the final answer.
+
+#     [Task]
+#     Your task is to answer questions accurately and concisely based on the context provided. You will strictly discuss topics within the scope of the provided context, including its activities, history, services, and impacts.
+#     Assess whether the user’s question aligns with the context or the specified areas of interest.
+#     If the information requested falls outside the provided context or scope, politely decline the request and advise the user to ask relevant questions within the given context.
+#     Only provide answers based on the information available in the given context and the specified areas of interest.
+
+#     The answer should look like:
+#     Example: Artificial neural networks were inspired by information processing and distributed communication nodes in biological systems.
+#     ANNs have various differences from biological brains. Specifically, artificial neural networks tend to be static and symbolic,
+#     while the biological brain of most living organisms is dynamic (plastic) and analog (vec-001). However, ANNs have demonstrated
+#     significant capabilities in tasks such as pattern recognition, classification, and prediction, making them valuable tools in various fields
+#     (vec-002). Despite their effectiveness in certain applications, ANNs still lack the complexity and adaptability of biological brains,
+#     leading some researchers to view them as low-quality models for brain function (vec-003). Furthermore, advancements in neuroscience
+#     continue to provide insights into the workings of the brain, which may further differentiate artificial neural networks from their biological counterparts
+#     (vec-004). However, ANNs remain a powerful computational tool, offering practical solutions to a wide range of problems, albeit with inherent limitations
+#     compared to biological systems (vec-005).
+#     Make sure your answer contains source_id in the given format.
+
+#     Output should follow the pattern defined in the schema, and the output should be in JSON format only, so that it can be directly used with `json.loads()`.
+
+#     Steps:
+#     1. Use the provided data.
+#     2. Check if the question is related to the context or related to the provided scope. If the question is not related, politely ask the user to ask relevant questions. Do not proceed with other steps and use the wrong_question_schema to output the answer.
+#     3. Use one source and extract the sentences from it that justify the answer. All source sentences must be unique.
+#     4. Use the sentences to generate the answer and append its source_id to the answer with prefix "vec-". You must append its source_id only if it exists. Do not hallucinate or add from your imagination. Only add source_id which you get from the context. This is a strict requirement.
+#     5. Use other sources and extract the sentences from them that justify the answer. NOTE: Use at least 3 sources to generate the answer.
+#     6. If the user has asked a question related to the context or scope, use the entire answer and the user's question to generate 3 related questions. If the question is outside the context, do not generate related questions.
+
+#     NOTE: Ensure that no out-of-bounds questions or questions outside the context are answered. It is mandatory that no out-of-context questions are addressed.
+# """
+
+# # Instantiate the Gemini RAG assistant
+# gemini_rag_assistant = GeminiAssistant(
+#     model=model_name,
+#     api_key=api_key,
+#     system_prompt=prompt,
+#     output_model=GenerativeAnswer,
+# )
+
+import os
+from typing import List
+
+from dotenv import load_dotenv
+from pydantic import BaseModel, Field
+
+from .gemini_assistant import GeminiAssistant
+
+# Load environment variables
+load_dotenv()
+
+# Load the API key from the environment
+api_key = os.getenv("GEMINI_API_KEY")
+model_name = "gemini-2.0-flash-exp"
+
+
+# Define the output model schema using Pydantic
+class GenerativeAnswer(BaseModel):
+    answer: str = Field(
+        ...,
+        description="""The answer to the user question, which is based only on the given sources.
+        Example:
+        Artificial neural networks were inspired by information processing and distributed communication nodes in biological systems. 
+        ANNs have various differences from biological brains. Specifically, artificial neural networks tend to be static and symbolic, 
+        while the biological brain of most living organisms is dynamic (plastic) and analog (vec-001). However, ANNs have demonstrated 
+        significant capabilities in tasks such as pattern recognition, classification, and prediction, making them valuable tools in various fields 
+        (vec-002). Despite their effectiveness in certain applications, ANNs still lack the complexity and adaptability of biological brains, 
+        leading some researchers to view them as low-quality models for brain function (vec-003). Furthermore, advancements in neuroscience 
+        continue to provide insights into the workings of the brain, which may further differentiate artificial neural networks from their biological counterparts 
+        (vec-004). However, ANNs remain a powerful computational tool, offering practical solutions to a wide range of problems, albeit with inherent limitations 
+        compared to biological systems (vec-005).""",
+    )
+    related_questions: List[str] = Field(
+        ...,
+        description="Related questions based on the user question and generated answer.",
+    )
+
+
+# Define the system prompt using Langchain's PromptTemplate
+prompt_template = """
+    You are a friendly and helpful assistant.
+    Ensure your answers are complete, unless the user requests a more concise approach.
+    When generating code, offer explanations for code segments as necessary and maintain good coding practices.
+    When presented with inquiries seeking information, provide answers that reflect a deep understanding of the field, guaranteeing their correctness.
+    For any non-English queries, respond in the same language as the prompt unless otherwise specified by the user.
+    For prompts involving reasoning, provide a clear explanation of each step in the reasoning process before presenting the final answer.
+
+    [Task]
+    Your task is to answer questions accurately and concisely based on the context provided. You will strictly discuss topics within the scope of the provided context, including its activities, history, services, and impacts.
+    Assess whether the user’s question aligns with the context or the specified areas of interest.
+    If the information requested falls outside the provided context or scope, politely decline the request and advise the user to ask relevant questions within the given context.
+    Only provide answers based on the information available in the given context and the specified areas of interest.
+
+    The answer should look like:
+    Example: Artificial neural networks were inspired by information processing and distributed communication nodes in biological systems. 
+    ANNs have various differences from biological brains. Specifically, artificial neural networks tend to be static and symbolic, 
+    while the biological brain of most living organisms is dynamic (plastic) and analog (vec-001). However, ANNs have demonstrated 
+    significant capabilities in tasks such as pattern recognition, classification, and prediction, making them valuable tools in various fields 
+    (vec-002). Despite their effectiveness in certain applications, ANNs still lack the complexity and adaptability of biological brains, 
+    leading some researchers to view them as low-quality models for brain function (vec-003). Furthermore, advancements in neuroscience 
+    continue to provide insights into the workings of the brain, which may further differentiate artificial neural networks from their biological counterparts 
+    (vec-004). However, ANNs remain a powerful computational tool, offering practical solutions to a wide range of problems, albeit with inherent limitations 
+    compared to biological systems (vec-005).
+
+    The final output should be structured like this:
+    1. The complete answer to the user question.
+    2. The source IDs appended to the answer, e.g., (vec-001, vec-002, vec-003).
+    3. Related questions that might arise from the user's question.
+
+    Output should be in the following format:
+    {
+      "answer": "<generated_answer_with_sources>",
+      "related_questions": [<related_question_1>, <related_question_2>, <related_question_3>]
+    }
+"""
+
+
+# Instantiate the Gemini RAG assistant with the prompt template and output parser
+gemini_rag_assistant = GeminiAssistant(
+    model=model_name,
+    api_key=api_key,
+    system_prompt=prompt_template,
+    output_model=GenerativeAnswer,
+)

src/assistants/gemini_assistant.py

@@ -0,0 +1,481 @@
+# import json
+# import os
+# from dataclasses import dataclass
+# from enum import Enum
+# from typing import Any, Dict, List, Optional, Union
+
+# import google.generativeai as genai
+# from pydantic import BaseModel
+
+
+# class Role(Enum):
+#     USER = "user"
+#     ASSISTANT = "assistant"
+#     SYSTEM = "system"
+
+
+# @dataclass
+# class Message:
+#     role: Role
+#     content: str
+
+
+# class GeminiAssistant:
+#     def __init__(
+#         self,
+#         api_key: str,
+#         model: str = "gemini-pro",
+#         temperature: float = 0.7,
+#         top_p: float = 0.95,
+#         top_k: int = 40,
+#         max_output_tokens: int = 2048,
+#         system_prompt: Optional[str] = None,
+#         output_model: Optional[type[BaseModel]] = None,
+#     ):
+#         genai.configure(api_key=api_key)
+
+#         self.model = genai.GenerativeModel(
+#             model_name=model,
+#             system_instruction=system_prompt,
+#             generation_config={
+#                 "temperature": temperature,
+#                 "top_p": top_p,
+#                 "top_k": top_k,
+#                 "max_output_tokens": max_output_tokens,
+#                 "response_mime_type": "application/json",
+#             },
+#         )
+
+#         self.system_prompt = system_prompt
+#         self.output_model = output_model
+#         self.conversation_history: List[Message] = []
+#         if system_prompt:
+#             self.conversation_history.append(
+#                 Message(role=Role.SYSTEM, content=system_prompt)
+#             )
+
+#     def add_message(self, role: Union[Role, str], content: str) -> None:
+#         if isinstance(role, str):
+#             role = Role(role)
+#         self.conversation_history.append(Message(role=role, content=content))
+
+#     def _format_messages_for_gemini(self) -> List[Dict]:
+#         """Convert internal message format to Gemini's expected format."""
+#         formatted_history = []
+
+#         for msg in self.conversation_history:
+#             if msg.role == Role.USER:
+#                 formatted_history.append(
+#                     {"role": "user", "parts": [{"text": msg.content}]}
+#                 )
+#             elif msg.role == Role.ASSISTANT:
+#                 formatted_history.append(
+#                     {"role": "model", "parts": [{"text": msg.content}]}
+#                 )
+#             # System messages are handled differently
+
+#         return formatted_history
+
+#     async def get_response(
+#         self,
+#         message: str,
+#         context_data: Optional[str] = None,
+#     ) -> Union[str, BaseModel]:
+#         """Get a response from the assistant for the given message."""
+#         # Prepare the full message with system prompt if it exists
+#         full_message = message
+#         if self.system_prompt and not self.conversation_history:
+#             full_message = (
+#                 f"{self.system_prompt}\n\n{message}\n\nContext:\n{context_data}\n"
+#             )
+#         else:
+
+#             full_message = f"{message}\n\nContext:\n{context_data}\n"
+
+#         print("********************")
+#         print(full_message)
+#         print("********************")
+
+#         # Add user message to history
+#         self.add_message(Role.USER, full_message)
+
+#         try:
+#             # Create chat session with properly formatted history
+#             chat = self.model.start_chat(history=self._format_messages_for_gemini())
+
+#             # Generate response
+#             response = await chat.send_message_async(full_message)
+#             response_text = response.text
+
+#             # Handle structured output if output_model is specified
+#             if self.output_model:
+#                 try:
+#                     # Try to parse the response as JSON
+#                     response_data = json.loads(response_text)
+#                     # Validate with Pydantic model
+#                     structured_response = self.output_model(**response_data)
+#                     # Add the formatted response to history
+#                     self.add_message(
+#                         Role.ASSISTANT, json.dumps(response_data, indent=2)
+#                     )
+#                     return structured_response
+#                 except (json.JSONDecodeError, ValueError) as e:
+#                     # If parsing fails, try to fix the response format
+#                     retry_prompt = (
+#                         "Please format your previous response as a valid JSON object "
+#                         "that matches the specified schema. Include all required fields."
+#                     )
+#                     response = await chat.send_message_async(retry_prompt)
+#                     try:
+#                         print(response.text)
+#                         response_data = json.loads(response.text)
+#                         structured_response = self.output_model(**response_data)
+#                         self.add_message(
+#                             Role.ASSISTANT, json.dumps(response_data, indent=2)
+#                         )
+#                         return structured_response
+#                     except (json.JSONDecodeError, ValueError) as e:
+#                         raise ValueError(
+#                             f"Failed to generate properly formatted response: {str(e)}"
+#                         )
+#             else:
+#                 # Handle regular string response
+#                 self.add_message(Role.ASSISTANT, response_text)
+#                 return response_text
+
+#         except Exception as e:
+#             raise Exception(f"Error generating response: {str(e)}")
+
+#     def clear_history(self) -> None:
+#         """Clear the conversation history."""
+#         self.conversation_history = []
+#         if self.system_prompt:
+#             self.add_message(Role.SYSTEM, self.system_prompt)
+
+
+# # Example usage
+# async def main():
+#     # Initialize assistant
+#     assistant = GeminiAssistant(
+#         api_key="your-api-key-here", system_prompt="You are a helpful AI assistant."
+#     )
+
+#     # Get response
+#     response = await assistant.get_response("What is machine learning?")
+#     print(response)
+
+#     # Continue conversation
+#     response = await assistant.get_response("Can you give me an example?")
+#     print(response)
+
+
+# if __name__ == "__main__":
+#     import asyncio
+
+#     asyncio.run(main())
+
+
+# import json
+# import os
+# from dataclasses import dataclass
+# from enum import Enum
+# from typing import Any, Dict, List, Optional, Union
+
+# import google.generativeai as genai
+# from pydantic import BaseModel
+
+
+# class Role(Enum):
+#     USER = "user"
+#     ASSISTANT = "assistant"
+#     SYSTEM = "system"
+
+
+# @dataclass
+# class Message:
+#     role: Role
+#     content: str
+
+
+# class GeminiAssistant:
+#     def __init__(
+#         self,
+#         api_key: str,
+#         model: str = "gemini-pro",
+#         temperature: float = 0.7,
+#         top_p: float = 0.95,
+#         top_k: int = 40,
+#         max_output_tokens: int = 2048,
+#         system_prompt: Optional[str] = None,
+#         output_model: Optional[type[BaseModel]] = None,
+#     ):
+#         genai.configure(api_key=api_key)
+
+#         self.model = genai.GenerativeModel(
+#             model_name=model,
+#             system_instruction=system_prompt,
+#             generation_config={
+#                 "temperature": temperature,
+#                 "top_p": top_p,
+#                 "top_k": top_k,
+#                 "max_output_tokens": max_output_tokens,
+#                 "response_mime_type": "application/json",
+#             },
+#         )
+
+#         self.system_prompt = system_prompt
+#         self.output_model = output_model
+#         self.conversation_history: List[Message] = []
+#         if system_prompt:
+#             self.conversation_history.append(
+#                 Message(role=Role.SYSTEM, content=system_prompt)
+#             )
+
+#     def add_message(self, role: Union[Role, str], content: str) -> None:
+#         if isinstance(role, str):
+#             role = Role(role)
+#         self.conversation_history.append(Message(role=role, content=content))
+
+#     def _format_messages_for_gemini(self) -> List[Dict]:
+#         """Convert internal message format to Gemini's expected format."""
+#         formatted_history = []
+
+#         for msg in self.conversation_history:
+#             if msg.role == Role.USER:
+#                 formatted_history.append(
+#                     {"role": "user", "parts": [{"text": msg.content}]}
+#                 )
+#             elif msg.role == Role.ASSISTANT:
+#                 formatted_history.append(
+#                     {"role": "model", "parts": [{"text": msg.content}]}
+#                 )
+#             # System messages are handled differently
+
+#         return formatted_history
+
+#     async def get_response_gemini(
+#         self,
+#         message: str,
+#         context_data: Optional[str] = None,
+#     ) -> Union[str, BaseModel]:
+#         """Get a response from the assistant for the given message."""
+#         # Prepare the full message with system prompt if it exists
+#         full_message = message
+#         if self.system_prompt and not self.conversation_history:
+#             full_message = (
+#                 f"{self.system_prompt}\n\n{message}\n\nContext:\n{context_data}\n"
+#             )
+#         else:
+#             full_message = f"{message}\n\nContext:\n{context_data}\n"
+
+#         print("********************")
+#         print(full_message)
+#         print("********************")
+
+#         # Add user message to history
+#         self.add_message(Role.USER, full_message)
+
+#         try:
+#             # Create chat session with properly formatted history
+#             chat = self.model.start_chat(history=self._format_messages_for_gemini())
+
+#             # Generate response
+#             response = await chat.send_message_async(full_message)
+#             response_text = response.text
+
+#             # Handle structured output if output_model is specified
+#             if self.output_model:
+#                 try:
+#                     # Try to parse the response as JSON
+#                     response_data = json.loads(response_text)
+
+#                     # Ensure required fields are present
+#                     if "source_id" not in response_data:
+#                         response_data["source_id"] = "default_source_id"
+#                     if "quote" not in response_data:
+#                         response_data["quote"] = "default_quote"
+
+#                     # Validate with Pydantic model
+#                     structured_response = self.output_model(**response_data)
+#                     # Add the formatted response to history
+#                     self.add_message(
+#                         Role.ASSISTANT, json.dumps(response_data, indent=2)
+#                     )
+#                     return structured_response
+#                 except (json.JSONDecodeError, ValueError) as e:
+#                     # If parsing fails, try to fix the response format
+#                     retry_prompt = (
+#                         "Please format your previous response as a valid JSON object "
+#                         "that matches the specified schema. Include all required fields."
+#                     )
+#                     response = await chat.send_message_async(retry_prompt)
+#                     try:
+#                         print(response.text)
+#                         response_data = json.loads(response.text)
+
+#                         # Ensure required fields are present
+#                         if "source_id" not in response_data:
+#                             response_data["source_id"] = "default_source_id"
+#                         if "quote" not in response_data:
+#                             response_data["quote"] = "default_quote"
+
+#                         structured_response = self.output_model(**response_data)
+#                         self.add_message(
+#                             Role.ASSISTANT, json.dumps(response_data, indent=2)
+#                         )
+#                         return structured_response
+#                     except (json.JSONDecodeError, ValueError) as e:
+#                         raise ValueError(
+#                             f"Failed to generate properly formatted response: {str(e)}"
+#                         )
+#             else:
+#                 # Handle regular string response
+#                 self.add_message(Role.ASSISTANT, response_text)
+#                 return response_text
+
+#         except Exception as e:
+#             raise Exception(f"Error generating response: {str(e)}")
+
+#     def clear_history(self) -> None:
+#         """Clear the conversation history."""
+#         self.conversation_history = []
+#         if self.system_prompt:
+#             self.add_message(Role.SYSTEM, self.system_prompt)
+
+
+import json
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any, Dict, List, Optional, Union
+
+import google.generativeai as genai
+from pydantic import BaseModel
+
+
+class Role(Enum):
+    USER = "user"
+    ASSISTANT = "assistant"
+    SYSTEM = "system"
+
+
+@dataclass
+class Message:
+    role: Role
+    content: str
+
+
+class GeminiAssistant:
+    def __init__(
+        self,
+        api_key: str,
+        model: str = "gemini-pro",
+        temperature: float = 0.7,
+        top_p: float = 0.95,
+        top_k: int = 40,
+        max_output_tokens: int = 2048,
+        system_prompt: Optional[str] = None,
+        output_model: Optional[type[BaseModel]] = None,
+    ):
+        genai.configure(api_key=api_key)
+
+        self.model = genai.GenerativeModel(
+            model_name=model,
+            system_instruction=system_prompt,
+            generation_config={
+                "temperature": temperature,
+                "top_p": top_p,
+                "top_k": top_k,
+                "max_output_tokens": max_output_tokens,
+                "response_mime_type": "application/json",
+            },
+        )
+
+        self.system_prompt = system_prompt
+        self.output_model = output_model
+        self.conversation_history: List[Message] = []
+        if system_prompt:
+            self.conversation_history.append(
+                Message(role=Role.SYSTEM, content=system_prompt)
+            )
+
+    def add_message(self, role: Union[Role, str], content: str) -> None:
+        if isinstance(role, str):
+            role = Role(role)
+        self.conversation_history.append(Message(role=role, content=content))
+
+    def _format_messages_for_gemini(self) -> List[Dict]:
+        """Convert internal message format to Gemini's expected format."""
+        formatted_history = []
+
+        for msg in self.conversation_history:
+            if msg.role == Role.USER:
+                formatted_history.append(
+                    {"role": "user", "parts": [{"text": msg.content}]}
+                )
+            elif msg.role == Role.ASSISTANT:
+                formatted_history.append(
+                    {"role": "model", "parts": [{"text": msg.content}]}
+                )
+
+        return formatted_history
+
+    def get_response_gemini(
+        self,
+        message: str,
+        context_data: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Get a response from the assistant for the given message."""
+        full_message = message
+        if self.system_prompt and not self.conversation_history:
+            full_message = (
+                f"{self.system_prompt}\n\n{message}\n\nContext:\n{context_data}\n"
+            )
+        else:
+            full_message = f"{message}\n\nContext:\n{context_data}\n"
+
+        # Add user message to history
+        self.add_message(Role.USER, full_message)
+
+        try:
+            # Create chat session with properly formatted history
+            chat = self.model.start_chat(history=self._format_messages_for_gemini())
+
+            # Generate response
+            response = chat.send_message(full_message)
+            response_text = response.text
+
+            # Parse response as JSON
+            try:
+                response_data = json.loads(response_text)
+
+                # Ensure required fields are present
+                if "source_id" not in response_data:
+                    response_data["source_id"] = "default_source_id"
+                if "quote" not in response_data:
+                    response_data["quote"] = "default_quote"
+
+                # Validate with Pydantic model if specified
+                if self.output_model:
+                    structured_response = self.output_model(**response_data)
+                    self.add_message(
+                        Role.ASSISTANT, json.dumps(response_data, indent=2)
+                    )
+                    return structured_response.dict()
+
+                # Add response to history and return as dict
+                self.add_message(Role.ASSISTANT, response_text)
+                return response_data
+
+            except (json.JSONDecodeError, ValueError) as e:
+                raise ValueError(
+                    f"Failed to parse response as JSON: {str(e)}. Response: {response_text}"
+                )
+
+        except Exception as e:
+            raise Exception(f"Error generating response: {str(e)}")
+
+    def clear_history(self) -> None:
+        """Clear the conversation history."""
+        self.conversation_history = []
+        if self.system_prompt:
+            self.add_message(Role.SYSTEM, self.system_prompt)

src/chroma_rag/chroma.sqlite3

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:b91a33bf0cab01f5a6e14731ab350378bc396ef1ee7d6c339b279327f33ac337
+size 30126080

src/chroma_rag/e8532e8c-980b-42a2-acd8-df1effcd172a/data_level0.bin

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:a13e72541800c513c73dccea69f79e39cf4baef4fa23f7e117c0d6b0f5f99670
+size 3212000

src/chroma_rag/e8532e8c-980b-42a2-acd8-df1effcd172a/header.bin

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:0ec6df10978b056a10062ed99efeef2702fa4a1301fad702b53dd2517103c746
+size 100

src/chroma_rag/e8532e8c-980b-42a2-acd8-df1effcd172a/length.bin

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:3d7d015a7ab0459c0f4812e91eb55167f930ea9df777241555219a13bc4f9512
+size 4000

src/notebooks/read_pdf.py

@@ -0,0 +1,15 @@
+import fitz  # PyMuPDF
+
+# Load the PDF file
+pdf_path = "../data/Artificial.Intelligence.A.Modern.Approach.4th.Edition.Peter.Norvig. Stuart.Russell.Pearson.9780134610993.EBooksWorld.ir.pdf"
+doc = fitz.open(pdf_path)
+
+# Extract text from each page
+for page_number in range(len(doc)):
+    page = doc[page_number]
+    print(f"Page {page_number + 1}:")
+    print(page.get_text())
+    print("-" * 50)
+
+# Close the document
+doc.close()

src/notebooks/temp.py

@@ -0,0 +1,58 @@
+import fitz  # PyMuPDF
+import pytesseract
+from PIL import Image
+
+
+# Define a function to extract text from all pages of a PDF
+def extract_text_from_pdf(pdf_path, dpi=300):
+    """
+    Extract text from all pages of a PDF.
+
+    Args:
+        pdf_path (str): Path to the PDF file.
+        dpi (int): Resolution for converting PDF pages to images (default: 300).
+
+    Returns:
+        dict: A dictionary where keys are page numbers (1-based) and values are extracted text.
+    """
+    # Open the PDF file
+    pdf_document = fitz.open(pdf_path)
+    extracted_text = {}
+
+    for page_number in range(len(pdf_document)):
+        # Select the page
+        page = pdf_document[page_number]
+
+        # Convert the page to an image
+        pixmap = page.get_pixmap(dpi=dpi)
+
+        # Save the image to a temporary file
+        image = Image.frombytes("RGB", [pixmap.width, pixmap.height], pixmap.samples)
+
+        # Extract text from the image using Tesseract
+        text = pytesseract.image_to_string(image)
+
+        print(text)
+
+        # Store the text in the dictionary
+        extracted_text[page_number + 1] = text
+
+    # Close the PDF document
+    pdf_document.close()
+
+    return extracted_text
+
+
+# Usage example
+if __name__ == "__main__":
+    # pdf_path = "c:/Abhi-MTech/Sem-1/AI/Books/Artificial.Intelligence.A.Modern.Approach.4th.Edition.Peter.Norvig. Stuart.Russell.Pearson.9780134610993.EBooksWorld.ir.pdf"  # Path to your PDF file
+    pdf_path = "c:/Abhi-MTech/Sem-1/AI/AI Technical.pdf"  # Path to your PDF file
+
+    try:
+        all_text = extract_text_from_pdf(pdf_path)
+        for page_num, text in all_text.items():
+            print(f"Page {page_num} Text:")
+            print(text)
+            print("-" * 80)  # Separator for readability
+    except Exception as e:
+        print(f"Error: {e}")

src/response_api.py

@@ -0,0 +1,292 @@
+# import os
+# import sys
+
+# sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+# import json
+
+# from pydantic import ValidationError
+# from src.assistants.assistant_v1 import gemini_rag_assistant
+# from src.utils.knowledge_base import AgenticRAG
+
+
+# async def get_response(query=True, is_uploaded=False):
+#     rag = AgenticRAG(query_value=query, is_uploaded=is_uploaded)
+
+#     # query = input("Enter your query: ")
+#     context = rag.query(query_text=query, n_results=10)
+#     print("\nQuery Results:")
+#     print(json.dumps(context, indent=2))
+
+#     try:
+#         response = await gemini_rag_assistant.get_response(
+#             message=query, context_data=context
+#         )
+#         print("\nAssistant Response:")
+#         print(response)
+
+#         return {"response": response, "context": context}
+
+#     except ValidationError as e:
+#         print("Validation Error:", e)
+#         return {
+#             "source_id": "validation_error",
+#             "content": str(e),
+#         }
+
+#     except Exception as e:
+#         print("Internal Server Error:", e)
+#         return {
+#             "source_id": "internal_error",
+#             "content": "Internal Server Error",
+#         }
+
+
+# async def main():
+#     rag = AgenticRAG(query_value=True)
+#     # while True:
+#     query = input("Enter your query: ")
+#     results = rag.query(query_text=query, n_results=10)
+#     print("\nQuery Results:")
+#     print(json.dumps(results, indent=2))
+
+#     try:
+#         print("gemini start generating answer")
+#         response = await gemini_rag_assistant.get_response(
+#             message=query, context_data=results
+#         )
+#         print("\nAssistant Response:")
+#         print(response)
+
+#     except ValidationError as e:
+#         print("Validation Error:", e)
+#         return {
+#             "source_id": "validation_error",
+#             "content": str(e),
+#         }
+
+#     except Exception as e:
+#         print("Internal Server Error:", e)
+#         return {
+#             "source_id": "internal_error",
+#             "content": "Internal Server Error",
+#         }
+
+
+# if __name__ == "__main__":
+#     import asyncio
+
+#     asyncio.run(main())
+
+
+import json
+import os
+import traceback
+
+import google.generativeai as genai
+from dotenv import load_dotenv
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel, ValidationError
+
+from agents_rag.crew import get_crew_response
+from assistants.assistant_v1 import gemini_rag_assistant
+from utils.knowledge_base import AgenticRAG
+from utils.vectorDB import VectorStore
+
+load_dotenv()
+
+GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
+genai.configure(api_key=GEMINI_API_KEY)
+
+# Initialize FastAPI app
+app = FastAPI(title="Gemini RAG Assistant API", version="1.0.0")
+
+
+# Pydantic model for request body
+class QueryRequest(BaseModel):
+    query: str
+    is_uploaded: bool = False
+    url: str
+
+
+# Pydantic model for response
+class QueryResponse(BaseModel):
+    response: str
+    context: dict
+
+
+def llm_answer(query=""):
+    try:
+        # Initialize Gemini RAG assistant
+        model = genai.GenerativeModel(model_name="gemini-2.0-flash-exp")
+
+        print("query", query)
+
+        response = model.generate_content(query)
+        print(response.text)
+
+        return {"response": response.text, "status": "success"}
+
+    except Exception as e:
+        print(f"Error in Gemini chunking: {e}")
+        return [
+            {
+                "response": "",
+                "status": "fail",
+            }
+        ]
+
+
+@app.post("/get-response")
+def get_response(request: QueryRequest):
+    """
+    Endpoint to process a query and get a response from the assistant.
+    """
+    try:
+        # Initialize AgenticRAG and fetch context
+        rag = AgenticRAG(query_value=request.query, is_uploaded=request.is_uploaded)
+        context = rag.query(query_text=request.query, n_results=15)
+
+        print("Generate answer form gemini")
+        # Fetch response from gemini assistant
+        # response = gemini_rag_assistant.get_response_gemini(
+        #     message=request.query, context_data=context
+        # )
+
+        response = get_crew_response(
+            query=request.query, context=context, url=request.url
+        )
+
+        print(response)
+
+        cleaned_text = "".join(
+            char for char in response if ord(char) >= 32 or char in "\n\r\t"
+        )
+        result = json.loads(cleaned_text)
+        print(result)
+
+        result = {
+            "response": result["Answer"],
+            "context": result["context"],
+            "citations": result["citations"],
+        }
+        # print(result)
+
+        return result
+
+    except ValidationError as e:
+        raise HTTPException(status_code=422, detail=f"Validation Error: {e}")
+
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=f"Value Error: {e}")
+
+    except Exception as e:
+        traceback.print_exc()  # Log the full traceback
+        raise HTTPException(status_code=500, detail=f"Internal Server Error: {str(e)}")
+
+
+@app.post("/llm-response")
+def get_response_llm(request: dict):
+    """
+    Endpoint to process a query and get a response from the assistant.
+    """
+    try:
+        # Initialize AgenticRAG and fetch context
+
+        print("Generate answer form gemini")
+        # Fetch response from gemini assistant
+
+        result = llm_answer(query=request["query"])
+
+        result = {
+            "response": result["response"],
+        }
+        # print(result)
+
+        return result
+
+    except ValidationError as e:
+        raise HTTPException(status_code=422, detail=f"Validation Error: {e}")
+
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=f"Value Error: {e}")
+
+    except Exception as e:
+        traceback.print_exc()  # Log the full traceback
+        raise HTTPException(status_code=500, detail=f"Internal Server Error: {str(e)}")
+
+
+@app.get("/health")
+def health_check():
+    """
+    Endpoint for health check.
+    """
+    return {"status": "ok"}
+
+
+@app.post("/delete-file")
+async def process_upload_data(request: dict):
+    """
+    Endpoint to retrieve do emedding of new file and store the result in Vector database.
+    """
+    try:
+        db = VectorStore()
+
+        print("deletion started.")
+        db.delete_documents_by_filename(request["file_path"])
+        print("deletion end.")
+
+        return {"response": 200}
+
+    except ValidationError as e:
+        raise HTTPException(status_code=422, detail=f"Validation Error: {e}")
+
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=f"Value Error: {e}")
+
+    except Exception as e:
+        traceback.print_exc()  # Log the full traceback
+        raise HTTPException(status_code=500, detail=f"Internal Server Error: {str(e)}")
+
+
+@app.post("/process-file")
+async def process_upload_data(request: dict):
+    """
+    Endpoint to retrieve do emedding of new file and store the result in Vector database.
+    """
+    try:
+        # Initialize AgenticRAG and fetch context
+        rag = AgenticRAG(is_uploaded=False)
+
+        print("process started.")
+        rag.process_file(request["file_path"])
+        print("process end.")
+
+        # rag = AgenticRAG(query_value=request.query)
+        # context = rag.query(query_text=request.query, n_results=10)
+
+        # # Fetch response from gemini assistant
+        # response = await gemini_rag_assistant.get_response(
+        #     message=request.query, context_data=context
+        # )
+
+        # # Ensure response is in correct format
+        # if not isinstance(response, str):
+        #     raise ValueError("Unexpected response format from gemini_rag_assistant.")
+
+        return {"response": 200}
+
+    except ValidationError as e:
+        raise HTTPException(status_code=422, detail=f"Validation Error: {e}")
+
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=f"Value Error: {e}")
+
+    except Exception as e:
+        traceback.print_exc()  # Log the full traceback
+        raise HTTPException(status_code=500, detail=f"Internal Server Error: {str(e)}")
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run(app, host="0.0.0.0", port=8000)

src/run_app.py

@@ -0,0 +1,45 @@
+import os
+import sys
+
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+import json
+
+from pydantic import ValidationError
+
+from src.assistants.assistant_v1 import gemini_rag_assistant
+from src.utils.knowledge_base import AgenticRAG
+
+
+def main():
+    rag = AgenticRAG(query_value=True)
+    while True:
+        query = input("Enter your query: ")
+        results = rag.query(query_text=query, n_results=10)
+        print("\nQuery Results:")
+        print(json.dumps(results, indent=2))
+
+        try:
+            print("gemini start generating answer")
+            response = gemini_rag_assistant.get_response_gemini(
+                message=query, context_data=results
+            )
+            print("\nAssistant Response:")
+            print(response)
+
+        except ValidationError as e:
+            print("Validation Error:", e)
+            return {
+                "source_id": "validation_error",
+                "content": str(e),
+            }
+
+        except Exception as e:
+            print("Internal Server Error:", e)
+            return {
+                "source_id": "internal_error",
+                "content": "Internal Server Error",
+            }
+
+
+if __name__ == "__main__":
+    main()

src/utils/download.py

@@ -0,0 +1,126 @@
+import os
+import traceback
+
+import requests
+import yt_dlp
+from bs4 import BeautifulSoup
+from download_video import downlaod_video_from_url
+from pytube import YouTube
+
+
+def download_youtube_video(url, download_path="../data/"):
+    try:
+        yt = YouTube(url)
+
+        # Get the best stream (highest resolution video)
+        video_stream = (
+            yt.streams.filter(progressive=True, file_extension="mp4")
+            .order_by("resolution")
+            .desc()
+            .first()
+        )
+
+        # If the stream exists, download it
+        if video_stream:
+            video_stream.download(output_path=download_path)
+            print(f"Video downloaded successfully to {download_path}")
+        else:
+            print("No suitable video stream found")
+    except Exception as e:
+        print(f"Error in downloading YouTube video: {e}")
+
+
+def download_audio(url, download_path="../data/"):
+    """
+    Download audio from YouTube and convert to MP3 format.
+
+    Args:
+        url: YouTube video URL
+        download_path: Path where the MP3 file will be saved
+    """
+    ydl_opts = {
+        "outtmpl": f"{download_path}%(title)s.%(ext)s",
+        "format": "bestaudio/best",
+        "geo-bypass": True,
+        "noplaylist": True,
+        "force-ipv4": True,
+        # Add postprocessors for MP3 conversion
+        "postprocessors": [
+            {
+                "key": "FFmpegExtractAudio",
+                "preferredcodec": "mp3",
+                "preferredquality": "192",
+            }
+        ],
+        "headers": {
+            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
+        },
+    }
+
+    try:
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            ydl.download([url])
+        print(f"Audio downloaded and converted to MP3 successfully at {download_path}")
+    except Exception as e:
+        print(f"An error occurred: {e}")
+
+
+# Function to download PDF, DOC, or other files
+def download_file(url, download_path="../data/"):
+    try:
+        response = requests.get(url, stream=True)
+        response.raise_for_status()  # Check if the request was successful
+        filename = os.path.join(download_path, url.split("/")[-1])
+
+        with open(filename, "wb") as file:
+            for chunk in response.iter_content(chunk_size=1024):
+                if chunk:
+                    file.write(chunk)
+        print(f"File downloaded successfully to {filename}")
+    except Exception as e:
+        print(f"An error occurred: {e}")
+
+
+# Function to download text or webpage content
+def download_text_or_webpage(url, download_path="../data/", is_text=False):
+    try:
+        response = requests.get(url)
+        response.raise_for_status()
+
+        if is_text:
+            filename = os.path.join(download_path, url.split("/")[-1] + ".txt")
+            with open(filename, "w") as file:
+                file.write(response.text)
+            print(f"Text file downloaded successfully to {filename}")
+        else:
+            soup = BeautifulSoup(response.text, "html.parser")
+            filename = os.path.join(download_path, url.split("/")[-1] + ".html")
+            with open(filename, "w", encoding="utf-8") as file:
+                file.write(soup.prettify())
+            print(f"Webpage downloaded successfully to {filename}")
+
+    except Exception as e:
+        print(f"An error occurred: {e}")
+
+
+def main():
+    # Example Usage:
+    # url_video = "https://www.youtube.com/watch?v=dIYmzf21d1g"
+    # downlaod_video_from_url(
+    #     youtube_url=url_video, download_path="../data/"
+    # )  # Download video
+    url_audio = "https://www.youtube.com/watch?v=8OHYynw7Yh4"
+    download_audio(url_audio)  # Download audio
+
+    # url_pdf = "https://example.com/somefile.pdf"
+    # download_file(url_pdf)  # Download PDF, DOC, or any other file
+
+    # url_text = "https://example.com/sometextfile"
+    # download_text_or_webpage(url_text, is_text=True)  # Download text
+
+    # url_webpage = "https://en.wikipedia.org/wiki/Microsoft"
+    # download_text_or_webpage(url_webpage)  # Download webpage content
+
+
+if __name__ == "__main__":
+    main()

src/utils/download_video.py

@@ -0,0 +1,170 @@
+import logging
+import os
+import random
+import time
+from typing import Any, Dict, List, Optional
+
+import requests
+import yt_dlp
+from bs4 import BeautifulSoup
+from pytube import YouTube
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger(__name__)
+
+
+class ContentDownloader:
+    def __init__(self, download_path: str = "./downloads/"):
+        self.download_path = download_path
+        self.create_download_directory()
+
+    def create_download_directory(self) -> None:
+        """Create download directory if it doesn't exist."""
+        os.makedirs(self.download_path, exist_ok=True)
+
+    def _get_available_formats(self, url: str) -> List[Dict]:
+        """Get list of available formats for a YouTube video."""
+        ydl_opts = {"quiet": True, "no_warnings": True, "extract_flat": True}
+
+        try:
+            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                info = ydl.extract_info(url, download=False)
+                formats = info.get("formats", [])
+                # Filter for formats that have both video and audio
+                combined_formats = [
+                    f
+                    for f in formats
+                    if f.get("vcodec") != "none" and f.get("acodec") != "none"
+                ]
+                return combined_formats
+        except Exception as e:
+            logger.error(f"Error getting formats: {str(e)}")
+            return []
+
+    def download_youtube_content(
+        self, url: str, download_audio: bool = False
+    ) -> Optional[str]:
+        """
+        Download YouTube content with automatic format selection.
+        """
+        if download_audio:
+            ydl_opts = {
+                "outtmpl": os.path.join(self.download_path, "%(title)s.%(ext)s"),
+                "format": "bestaudio/best",
+                "postprocessors": [
+                    {
+                        "key": "FFmpegExtractAudio",
+                        "preferredcodec": "mp3",
+                    }
+                ],
+            }
+        else:
+            # Get available formats first
+            formats = self._get_available_formats(url)
+            if not formats:
+                logger.error("No suitable formats found")
+                return None
+
+            # Configure options for video download
+            ydl_opts = {
+                "outtmpl": os.path.join(self.download_path, "%(title)s.%(ext)s"),
+                "format": "bv*[ext=mp4]+ba[ext=m4a]/b[ext=mp4] / bv*+ba/b",  # Prefer MP4 format
+                "merge_output_format": "mp4",
+                "postprocessors": [
+                    {
+                        "key": "FFmpegVideoRemuxer",
+                        "preferedformat": "mp4",
+                    }
+                ],
+                "quiet": False,
+                "no_warnings": False,
+                "max_filesize": 2048 * 1024 * 1024,  # 2GB max
+                "geo_bypass": True,
+                "nocheckcertificate": True,
+                "http_headers": {
+                    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
+                    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+                    "Accept-Language": "en-us,en;q=0.5",
+                    "Sec-Fetch-Mode": "navigate",
+                },
+            }
+
+        try:
+            # First update yt-dlp
+            os.system("yt-dlp -U")
+
+            # Attempt download with yt-dlp
+            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                logger.info("Attempting download with yt-dlp...")
+                ydl.download([url])
+                return self.download_path
+
+        except Exception as e:
+            logger.warning(f"yt-dlp download failed: {str(e)}")
+            logger.info("Attempting fallback to direct stream download...")
+            return self._download_with_direct_stream(url)
+
+    def _download_with_direct_stream(
+        self, url: str, max_retries: int = 3
+    ) -> Optional[str]:
+        """Alternative download method using direct stream access."""
+        for attempt in range(max_retries):
+            try:
+                if attempt > 0:
+                    time.sleep(random.uniform(2, 5))
+
+                yt = YouTube(url)
+                # Sort streams by both resolution and bitrate
+                streams = yt.streams.filter(progressive=True, file_extension="mp4")
+                stream = streams.order_by("resolution").desc().first()
+
+                if stream:
+                    # Add random query parameter to avoid caching
+                    timestamp = int(time.time())
+                    stream.url = f"{stream.url}&_={timestamp}"
+
+                    file_path = stream.download(
+                        output_path=self.download_path,
+                        filename_prefix=f"video_{timestamp}_",
+                    )
+                    logger.info(f"Successfully downloaded to: {file_path}")
+                    return file_path
+                else:
+                    logger.error("No suitable stream found")
+                    return None
+
+            except Exception as e:
+                logger.error(f"Download attempt {attempt + 1} failed: {str(e)}")
+                if attempt == max_retries - 1:
+                    logger.error("All download attempts failed")
+                    return None
+
+
+def downlaod_video_from_url(youtube_url="", download_path="./downloads/"):
+    # Update yt-dlp first
+    os.system("yt-dlp -U")
+
+    downloader = ContentDownloader(download_path=download_path)
+
+
+
+    # First, check available formats
+    formats = downloader._get_available_formats(youtube_url)
+    if formats:
+        print("\nAvailable formats:")
+        for f in formats:
+            print(
+                f"Format ID: {f.get('format_id')} - "
+                f"Resolution: {f.get('resolution')} - "
+                f"Filesize: {f.get('filesize_approx', 'unknown')} bytes"
+            )
+
+    # Download video with audio
+    video_path = downloader.download_youtube_content(youtube_url)
+    if video_path:
+        print(f"\nVideo downloaded to: {video_path}")
+    else:
+        print("\nDownload failed")

src/utils/json_file_record.json

@@ -0,0 +1,14 @@
+[
+    {
+        "id": "esnXVYfJ",
+        "file_path": "data\\What Makes Hanuman Chalisa So POWERFUL - Tantric Explains  TRS Clips.mp3"
+    },
+    {
+        "id": "iVeOlEJl",
+        "file_path": "data\\The Science Of Building Extreme Discipline - Andrew Huberman.mp3"
+    },
+    {
+        "id": "nrrAW71c",
+        "file_path": "data\\attenstion_is_all_you_need.pdf"
+    }
+]

src/utils/knowledge_base.py

@@ -0,0 +1,523 @@
+import os
+import sys
+
+# Add the project root to Python path
+project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))
+sys.path.append(project_root)
+
+import json
+import os
+import re
+import time
+from typing import Any, Dict, List, Optional
+from urllib.parse import urlparse
+
+import docx
+import exceptiongroup
+import google.generativeai as genai
+import numpy as np
+import pandas as pd
+import PyPDF2
+import requests
+from bs4 import BeautifulSoup
+from dotenv import load_dotenv
+from nanoid import generate
+from pydantic import BaseModel, Field
+
+from src.utils.vectorDB import VectorStore
+
+load_dotenv()
+
+
+class MetadataPDF(BaseModel):
+    key_concepts: List[str] = Field(
+        ..., description="Key concepts related to the topic"
+    )
+    page_number: int = Field(
+        ...,
+        alias="page-number",
+        description="The page number where this content is located",
+    )
+
+
+class SegmentPDF(BaseModel):
+    content: str = Field(..., description="The main text of the segment")
+    metadata: MetadataPDF
+
+
+class AnalyzedContentPDF(BaseModel):
+    segments: List[SegmentPDF] = Field(
+        ..., description="List of meaningful content segments"
+    )
+
+
+# for text data
+
+
+class MetadataTxt(BaseModel):
+    key_concepts: List[str] = Field(
+        ..., description="Key concepts related to the topic"
+    )
+
+
+class SegmentTxt(BaseModel):
+    content: str = Field(..., description="The main text of the segment")
+    metadata: MetadataTxt
+
+
+class AnalyzedContentTxt(BaseModel):
+    segments: List[SegmentTxt] = Field(
+        ..., description="List of meaningful content segments"
+    )
+
+
+class GeminiChunker:
+    def __init__(self):
+        self.api_key = os.getenv("GEMINI_API_KEY")
+        genai.configure(api_key=self.api_key)
+        # self.model = genai.GenerativeModel(model_name="gemini-2.0-flash-exp")
+        self.model = genai.GenerativeModel(model_name="gemini-1.5-flash")
+        # self.model = genai.GenerativeModel(model_name="gemini-1.5-pro")
+
+    def check_file_ready(self, file):
+        while file.state.name == "PROCESSING":
+            print(".", end="")
+            time.sleep(10)
+            file = genai.get_file(file.name)
+
+        if file.state.name == "FAILED":
+            raise ValueError(f"File processing failed: {file.state.name}")
+
+    def chunk_with_gemini(
+        self, content: str, content_type: str
+    ) -> List[Dict[str, Any]]:
+        # cleaned_content = "".join(
+        #     char for char in content if ord(char) >= 32 or char in "\n\r\t"
+        # )
+
+        # # Encode and decode to handle unsupported characters
+        # safe_string = cleaned_content.encode("utf-8", errors="replace").decode("utf-8")
+        # print(safe_string)
+
+        if content_type == "pdf" or content_type == "docx":
+            """Use Gemini to intelligently chunk content based on semantic understanding"""
+            prompt = f"""
+            Analyze the following {content_type} content first means read whole content first then after divide it into complete and meaningful segments (chunks). 
+            Each chunk size has 512 token should:
+            1. Be self-contained and end at logical boundaries (e.g., complete sentences or sections).
+            2. Include all text that belongs to a single segment without truncation.
+            3. Ensure the last chunk is fully complete and not cut off.
+
+            Return the response strictly in the specified schema format:
+
+                {{
+                    "content": "segment text here",
+                    "metadata": {{
+                        "key_concepts": ["concept1", "concept2"],
+                        "page-number": 64
+                    }}
+                }},
+                // more segments...
+
+            Content to analyze:
+            {content}
+
+            Keep the response as pure JSON without any additional text or explanation. Avoid splitting content mid-sentence or mid-thought.
+            All chunks should be complete.
+            """
+            schema = AnalyzedContentPDF
+        else:
+            prompt = f"""
+            Analyze the following {content_type} content first means read whole content first then after divide it into complete and meaningful segments (chunks). 
+            Each chunk should:
+            1. Be self-contained and end at logical boundaries (e.g., complete sentences or sections).
+            2. Include all text that belongs to a single segment without truncation.
+            3. Ensure the last chunk is fully complete and not cut off.
+
+            Return the response strictly in the specified schema format:
+
+                {{
+                    "content": "segment text here",
+                    "metadata": {{
+                        "key_concepts": ["concept1", "concept2"],
+                        "page-number": NA
+                    }}
+                }},
+                // more segments...
+
+            Content to analyze:
+            {content}
+
+            Keep the response as pure JSON without any additional text or explanation. Avoid splitting content mid-sentence or mid-thought.
+            All chunks should be complete.
+            """
+            schema = AnalyzedContentTxt
+
+        print(schema)
+
+        try:
+            response = self.model.generate_content(
+                prompt,
+                generation_config=genai.GenerationConfig(
+                    response_mime_type="application/json",
+                    response_schema=schema,
+                ),
+            )
+            # print(response.text)
+            cleaned_text = "".join(
+                char for char in response.text if ord(char) >= 32 or char in "\n\r\t"
+            )
+            with open("chunking_text.txt", "w", encoding="utf-8") as file_text:
+                print("^^^^^^^^^^^^^^^^^^^^^^^^^^^^^")
+                file_text.write(cleaned_text)
+                print("^^^^^^^^^^^^^^^^^^^^^^^^^^^^^")
+
+            cleaned_text = cleaned_text.encode("utf-8", errors="replace").decode(
+                "utf-8"
+            )
+
+            result = json.loads(cleaned_text)
+
+            # print(result)
+
+            print("pdf parsing")
+
+            chunks = []
+
+            for segment in result.get("segments", []):
+                # print("################################")
+                # print(segment.get("content", ""))
+                # print("################################")
+                temp_metadata = segment["metadata"]
+                if content_type == "pdf":
+                    chunk = {
+                        "content": segment.get("content", ""),
+                        "metadata": {
+                            "topics": temp_metadata.get("key_concepts", []),
+                            "page-number": temp_metadata.get("page-number", ""),
+                            "type": "pdf",
+                        },
+                    }
+                else:
+                    chunk = {
+                        "content": segment.get("content", ""),
+                        "metadata": {
+                            "topics": segment.get("key_concepts", []),
+                            "page-number": segment.get("page-number", ""),
+                            "type": "text",
+                        },
+                    }
+                chunks.append(chunk)
+
+            # print()
+            # print()
+            # print("chunks:")
+            # print(chunks)
+            # print()
+            # print()
+
+            return chunks
+
+        except Exception as e:
+            print(f"Error in Gemini chunking: {e}")
+            return [
+                {
+                    "content": content,
+                    "metadata": {
+                        "topics": "",
+                        "page-number": "0",
+                    },
+                }
+            ]
+
+    def process_media_file(
+        self, file_path: str, media_type: str
+    ) -> List[Dict[str, Any]]:
+        """Process video or audio file using Gemini's media understanding"""
+        try:
+            print("here-0!!!")
+            media_file = genai.upload_file(path=file_path)
+            self.check_file_ready(media_file)
+            print("here-1!!!")
+
+            if media_type == "video":
+                schema = {
+                    "type": "object",
+                    "properties": {
+                        "segments": {
+                            "type": "array",
+                            "items": {
+                                "type": "object",
+                                "properties": {
+                                    "timestamp": {"type": "string"},
+                                    "description": {"type": "string"},
+                                    "topics": {
+                                        "type": "array",
+                                        "items": {"type": "string"},
+                                    },
+                                },
+                            },
+                        }
+                    },
+                }
+
+                prompt = "Describe this video in detail, breaking it into timestamped segments. Include key events and actions."
+            else:  # audio
+                schema = {
+                    "type": "object",
+                    "properties": {
+                        "segments": {
+                            "type": "array",
+                            "items": {
+                                "type": "object",
+                                "properties": {
+                                    "timestamp": {"type": "string"},
+                                    "transcription": {"type": "string"},
+                                    "speaker": {"type": "string"},
+                                    "topics": {
+                                        "type": "array",
+                                        "items": {"type": "string"},
+                                    },
+                                },
+                            },
+                        }
+                    },
+                }
+
+                prompt = "Transcribe this audio, identifying speakers and key topics discussed."
+
+            print("Here-2!!!")
+            response = self.model.generate_content(
+                [media_file, prompt],
+                generation_config=genai.GenerationConfig(
+                    response_schema=schema, response_mime_type="application/json"
+                ),
+            )
+            print("Here-3!!!")
+
+            # Convert Gemini's media response to our standard chunk format
+            print(response.text)
+            print("Here-4!!!")
+            cleaned_text = "".join(
+                char for char in response.text if ord(char) >= 32 or char in "\n\r\t"
+            )
+            result = json.loads(cleaned_text)
+            # result = json.loads(response.text)
+            chunks = []
+            print("Here-5!!!")
+
+            for segment in result.get("segments", []):
+                if media_type == "video":
+                    chunk = {
+                        "content": segment.get("description", ""),
+                        "metadata": {
+                            "timestamp": segment.get("timestamp", ""),
+                            "topics": segment.get("key_events", []),
+                            "type": "video",
+                        },
+                    }
+                else:
+                    chunk = {
+                        "content": segment.get("transcription", ""),
+                        "metadata": {
+                            "timestamp": segment.get("timestamp", ""),
+                            "speaker": segment.get("speaker", ""),
+                            "topics": segment.get("topics", []),
+                            "type": "audio",
+                        },
+                    }
+                chunks.append(chunk)
+            print("Here-6!!!")
+
+            return chunks
+
+        except Exception as e:
+            print(f"Error processing {media_type} file: {e}")
+            return [
+                {
+                    "content": f"Error processing {media_type} file",
+                    "metadata": {"type": media_type, "error": str(e)},
+                }
+            ]
+
+
+class ContentProcessor:
+    def __init__(self):
+        self.gemini_chunker = GeminiChunker()
+
+    def process_text(self, text: str, source_type: str) -> List[Dict[str, Any]]:
+        """Process any text content using Gemini chunking"""
+        chunks = self.gemini_chunker.chunk_with_gemini(text, source_type)
+        for chunk in chunks:
+            chunk["metadata"]["source_type"] = source_type
+        return chunks
+
+    def process_pdf(self, file_path: str) -> List[Dict[str, Any]]:
+        with open(file_path, "rb") as file:
+            pdf_reader = PyPDF2.PdfReader(file)
+            full_text = ""
+            for page in pdf_reader.pages:
+                full_text += page.extract_text() + " "
+        return self.process_text(full_text, "pdf")
+
+    def process_docx(self, file_path: str) -> List[Dict[str, Any]]:
+        doc = docx.Document(file_path)
+        full_text = " ".join([paragraph.text for paragraph in doc.paragraphs])
+        return self.process_text(full_text, "docx")
+
+    def process_csv(self, file_path: str) -> List[Dict[str, Any]]:
+        df = pd.read_csv(file_path)
+        # Convert DataFrame to a more readable format for Gemini
+        text_content = df.to_string()
+        return self.process_text(text_content, "csv")
+
+    def process_webpage(self, url: str) -> List[Dict[str, Any]]:
+        response = requests.get(url)
+        soup = BeautifulSoup(response.text, "html.parser")
+        for script in soup(["script", "style"]):
+            script.decompose()
+        text = soup.get_text()
+        return self.process_text(text, "webpage")
+
+    def process_video(self, file_path: str) -> List[Dict[str, Any]]:
+        """Process video using Gemini's video understanding capabilities"""
+        print("in process function of video")
+        video_file = genai.upload_file(path=file_path)
+        self.gemini_chunker.check_file_ready(video_file)
+
+        chunks = self.gemini_chunker.process_media_file(
+            file_path=file_path, media_type="video"
+        )
+        return chunks
+
+    def process_audio(self, file_path: str) -> List[Dict[str, Any]]:
+        """Process audio using Gemini's audio understanding capabilities"""
+        print("in process function of audio")
+        audio_file = genai.upload_file(path=file_path)
+        self.gemini_chunker.check_file_ready(audio_file)
+
+        chunks = self.gemini_chunker.process_media_file(
+            file_path=file_path, media_type="audio"
+        )
+        return chunks
+
+
+class AgenticRAG:
+    def __init__(self, query_value=False, is_uploaded=False):
+        self.processor = ContentProcessor()
+        self.vector_store = VectorStore(query=query_value, is_uploaded=is_uploaded)
+        if query_value == False and is_uploaded == True:
+            self.json_file_path = "json_file_record.json"
+        else:
+            self.json_file_path = "utils/json_file_record.json"
+
+    def process_file(self, file_path: str, file_type: Optional[str] = None):
+        if file_type is None:
+            file_type = self._detect_file_type(file_path)
+
+        if os.path.exists(self.json_file_path):
+            with open(self.json_file_path, "r") as json_file:
+                json_data = json.load(json_file)
+                for record in json_data:
+                    if record["file_path"] == file_path:
+                        return True  # File path exists
+        try:
+            chunks = []
+            if file_type == "pdf":
+                chunks = self.processor.process_pdf(file_path)
+            elif file_type == "docx":
+                chunks = self.processor.process_docx(file_path)
+            elif file_type == "csv":
+                chunks = self.processor.process_csv(file_path)
+            elif file_type == "url":
+                chunks = self.processor.process_webpage(file_path)
+            elif file_type == "video":
+                chunks = self.processor.process_video(file_path)
+            elif file_type == "audio":
+                chunks = self.processor.process_audio(file_path)
+            elif file_type == "text":
+                with open(file_path, "r") as file:
+                    chunks = self.processor.process_text(file.read(), "text")
+
+            if chunks:
+                # Add source information to metadata
+                print("in processfile fucntion file.")
+                for chunk in chunks:
+                    chunk["metadata"]["source"] = file_path
+
+                print(chunks)
+
+                # Add to Vector Database.
+                self.vector_store.add_documents(chunks)
+                print(f"Successfully processed {file_path} with {len(chunks)} chunks")
+
+                return True
+            return False
+
+        except Exception as e:
+            print(f"Error processing {file_path}: {e}")
+
+    def _detect_file_type(self, file_path: str) -> str:
+        if file_path.startswith("http"):
+            return "url"
+
+        extension = file_path.split(".")[-1].lower()
+        type_mapping = {
+            "pdf": "pdf",
+            "docx": "docx",
+            "doc": "docx",
+            "csv": "csv",
+            "txt": "text",
+            "mp3": "audio",
+            "wav": "audio",
+            "mp4": "video",
+            "mov": "video",
+        }
+        return type_mapping.get(extension, "unknown")
+
+    def query(self, query_text: str, n_results: int = 5) -> Dict:
+        return self.vector_store.query(query_text, n_results)
+
+
+# Define a function to determine the file type based on the extension
+def get_file_type(file_name: str) -> str:
+    if file_name.endswith(".mp3"):
+        return "audio"
+    elif file_name.endswith(".mp4"):
+        return "video"
+    elif file_name.endswith(".csv"):
+        return "csv"
+    elif file_name.endswith(".pdf"):
+        return "pdf"
+    elif file_name.endswith(".docx"):
+        return "docx"
+    elif file_name.startswith("http"):
+        return "url"
+    else:
+        return "unknown"
+
+
+def main():
+    # Initialize the RAG system
+    rag = AgenticRAG(is_uploaded=True)
+
+    # Automatically read the files in the 'data' directory
+    data_directory = "../data"
+    test_files = []
+
+    # Loop through all files in the 'data' directory
+    for filename in os.listdir(data_directory):
+        file_path = os.path.join(data_directory, filename)
+        if os.path.isfile(file_path):  # Check if it's a file
+            file_type = get_file_type(filename)
+            test_files.append((file_path, file_type))
+
+    # Process each file
+    for file_path, file_type in test_files:
+        print(f"\nProcessing {file_path}...")
+        time.sleep(5)
+        rag.process_file(file_path, file_type)
+
+
+if __name__ == "__main__":
+    main()

src/utils/vectorDB.py

@@ -0,0 +1,291 @@
+import json
+import os
+import sys
+from typing import Any, Dict, List
+
+# Add the project root to Python path
+project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+sys.path.append(project_root)
+
+import chromadb
+import numpy as np
+import torch
+from nanoid import generate
+from transformers import AutoModel, AutoTokenizer
+
+
+def append_to_json(new_entries, filename="json_file_record.json"):
+    """
+    Append new entries to an existing JSON array file, or create a new one if it doesn't exist.
+
+    Args:
+        new_entries (list): List of dictionaries to append
+        filename (str): Name of the JSON file
+    """
+    try:
+        # Read existing data if file exists
+        if os.path.exists(filename) and os.path.getsize(filename) > 0:
+            with open(filename, "r") as f:
+                data = json.load(f)
+            if not isinstance(data, list):
+                data = []
+        else:
+            data = []
+
+        # # Append new entries
+        # data.extend(new_entries)
+        # print(data)
+
+        # Write back the updated data
+        with open(filename, "w") as f:
+            json.dump(data, f, indent=4)
+
+    except json.JSONDecodeError:
+        # Handle case where file exists but is not valid JSON
+        data = new_entries
+        with open(filename, "w") as f:
+            json.dump(data, f, indent=4)
+
+
+class BERTEmbedder:
+    def __init__(self):
+        self.tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
+        self.model = AutoModel.from_pretrained("bert-base-uncased")
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        self.model.to(self.device)
+
+    def get_embeddings(self, texts: List[str]) -> np.ndarray:
+        embeddings = []
+        self.model.eval()
+        with torch.no_grad():
+            for text in texts:
+                inputs = self.tokenizer(
+                    text,
+                    padding=True,
+                    truncation=True,
+                    max_length=512,
+                    return_tensors="pt",
+                )
+                inputs = {k: v.to(self.device) for k, v in inputs.items()}
+                outputs = self.model(**inputs)
+                embeddings.append(outputs.last_hidden_state.mean(dim=1).cpu().numpy())
+        return np.vstack(embeddings)
+
+
+class VectorStore:
+    def __init__(
+        self, persist_directory: str = "../chroma_rag", query=False, is_uploaded=False
+    ):
+        try:
+            if query == False and is_uploaded == True:
+                print("Embbeding store mode.")
+                print(f"Initializing ChromaDB with directory: {persist_directory}")
+                self.client = chromadb.PersistentClient(path=persist_directory)
+                print("ChromaDB client created successfully")
+
+                self.collection = self.client.get_or_create_collection(
+                    name="documents",
+                    metadata={"hnsw:space": "cosine"},
+                    embedding_function=None,  # We're using our own embeddings
+                )
+                print(f"Collection 'documents' ready")
+
+                self.embedder = BERTEmbedder()
+                print("BERT embedder initialized")
+
+                # Check if collection has documents
+                content = self.collection.get()
+                print(f"Collection contains {len(content['documents'])} documents")
+                self.json_file_path = "json_file_record.json"
+
+            else:
+                print("query mode")
+                persist_directory = "chroma_rag"
+                print(f"Initializing ChromaDB with directory: {persist_directory}")
+                self.client = chromadb.PersistentClient(path=persist_directory)
+                print("ChromaDB client created successfully")
+
+                self.collection = self.client.get_or_create_collection(
+                    name="documents",
+                    metadata={"hnsw:space": "cosine"},
+                    embedding_function=None,  # We're using our own embeddings
+                )
+                print(f"Collection 'documents' ready")
+
+                self.embedder = BERTEmbedder()
+                print("BERT embedder initialized")
+
+                # Check if collection has documents
+                content = self.collection.get()
+                print(f"Collection contains {len(content['documents'])} documents")
+                self.json_file_path = "utils/json_file_record.json"
+
+        except Exception as e:
+            print(f"Error initializing VectorStore: {e}")
+            raise
+
+    def is_collection_empty(self) -> bool:
+        try:
+            content = self.collection.get()
+            return len(content["documents"]) == 0
+        except Exception as e:
+            print(f"Error checking collection: {e}")
+            return True
+
+    def add_documents(self, chunks: List[Dict[str, Any]]):
+        try:
+            texts = [chunk["content"] for chunk in chunks]
+            metadatas = [chunk["metadata"] for chunk in chunks]
+
+            print(f"Generating embeddings for {len(texts)} documents...")
+            print(texts)
+            embeddings = self.embedder.get_embeddings(texts)
+
+            id_val = str(generate(size=8))
+            print(f"Generated ID: {id_val}")
+
+            if os.path.exists(self.json_file_path):
+
+                with open(self.json_file_path, "r") as f:
+                    data = json.load(f)
+                    for chunk in chunks:
+                        temp = {"id": id_val, "file_path": chunk["metadata"]["source"]}
+
+                        break
+                    # Append the new entry
+                    data.append(temp)
+
+                # Write the updated JSON data back to the file
+                with open(self.json_file_path, "w") as file:
+                    json.dump(data, file, indent=4)
+            else:
+                # Usage in your code would be:
+                with open(self.json_file_path, "w") as f:
+                    temp = []
+                    for chunk in chunks:
+                        temp.append(
+                            {"id": id_val, "file_path": chunk["metadata"]["source"]}
+                        )
+                        break
+                # Write the updated JSON data back to the file
+                with open(self.json_file_path, "w") as file:
+                    json.dump(temp, file, indent=4)
+
+            print("*************")
+            count = 0
+            ids = []
+            # Clean metadata
+            for metadata in metadatas:
+                metadata["topics"] = str(metadata["topics"])
+                ids.append(f"{id_val}{count}")
+                count += 1
+            print(metadatas)
+            print("------------------------")
+            print(len(metadatas))
+
+            print(f"Adding {len(texts)} documents to collection...")
+            self.collection.add(
+                embeddings=embeddings.tolist(),
+                documents=texts,
+                metadatas=metadatas,
+                ids=ids,
+            )
+
+            # Verify addition
+            collection_content = self.collection.get()
+            print(
+                f"Collection now contains {len(collection_content['documents'])} documents"
+            )
+
+            return True
+        except Exception as e:
+            print(f"Error adding documents: {e}")
+            return False
+
+    def query(self, query_text: str, n_results: int = 3) -> Dict:
+        try:
+
+            print(f"Generating embedding for query: {query_text}")
+            query_embedding = self.embedder.get_embeddings([query_text])
+
+            print("Checking collection content:")
+            collection_content = self.collection.get()
+            print(
+                f"Number of documents in collection: {len(collection_content['documents'])}"
+            )
+
+            print("Executing query...")
+            query_vector = query_embedding.tolist()
+            results = self.collection.query(
+                n_results=min(n_results, len(collection_content["documents"])),
+                query_embeddings=query_vector,
+            )
+
+            # query_texts=[query_text],
+
+            print(f"Query results: {json.dumps(results, indent=2)}")
+            return results
+        except Exception as e:
+            print(f"Error during query: {e}")
+            return {"error": str(e)}
+
+    def delete_documents_by_filename(self, file_substring: str):
+        """
+        Delete documents from the collection and JSON file by matching a substring in the file path.
+
+        Args:
+            file_substring (str): Substring to match in the file paths.
+            json_file (str): Path to the JSON file containing document metadata.
+        """
+        try:
+            # Load JSON data
+            print(file_substring)
+            json_file = self.json_file_path
+            if not os.path.exists(json_file):
+                print(f"JSON file {json_file} does not exist.")
+                return
+
+            with open(json_file, "r") as f:
+                data = json.load(f)
+
+            # Find matching records
+            matching_records = [
+                record for record in data if file_substring in record["file_path"]
+            ]
+            if not matching_records:
+                print(f"No records found matching substring: {file_substring}")
+                return
+
+            # print("record", record)
+
+            # Get IDs of matching records
+            matching_ids = [record["id"] for record in matching_records]
+            print("maching_ids", matching_ids[0])
+
+            # Remove matching records from JSON file
+            updated_data = [
+                record for record in data if record["id"] not in matching_ids
+            ]
+
+            print("updated data", updated_data)
+
+            with open(json_file, "w") as f:
+                json.dump(updated_data, f, indent=4)
+
+            print(f"Deleted {len(matching_records)} records from JSON file.")
+            # Retrieve all IDs in the collection
+            all_ids = self.collection.get()["ids"]
+
+            # Filter IDs that contain the substring "LDtz9CG5"
+            ids_to_delete = [id_ for id_ in all_ids if matching_ids[0] in id_]
+
+            # Delete those IDs from the collection
+            if ids_to_delete:
+                self.collection.delete(ids=ids_to_delete)
+                print(
+                    f"Deleted {len(ids_to_delete)} records with IDs containing 'LDtz9CG5'."
+                )
+            else:
+                print("No matching IDs found.")
+        except Exception as e:
+            print(f"Error deleting documents: {e}")