Upload 6 files

App_Function_Libraries/RAG/ChromaDB_Library.py

@@ -0,0 +1,287 @@
+import configparser
+import logging
+import sqlite3
+from typing import List, Dict, Any
+
+import chromadb
+import requests
+from chromadb import Settings
+
+from App_Function_Libraries.Chunk_Lib import improved_chunking_process
+from App_Function_Libraries.DB.DB_Manager import add_media_chunk, update_fts_for_media
+from App_Function_Libraries.LLM_API_Calls import get_openai_embeddings
+
+#######################################################################################################################
+#
+# Functions for ChromaDB
+
+# Get ChromaDB settings
+# Load configuration
+config = configparser.ConfigParser()
+config.read('config.txt')
+chroma_db_path = config.get('Database', 'chroma_db_path', fallback='chroma_db')
+chroma_client = chromadb.PersistentClient(path=chroma_db_path, settings=Settings(anonymized_telemetry=False))
+
+# Get embedding settings
+embedding_provider = config.get('Embeddings', 'provider', fallback='openai')
+embedding_model = config.get('Embeddings', 'model', fallback='text-embedding-3-small')
+embedding_api_key = config.get('Embeddings', 'api_key', fallback='')
+embedding_api_url = config.get('Embeddings', 'api_url', fallback='')
+
+# Get chunking options
+chunk_options = {
+    'method': config.get('Chunking', 'method', fallback='words'),
+    'max_size': config.getint('Chunking', 'max_size', fallback=400),
+    'overlap': config.getint('Chunking', 'overlap', fallback=200),
+    'adaptive': config.getboolean('Chunking', 'adaptive', fallback=False),
+    'multi_level': config.getboolean('Chunking', 'multi_level', fallback=False),
+    'language': config.get('Chunking', 'language', fallback='english')
+}
+
+
+def auto_update_chroma_embeddings(media_id: int, content: str):
+    """
+    Automatically update ChromaDB embeddings when a new item is ingested into the SQLite database.
+
+    :param media_id: The ID of the newly ingested media item
+    :param content: The content of the newly ingested media item
+    """
+    collection_name = f"media_{media_id}"
+
+    # Initialize or get the ChromaDB collection
+    collection = chroma_client.get_or_create_collection(name=collection_name)
+
+    # Check if embeddings already exist for this media_id
+    existing_embeddings = collection.get(ids=[f"{media_id}_chunk_{i}" for i in range(len(content))])
+
+    if existing_embeddings and len(existing_embeddings) > 0:
+        logging.info(f"Embeddings already exist for media ID {media_id}, skipping...")
+    else:
+        # Process and store content if embeddings do not already exist
+        process_and_store_content(content, collection_name, media_id)
+        logging.info(f"Updated ChromaDB embeddings for media ID: {media_id}")
+
+
+# Function to process content, create chunks, embeddings, and store in ChromaDB and SQLite
+def process_and_store_content(content: str, collection_name: str, media_id: int):
+    # Process the content into chunks
+    chunks = improved_chunking_process(content, chunk_options)
+    texts = [chunk['text'] for chunk in chunks]
+
+    # Generate embeddings for each chunk
+    embeddings = [create_embedding(text) for text in texts]
+
+    # Create unique IDs for each chunk using the media_id and chunk index
+    ids = [f"{media_id}_chunk_{i}" for i in range(len(texts))]
+
+    # Store the texts, embeddings, and IDs in ChromaDB
+    store_in_chroma(collection_name, texts, embeddings, ids)
+
+    # Store the chunk metadata in SQLite
+    for i, chunk in enumerate(chunks):
+        add_media_chunk(media_id, chunk['text'], chunk['start'], chunk['end'], ids[i])
+
+    # Update the FTS table
+    update_fts_for_media(media_id)
+
+# Function to store documents and their embeddings in ChromaDB
+def store_in_chroma(collection_name: str, texts: List[str], embeddings: List[List[float]], ids: List[str]):
+    collection = chroma_client.get_or_create_collection(name=collection_name)
+    collection.add(
+        documents=texts,
+        embeddings=embeddings,
+        ids=ids
+    )
+
+# Function to perform vector search using ChromaDB
+def vector_search(collection_name: str, query: str, k: int = 10) -> List[str]:
+    query_embedding = create_embedding(query)
+    collection = chroma_client.get_collection(name=collection_name)
+    results = collection.query(
+        query_embeddings=[query_embedding],
+        n_results=k
+    )
+    return results['documents'][0]
+
+
+def create_embedding(text: str) -> List[float]:
+    global embedding_provider, embedding_model, embedding_api_url, embedding_api_key
+
+    if embedding_provider == 'openai':
+        return get_openai_embeddings(text, embedding_model)
+    elif embedding_provider == 'local':
+        response = requests.post(
+            embedding_api_url,
+            json={"text": text, "model": embedding_model},
+            headers={"Authorization": f"Bearer {embedding_api_key}"}
+        )
+        return response.json()['embedding']
+    elif embedding_provider == 'huggingface':
+        from transformers import AutoTokenizer, AutoModel
+        import torch
+
+        tokenizer = AutoTokenizer.from_pretrained(embedding_model)
+        model = AutoModel.from_pretrained(embedding_model)
+
+        inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True, max_length=512)
+        with torch.no_grad():
+            outputs = model(**inputs)
+
+        # Use the mean of the last hidden state as the sentence embedding
+        embeddings = outputs.last_hidden_state.mean(dim=1)
+        return embeddings[0].tolist()  # Convert to list for consistency
+    else:
+        raise ValueError(f"Unsupported embedding provider: {embedding_provider}")
+
+
+def create_all_embeddings(api_choice: str, model_or_url: str) -> str:
+    try:
+        all_content = get_all_content_from_database()
+
+        if not all_content:
+            return "No content found in the database."
+
+        texts_to_embed = []
+        embeddings_to_store = []
+        ids_to_store = []
+        collection_name = "all_content_embeddings"
+
+        # Initialize or get the ChromaDB collection
+        collection = chroma_client.get_or_create_collection(name=collection_name)
+
+        for content_item in all_content:
+            media_id = content_item['id']
+            text = content_item['content']
+
+            # Check if the embedding already exists in ChromaDB
+            embedding_exists = collection.get(ids=[f"doc_{media_id}"])
+
+            if embedding_exists:
+                logging.info(f"Embedding already exists for media ID {media_id}, skipping...")
+                continue  # Skip if embedding already exists
+
+            # Create the embedding
+            if api_choice == "openai":
+                embedding = create_openai_embedding(text, model_or_url)
+            else:  # Llama.cpp
+                embedding = create_llamacpp_embedding(text, model_or_url)
+
+            # Collect the text, embedding, and ID for batch storage
+            texts_to_embed.append(text)
+            embeddings_to_store.append(embedding)
+            ids_to_store.append(f"doc_{media_id}")
+
+        # Store all new embeddings in ChromaDB
+        if texts_to_embed and embeddings_to_store:
+            store_in_chroma(collection_name, texts_to_embed, embeddings_to_store, ids_to_store)
+
+        return "Embeddings created and stored successfully for all new content."
+    except Exception as e:
+        logging.error(f"Error during embedding creation: {str(e)}")
+        return f"Error: {str(e)}"
+
+
+def create_openai_embedding(text: str, model: str) -> List[float]:
+    openai_api_key = config['API']['openai_api_key']
+    embedding = get_openai_embeddings(text, model)
+    return embedding
+
+
+def create_llamacpp_embedding(text: str, api_url: str) -> List[float]:
+    response = requests.post(
+        api_url,
+        json={"input": text}
+    )
+    if response.status_code == 200:
+        return response.json()['embedding']
+    else:
+        raise Exception(f"Error from Llama.cpp API: {response.text}")
+
+
+def get_all_content_from_database() -> List[Dict[str, Any]]:
+    """
+    Retrieve all media content from the database that requires embedding.
+
+    Returns:
+        List[Dict[str, Any]]: A list of dictionaries, each containing the media ID, content, title, and other relevant fields.
+    """
+    try:
+        from App_Function_Libraries.DB.DB_Manager import db
+        with db.get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute("""
+                SELECT id, content, title, author, type
+                FROM Media
+                WHERE is_trash = 0  -- Exclude items marked as trash
+            """)
+            media_items = cursor.fetchall()
+
+            # Convert the results into a list of dictionaries
+            all_content = [
+                {
+                    'id': item[0],
+                    'content': item[1],
+                    'title': item[2],
+                    'author': item[3],
+                    'type': item[4]
+                }
+                for item in media_items
+            ]
+
+        return all_content
+
+    except sqlite3.Error as e:
+        logging.error(f"Error retrieving all content from database: {e}")
+        from App_Function_Libraries.DB.SQLite_DB import DatabaseError
+        raise DatabaseError(f"Error retrieving all content from database: {e}")
+
+
+def store_in_chroma_with_citation(collection_name: str, texts: List[str], embeddings: List[List[float]], ids: List[str], sources: List[str]):
+    collection = chroma_client.get_or_create_collection(name=collection_name)
+    collection.add(
+        documents=texts,
+        embeddings=embeddings,
+        ids=ids,
+        metadatas=[{'source': source} for source in sources]
+    )
+
+
+def check_embedding_status(selected_item):
+    if not selected_item:
+        return "Please select an item", ""
+    item_id = selected_item.split('(')[0].strip()
+    collection = chroma_client.get_or_create_collection(name="all_content_embeddings")
+    result = collection.get(ids=[f"doc_{item_id}"])
+    if result['ids']:
+        embedding = result['embeddings'][0]
+        embedding_preview = str(embedding[:50])  # Convert first 50 elements to string
+        return f"Embedding exists for item: {item_id}", f"Embedding preview: {embedding_preview}..."
+    else:
+        return f"No embedding found for item: {item_id}", ""
+
+
+def create_new_embedding(selected_item, api_choice, openai_model, llamacpp_url):
+    if not selected_item:
+        return "Please select an item"
+    item_id = selected_item.split('(')[0].strip()
+    items = get_all_content_from_database()
+    item = next((item for item in items if item['title'] == item_id), None)
+    if not item:
+        return f"Item not found: {item_id}"
+
+    try:
+        if api_choice == "OpenAI":
+            embedding = create_embedding(item['content'])
+        else:  # Llama.cpp
+            embedding = create_embedding(item['content'])
+
+        collection_name = "all_content_embeddings"
+        store_in_chroma(collection_name, [item['content']], [embedding], [f"doc_{item['id']}"])
+        return f"New embedding created and stored for item: {item_id}"
+    except Exception as e:
+        return f"Error creating embedding: {str(e)}"
+
+
+#
+# End of Functions for ChromaDB
+#######################################################################################################################

App_Function_Libraries/RAG/RAG_Examples.md

@@ -0,0 +1,556 @@
+
+```
+##################################################################################################################
+# RAG Pipeline 1
+# 0.62    0.61    0.75    63402.0
+# from langchain_openai import ChatOpenAI
+#
+# from langchain_community.document_loaders import WebBaseLoader
+# from langchain_openai import OpenAIEmbeddings
+# from langchain.text_splitter import RecursiveCharacterTextSplitter
+# from langchain_chroma import Chroma
+#
+# from langchain_community.retrievers import BM25Retriever
+# from langchain.retrievers import ParentDocumentRetriever
+# from langchain.storage import InMemoryStore
+# import os
+# from operator import itemgetter
+# from langchain import hub
+# from langchain_core.output_parsers import StrOutputParser
+# from langchain_core.runnables import RunnablePassthrough, RunnableParallel, RunnableLambda
+# from langchain.retrievers import MergerRetriever
+# from langchain.retrievers.document_compressors import DocumentCompressorPipeline
+
+
+# def rag_pipeline():
+#     try:
+#         def format_docs(docs):
+#             return "\n".join(doc.page_content for doc in docs)
+#
+#         llm = ChatOpenAI(model='gpt-4o-mini')
+#
+#         loader = WebBaseLoader('https://en.wikipedia.org/wiki/European_debt_crisis')
+#         docs = loader.load()
+#
+#         embedding = OpenAIEmbeddings(model='text-embedding-3-large')
+#
+#         splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=200)
+#         splits = splitter.split_documents(docs)
+#         c = Chroma.from_documents(documents=splits, embedding=embedding,
+#                                   collection_name='testindex-ragbuilder-1724657573', )
+#         retrievers = []
+#         retriever = c.as_retriever(search_type='mmr', search_kwargs={'k': 10})
+#         retrievers.append(retriever)
+#         retriever = BM25Retriever.from_documents(docs)
+#         retrievers.append(retriever)
+#
+#         parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1200, chunk_overlap=600)
+#         splits = parent_splitter.split_documents(docs)
+#         store = InMemoryStore()
+#         retriever = ParentDocumentRetriever(vectorstore=c, docstore=store, child_splitter=splitter,
+#                                             parent_splitter=parent_splitter)
+#         retriever.add_documents(docs)
+#         retrievers.append(retriever)
+#         retriever = MergerRetriever(retrievers=retrievers)
+#         prompt = hub.pull("rlm/rag-prompt")
+#         rag_chain = (
+#             RunnableParallel(context=retriever, question=RunnablePassthrough())
+#             .assign(context=itemgetter("context") | RunnableLambda(format_docs))
+#             .assign(answer=prompt | llm | StrOutputParser())
+#             .pick(["answer", "context"]))
+#         return rag_chain
+#     except Exception as e:
+#         print(f"An error occurred: {e}")
+
+
+# To get the answer and context, use the following code
+# res=rag_pipeline().invoke("your prompt here")
+# print(res["answer"])
+# print(res["context"])
+
+############################################################################################################
+
+
+############################################################################################################
+# RAG Pipeline 2
+
+# 0.6     0.73    0.68    3125.0
+# from langchain_openai import ChatOpenAI
+#
+# from langchain_community.document_loaders import WebBaseLoader
+# from langchain_openai import OpenAIEmbeddings
+# from langchain.text_splitter import RecursiveCharacterTextSplitter
+# from langchain_chroma import Chroma
+# from langchain.retrievers.multi_query import MultiQueryRetriever
+# from langchain.retrievers import ParentDocumentRetriever
+# from langchain.storage import InMemoryStore
+# from langchain_community.document_transformers import EmbeddingsRedundantFilter
+# from langchain.retrievers.document_compressors import LLMChainFilter
+# from langchain.retrievers.document_compressors import EmbeddingsFilter
+# from langchain.retrievers import ContextualCompressionRetriever
+# import os
+# from operator import itemgetter
+# from langchain import hub
+# from langchain_core.output_parsers import StrOutputParser
+# from langchain_core.runnables import RunnablePassthrough, RunnableParallel, RunnableLambda
+# from langchain.retrievers import MergerRetriever
+# from langchain.retrievers.document_compressors import DocumentCompressorPipeline
+
+
+# def rag_pipeline():
+#     try:
+#         def format_docs(docs):
+#             return "\n".join(doc.page_content for doc in docs)
+#
+#         llm = ChatOpenAI(model='gpt-4o-mini')
+#
+#         loader = WebBaseLoader('https://en.wikipedia.org/wiki/European_debt_crisis')
+#         docs = loader.load()
+#
+#         embedding = OpenAIEmbeddings(model='text-embedding-3-large')
+#
+#         splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=200)
+#         splits = splitter.split_documents(docs)
+#         c = Chroma.from_documents(documents=splits, embedding=embedding,
+#                                   collection_name='testindex-ragbuilder-1724650962', )
+#         retrievers = []
+#         retriever = MultiQueryRetriever.from_llm(c.as_retriever(search_type='similarity', search_kwargs={'k': 10}),
+#                                                  llm=llm)
+#         retrievers.append(retriever)
+#
+#         parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1200, chunk_overlap=600)
+#         splits = parent_splitter.split_documents(docs)
+#         store = InMemoryStore()
+#         retriever = ParentDocumentRetriever(vectorstore=c, docstore=store, child_splitter=splitter,
+#                                             parent_splitter=parent_splitter)
+#         retriever.add_documents(docs)
+#         retrievers.append(retriever)
+#         retriever = MergerRetriever(retrievers=retrievers)
+#         arr_comp = []
+#         arr_comp.append(EmbeddingsRedundantFilter(embeddings=embedding))
+#         arr_comp.append(LLMChainFilter.from_llm(llm))
+#         pipeline_compressor = DocumentCompressorPipeline(transformers=arr_comp)
+#         retriever = ContextualCompressionRetriever(base_retriever=retriever, base_compressor=pipeline_compressor)
+#         prompt = hub.pull("rlm/rag-prompt")
+#         rag_chain = (
+#             RunnableParallel(context=retriever, question=RunnablePassthrough())
+#             .assign(context=itemgetter("context") | RunnableLambda(format_docs))
+#             .assign(answer=prompt | llm | StrOutputParser())
+#             .pick(["answer", "context"]))
+#         return rag_chain
+#     except Exception as e:
+#         print(f"An error occurred: {e}")
+
+
+# To get the answer and context, use the following code
+# res=rag_pipeline().invoke("your prompt here")
+# print(res["answer"])
+# print(res["context"])
+
+#
+#
+#
+############################################################################################################
+# Plain bm25 retriever
+# class BM25Retriever(BaseRetriever):
+#     """`BM25` retriever without Elasticsearch."""
+#
+#     vectorizer: Any
+#     """ BM25 vectorizer."""
+#     docs: List[Document] = Field(repr=False)
+#     """ List of documents."""
+#     k: int = 4
+#     """ Number of documents to return."""
+#     preprocess_func: Callable[[str], List[str]] = default_preprocessing_func
+#     """ Preprocessing function to use on the text before BM25 vectorization."""
+#
+#     class Config:
+#         arbitrary_types_allowed = True
+#
+#     @classmethod
+#     def from_texts(
+#         cls,
+#         texts: Iterable[str],
+#         metadatas: Optional[Iterable[dict]] = None,
+#         bm25_params: Optional[Dict[str, Any]] = None,
+#         preprocess_func: Callable[[str], List[str]] = default_preprocessing_func,
+#         **kwargs: Any,
+#     ) -> BM25Retriever:
+#         """
+#         Create a BM25Retriever from a list of texts.
+#         Args:
+#             texts: A list of texts to vectorize.
+#             metadatas: A list of metadata dicts to associate with each text.
+#             bm25_params: Parameters to pass to the BM25 vectorizer.
+#             preprocess_func: A function to preprocess each text before vectorization.
+#             **kwargs: Any other arguments to pass to the retriever.
+#
+#         Returns:
+#             A BM25Retriever instance.
+#         """
+#         try:
+#             from rank_bm25 import BM25Okapi
+#         except ImportError:
+#             raise ImportError(
+#                 "Could not import rank_bm25, please install with `pip install "
+#                 "rank_bm25`."
+#             )
+#
+#         texts_processed = [preprocess_func(t) for t in texts]
+#         bm25_params = bm25_params or {}
+#         vectorizer = BM25Okapi(texts_processed, **bm25_params)
+#         metadatas = metadatas or ({} for _ in texts)
+#         docs = [Document(page_content=t, metadata=m) for t, m in zip(texts, metadatas)]
+#         return cls(
+#             vectorizer=vectorizer, docs=docs, preprocess_func=preprocess_func, **kwargs
+#         )
+#
+#     @classmethod
+#     def from_documents(
+#         cls,
+#         documents: Iterable[Document],
+#         *,
+#         bm25_params: Optional[Dict[str, Any]] = None,
+#         preprocess_func: Callable[[str], List[str]] = default_preprocessing_func,
+#         **kwargs: Any,
+#     ) -> BM25Retriever:
+#         """
+#         Create a BM25Retriever from a list of Documents.
+#         Args:
+#             documents: A list of Documents to vectorize.
+#             bm25_params: Parameters to pass to the BM25 vectorizer.
+#             preprocess_func: A function to preprocess each text before vectorization.
+#             **kwargs: Any other arguments to pass to the retriever.
+#
+#         Returns:
+#             A BM25Retriever instance.
+#         """
+#         texts, metadatas = zip(*((d.page_content, d.metadata) for d in documents))
+#         return cls.from_texts(
+#             texts=texts,
+#             bm25_params=bm25_params,
+#             metadatas=metadatas,
+#             preprocess_func=preprocess_func,
+#             **kwargs,
+#         )
+#
+#     def _get_relevant_documents(
+#         self, query: str, *, run_manager: CallbackManagerForRetrieverRun
+#     ) -> List[Document]:
+#         processed_query = self.preprocess_func(query)
+#         return_docs = self.vectorizer.get_top_n(processed_query, self.docs, n=self.k)
+#         return return_docs
+############################################################################################################
+
+############################################################################################################
+# ElasticSearch BM25 Retriever
+# class ElasticSearchBM25Retriever(BaseRetriever):
+#     """`Elasticsearch` retriever that uses `BM25`.
+#
+#     To connect to an Elasticsearch instance that requires login credentials,
+#     including Elastic Cloud, use the Elasticsearch URL format
+#     https://username:password@es_host:9243. For example, to connect to Elastic
+#     Cloud, create the Elasticsearch URL with the required authentication details and
+#     pass it to the ElasticVectorSearch constructor as the named parameter
+#     elasticsearch_url.
+#
+#     You can obtain your Elastic Cloud URL and login credentials by logging in to the
+#     Elastic Cloud console at https://cloud.elastic.co, selecting your deployment, and
+#     navigating to the "Deployments" page.
+#
+#     To obtain your Elastic Cloud password for the default "elastic" user:
+#
+#     1. Log in to the Elastic Cloud console at https://cloud.elastic.co
+#     2. Go to "Security" > "Users"
+#     3. Locate the "elastic" user and click "Edit"
+#     4. Click "Reset password"
+#     5. Follow the prompts to reset the password
+#
+#     The format for Elastic Cloud URLs is
+#     https://username:password@cluster_id.region_id.gcp.cloud.es.io:9243.
+#     """
+#
+#     client: Any
+#     """Elasticsearch client."""
+#     index_name: str
+#     """Name of the index to use in Elasticsearch."""
+#
+#     @classmethod
+#     def create(
+#         cls, elasticsearch_url: str, index_name: str, k1: float = 2.0, b: float = 0.75
+#     ) -> ElasticSearchBM25Retriever:
+#         """
+#         Create a ElasticSearchBM25Retriever from a list of texts.
+#
+#         Args:
+#             elasticsearch_url: URL of the Elasticsearch instance to connect to.
+#             index_name: Name of the index to use in Elasticsearch.
+#             k1: BM25 parameter k1.
+#             b: BM25 parameter b.
+#
+#         Returns:
+#
+#         """
+#         from elasticsearch import Elasticsearch
+#
+#         # Create an Elasticsearch client instance
+#         es = Elasticsearch(elasticsearch_url)
+#
+#         # Define the index settings and mappings
+#         settings = {
+#             "analysis": {"analyzer": {"default": {"type": "standard"}}},
+#             "similarity": {
+#                 "custom_bm25": {
+#                     "type": "BM25",
+#                     "k1": k1,
+#                     "b": b,
+#                 }
+#             },
+#         }
+#         mappings = {
+#             "properties": {
+#                 "content": {
+#                     "type": "text",
+#                     "similarity": "custom_bm25",  # Use the custom BM25 similarity
+#                 }
+#             }
+#         }
+#
+#         # Create the index with the specified settings and mappings
+#         es.indices.create(index=index_name, mappings=mappings, settings=settings)
+#         return cls(client=es, index_name=index_name)
+#
+#     def add_texts(
+#         self,
+#         texts: Iterable[str],
+#         refresh_indices: bool = True,
+#     ) -> List[str]:
+#         """Run more texts through the embeddings and add to the retriever.
+#
+#         Args:
+#             texts: Iterable of strings to add to the retriever.
+#             refresh_indices: bool to refresh ElasticSearch indices
+#
+#         Returns:
+#             List of ids from adding the texts into the retriever.
+#         """
+#         try:
+#             from elasticsearch.helpers import bulk
+#         except ImportError:
+#             raise ImportError(
+#                 "Could not import elasticsearch python package. "
+#                 "Please install it with `pip install elasticsearch`."
+#             )
+#         requests = []
+#         ids = []
+#         for i, text in enumerate(texts):
+#             _id = str(uuid.uuid4())
+#             request = {
+#                 "_op_type": "index",
+#                 "_index": self.index_name,
+#                 "content": text,
+#                 "_id": _id,
+#             }
+#             ids.append(_id)
+#             requests.append(request)
+#         bulk(self.client, requests)
+#
+#         if refresh_indices:
+#             self.client.indices.refresh(index=self.index_name)
+#         return ids
+#
+#     def _get_relevant_documents(
+#         self, query: str, *, run_manager: CallbackManagerForRetrieverRun
+#     ) -> List[Document]:
+#         query_dict = {"query": {"match": {"content": query}}}
+#         res = self.client.search(index=self.index_name, body=query_dict)
+#
+#         docs = []
+#         for r in res["hits"]["hits"]:
+#             docs.append(Document(page_content=r["_source"]["content"]))
+#         return docs
+############################################################################################################
+
+
+############################################################################################################
+# Multi Query Retriever
+# class MultiQueryRetriever(BaseRetriever):
+#     """Given a query, use an LLM to write a set of queries.
+#
+#     Retrieve docs for each query. Return the unique union of all retrieved docs.
+#     """
+#
+#     retriever: BaseRetriever
+#     llm_chain: Runnable
+#     verbose: bool = True
+#     parser_key: str = "lines"
+#     """DEPRECATED. parser_key is no longer used and should not be specified."""
+#     include_original: bool = False
+#     """Whether to include the original query in the list of generated queries."""
+#
+#     @classmethod
+#     def from_llm(
+#         cls,
+#         retriever: BaseRetriever,
+#         llm: BaseLanguageModel,
+#         prompt: BasePromptTemplate = DEFAULT_QUERY_PROMPT,
+#         parser_key: Optional[str] = None,
+#         include_original: bool = False,
+#     ) -> "MultiQueryRetriever":
+#         """Initialize from llm using default template.
+#
+#         Args:
+#             retriever: retriever to query documents from
+#             llm: llm for query generation using DEFAULT_QUERY_PROMPT
+#             prompt: The prompt which aims to generate several different versions
+#                 of the given user query
+#             include_original: Whether to include the original query in the list of
+#                 generated queries.
+#
+#         Returns:
+#             MultiQueryRetriever
+#         """
+#         output_parser = LineListOutputParser()
+#         llm_chain = prompt | llm | output_parser
+#         return cls(
+#             retriever=retriever,
+#             llm_chain=llm_chain,
+#             include_original=include_original,
+#         )
+#
+#     async def _aget_relevant_documents(
+#         self,
+#         query: str,
+#         *,
+#         run_manager: AsyncCallbackManagerForRetrieverRun,
+#     ) -> List[Document]:
+#         """Get relevant documents given a user query.
+#
+#         Args:
+#             query: user query
+#
+#         Returns:
+#             Unique union of relevant documents from all generated queries
+#         """
+#         queries = await self.agenerate_queries(query, run_manager)
+#         if self.include_original:
+#             queries.append(query)
+#         documents = await self.aretrieve_documents(queries, run_manager)
+#         return self.unique_union(documents)
+#
+#     async def agenerate_queries(
+#         self, question: str, run_manager: AsyncCallbackManagerForRetrieverRun
+#     ) -> List[str]:
+#         """Generate queries based upon user input.
+#
+#         Args:
+#             question: user query
+#
+#         Returns:
+#             List of LLM generated queries that are similar to the user input
+#         """
+#         response = await self.llm_chain.ainvoke(
+#             {"question": question}, config={"callbacks": run_manager.get_child()}
+#         )
+#         if isinstance(self.llm_chain, LLMChain):
+#             lines = response["text"]
+#         else:
+#             lines = response
+#         if self.verbose:
+#             logger.info(f"Generated queries: {lines}")
+#         return lines
+#
+#     async def aretrieve_documents(
+#         self, queries: List[str], run_manager: AsyncCallbackManagerForRetrieverRun
+#     ) -> List[Document]:
+#         """Run all LLM generated queries.
+#
+#         Args:
+#             queries: query list
+#
+#         Returns:
+#             List of retrieved Documents
+#         """
+#         document_lists = await asyncio.gather(
+#             *(
+#                 self.retriever.ainvoke(
+#                     query, config={"callbacks": run_manager.get_child()}
+#                 )
+#                 for query in queries
+#             )
+#         )
+#         return [doc for docs in document_lists for doc in docs]
+#
+#     def _get_relevant_documents(
+#         self,
+#         query: str,
+#         *,
+#         run_manager: CallbackManagerForRetrieverRun,
+#     ) -> List[Document]:
+#         """Get relevant documents given a user query.
+#
+#         Args:
+#             query: user query
+#
+#         Returns:
+#             Unique union of relevant documents from all generated queries
+#         """
+#         queries = self.generate_queries(query, run_manager)
+#         if self.include_original:
+#             queries.append(query)
+#         documents = self.retrieve_documents(queries, run_manager)
+#         return self.unique_union(documents)
+#
+#     def generate_queries(
+#         self, question: str, run_manager: CallbackManagerForRetrieverRun
+#     ) -> List[str]:
+#         """Generate queries based upon user input.
+#
+#         Args:
+#             question: user query
+#
+#         Returns:
+#             List of LLM generated queries that are similar to the user input
+#         """
+#         response = self.llm_chain.invoke(
+#             {"question": question}, config={"callbacks": run_manager.get_child()}
+#         )
+#         if isinstance(self.llm_chain, LLMChain):
+#             lines = response["text"]
+#         else:
+#             lines = response
+#         if self.verbose:
+#             logger.info(f"Generated queries: {lines}")
+#         return lines
+#
+#     def retrieve_documents(
+#         self, queries: List[str], run_manager: CallbackManagerForRetrieverRun
+#     ) -> List[Document]:
+#         """Run all LLM generated queries.
+#
+#         Args:
+#             queries: query list
+#
+#         Returns:
+#             List of retrieved Documents
+#         """
+#         documents = []
+#         for query in queries:
+#             docs = self.retriever.invoke(
+#                 query, config={"callbacks": run_manager.get_child()}
+#             )
+#             documents.extend(docs)
+#         return documents
+#
+#     def unique_union(self, documents: List[Document]) -> List[Document]:
+#         """Get unique Documents.
+#
+#         Args:
+#             documents: List of retrieved Documents
+#
+#         Returns:
+#             List of unique retrieved Documents
+#         """
+#         return _unique_documents(documents)
+############################################################################################################
+```

App_Function_Libraries/RAG/RAG_Libary_2.py

@@ -0,0 +1,172 @@
+# RAG_Library_2.py
+# Description: This script contains the main RAG pipeline function and related functions for the RAG pipeline.
+#
+# Import necessary modules and functions
+import configparser
+from typing import Dict, Any
+# Local Imports
+from App_Function_Libraries.RAG.ChromaDB_Library import process_and_store_content, vector_search, chroma_client
+from App_Function_Libraries.Article_Extractor_Lib import scrape_article
+from App_Function_Libraries.DB.DB_Manager import add_media_to_database, search_db, get_unprocessed_media
+# 3rd-Party Imports
+import openai
+#
+########################################################################################################################
+#
+# Functions:
+
+# Initialize OpenAI client (adjust this based on your API key management)
+openai.api_key = "your-openai-api-key"
+
+config = configparser.ConfigParser()
+config.read('config.txt')
+
+# Main RAG pipeline function
+def rag_pipeline(url: str, query: str, api_choice=None) -> Dict[str, Any]:
+    # Extract content
+    article_data = scrape_article(url)
+    content = article_data['content']
+    title = article_data['title']
+
+    # Store the article in the database and get the media_id
+    media_id = add_media_to_database(url, title, 'article', content)
+
+    # Process and store content
+    collection_name = f"article_{media_id}"
+    process_and_store_content(content, collection_name, media_id)
+
+    # Perform searches
+    vector_results = vector_search(collection_name, query, k=5)
+    fts_results = search_db(query, ["content"], "", page=1, results_per_page=5)
+
+    # Combine results
+    all_results = vector_results + [result['content'] for result in fts_results]
+    context = "\n".join(all_results)
+
+    # Generate answer using the selected API
+    answer = generate_answer(api_choice, context, query)
+
+    return {
+        "answer": answer,
+        "context": context
+    }
+
+
+def generate_answer(api_choice: str, context: str, query: str) -> str:
+    prompt = f"Context: {context}\n\nQuestion: {query}"
+    if api_choice == "OpenAI":
+        from App_Function_Libraries.Summarization_General_Lib import summarize_with_openai
+        return summarize_with_openai(config['API']['openai_api_key'], prompt, "")
+    elif api_choice == "Anthropic":
+        from App_Function_Libraries.Summarization_General_Lib import summarize_with_anthropic
+        return summarize_with_anthropic(config['API']['anthropic_api_key'], prompt, "")
+    elif api_choice == "Cohere":
+        from App_Function_Libraries.Summarization_General_Lib import summarize_with_cohere
+        return summarize_with_cohere(config['API']['cohere_api_key'], prompt, "")
+    elif api_choice == "Groq":
+        from App_Function_Libraries.Summarization_General_Lib import summarize_with_groq
+        return summarize_with_groq(config['API']['groq_api_key'], prompt, "")
+    elif api_choice == "OpenRouter":
+        from App_Function_Libraries.Summarization_General_Lib import summarize_with_openrouter
+        return summarize_with_openrouter(config['API']['openrouter_api_key'], prompt, "")
+    elif api_choice == "HuggingFace":
+        from App_Function_Libraries.Summarization_General_Lib import summarize_with_huggingface
+        return summarize_with_huggingface(config['API']['huggingface_api_key'], prompt, "")
+    elif api_choice == "DeepSeek":
+        from App_Function_Libraries.Summarization_General_Lib import summarize_with_deepseek
+        return summarize_with_deepseek(config['API']['deepseek_api_key'], prompt, "")
+    elif api_choice == "Mistral":
+        from App_Function_Libraries.Summarization_General_Lib import summarize_with_mistral
+        return summarize_with_mistral(config['API']['mistral_api_key'], prompt, "")
+    elif api_choice == "Local-LLM":
+        from App_Function_Libraries.Local_Summarization_Lib import summarize_with_local_llm
+        return summarize_with_local_llm(config['API']['local_llm_path'], prompt, "")
+    elif api_choice == "Llama.cpp":
+        from App_Function_Libraries.Local_Summarization_Lib import summarize_with_llama
+        return summarize_with_llama(config['API']['llama_api_key'], prompt, "")
+    elif api_choice == "Kobold":
+        from App_Function_Libraries.Local_Summarization_Lib import summarize_with_kobold
+        return summarize_with_kobold(config['API']['kobold_api_key'], prompt, "")
+    elif api_choice == "Ooba":
+        from App_Function_Libraries.Local_Summarization_Lib import summarize_with_oobabooga
+        return summarize_with_oobabooga(config['API']['ooba_api_key'], prompt, "")
+    elif api_choice == "TabbyAPI":
+        from App_Function_Libraries.Local_Summarization_Lib import summarize_with_tabbyapi
+        return summarize_with_tabbyapi(config['API']['tabby_api_key'], prompt, "")
+    elif api_choice == "vLLM":
+        from App_Function_Libraries.Local_Summarization_Lib import summarize_with_vllm
+        return summarize_with_vllm(config['API']['vllm_api_key'], prompt, "")
+    elif api_choice == "ollama":
+        from App_Function_Libraries.Local_Summarization_Lib import summarize_with_ollama
+        return summarize_with_ollama(config['API']['ollama_api_key'], prompt, "")
+    else:
+        raise ValueError(f"Unsupported API choice: {api_choice}")
+
+# Function to preprocess and store all existing content in the database
+def preprocess_all_content():
+    unprocessed_media = get_unprocessed_media()
+    for row in unprocessed_media:
+        media_id = row[0]
+        content = row[1]
+        media_type = row[2]
+        collection_name = f"{media_type}_{media_id}"
+        process_and_store_content(content, collection_name, media_id)
+
+
+# Function to perform RAG search across all stored content
+def rag_search(query: str, api_choice: str) -> Dict[str, Any]:
+    # Perform vector search across all collections
+    all_collections = chroma_client.list_collections()
+    vector_results = []
+    for collection in all_collections:
+        vector_results.extend(vector_search(collection.name, query, k=2))
+
+    # Perform FTS search
+    fts_results = search_db(query, ["content"], "", page=1, results_per_page=10)
+
+    # Combine results
+    all_results = vector_results + [result['content'] for result in fts_results]
+    context = "\n".join(all_results[:10])  # Limit to top 10 results
+
+    # Generate answer using the selected API
+    answer = generate_answer(api_choice, context, query)
+
+    return {
+        "answer": answer,
+        "context": context
+    }
+
+
+# Example usage:
+# 1. Initialize the system:
+# create_tables(db)  # Ensure FTS tables are set up
+#
+# 2. Create ChromaDB
+# chroma_client = ChromaDBClient()
+#
+# 3. Create Embeddings
+# Store embeddings in ChromaDB
+# preprocess_all_content() or create_embeddings()
+#
+# 4. Perform RAG search across all content:
+# result = rag_search("What are the key points about climate change?")
+# print(result['answer'])
+#
+# (Extra)5. Perform RAG on a specific URL:
+# result = rag_pipeline("https://example.com/article", "What is the main topic of this article?")
+# print(result['answer'])
+#
+########################################################################################################################
+
+
+############################################################################################################
+#
+# ElasticSearch Retriever
+
+# https://github.com/langchain-ai/langchain/tree/44e3e2391c48bfd0a8e6a20adde0b6567f4f43c3/templates/rag-elasticsearch
+#
+# https://github.com/langchain-ai/langchain/tree/44e3e2391c48bfd0a8e6a20adde0b6567f4f43c3/templates/rag-self-query
+
+#
+# End of RAG_Library_2.py
+############################################################################################################

App_Function_Libraries/RAG/RAG_Library.py

@@ -0,0 +1,396 @@
+import numpy as np
+from typing import List, Tuple, Dict
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.metrics.pairwise import cosine_similarity
+from sentence_transformers import SentenceTransformer
+import math
+from functools import lru_cache
+from concurrent.futures import ThreadPoolExecutor
+import openai
+from transformers import T5ForConditionalGeneration, T5Tokenizer
+import torch
+import re
+import psycopg2
+from psycopg2.extras import execute_values
+import sqlite3
+import logging
+
+
+
+########################################################################################################################################################################################################################################
+#
+# RAG Chunking
+# To fully integrate this chunking system, you'd need to:
+#
+# Create the UnvectorizedMediaChunks table in your SQLite database.
+# Modify your document ingestion process to use chunk_and_store_unvectorized.
+# Implement a background process that periodically calls vectorize_all_documents to process unvectorized chunks.
+
+# This chunking is pretty weak and needs improvement
+# See notes for improvements #FIXME
+import json
+from typing import List, Dict, Any
+from datetime import datetime
+
+
+def chunk_and_store_unvectorized(
+        db_connection,
+        media_id: int,
+        text: str,
+        chunk_size: int = 1000,
+        overlap: int = 100,
+        chunk_type: str = 'fixed-length'
+) -> List[int]:
+    chunks = create_chunks(text, chunk_size, overlap)
+    return store_unvectorized_chunks(db_connection, media_id, chunks, chunk_type)
+
+
+def create_chunks(text: str, chunk_size: int, overlap: int) -> List[Dict[str, Any]]:
+    words = text.split()
+    chunks = []
+    for i in range(0, len(words), chunk_size - overlap):
+        chunk_text = ' '.join(words[i:i + chunk_size])
+        start_char = text.index(words[i])
+        end_char = start_char + len(chunk_text)
+        chunks.append({
+            'text': chunk_text,
+            'start_char': start_char,
+            'end_char': end_char,
+            'index': len(chunks)
+        })
+    return chunks
+
+
+def store_unvectorized_chunks(
+        db_connection,
+        media_id: int,
+        chunks: List[Dict[str, Any]],
+        chunk_type: str
+) -> List[int]:
+    cursor = db_connection.cursor()
+    chunk_ids = []
+    for chunk in chunks:
+        cursor.execute("""
+            INSERT INTO UnvectorizedMediaChunks 
+            (media_id, chunk_text, chunk_index, start_char, end_char, chunk_type, metadata)
+            VALUES (?, ?, ?, ?, ?, ?, ?)
+        """, (
+            media_id,
+            chunk['text'],
+            chunk['index'],
+            chunk['start_char'],
+            chunk['end_char'],
+            chunk_type,
+            json.dumps({'length': len(chunk['text'])})  # Example metadata
+        ))
+        chunk_ids.append(cursor.lastrowid)
+    db_connection.commit()
+    return chunk_ids
+
+
+def get_unvectorized_chunks(
+        db_connection,
+        media_id: int,
+        limit: int = 100,
+        offset: int = 0
+) -> List[Dict[str, Any]]:
+    cursor = db_connection.cursor()
+    cursor.execute("""
+        SELECT id, chunk_text, chunk_index, start_char, end_char, chunk_type, metadata
+        FROM UnvectorizedMediaChunks
+        WHERE media_id = ? AND is_processed = FALSE
+        ORDER BY chunk_index
+        LIMIT ? OFFSET ?
+    """, (media_id, limit, offset))
+    return [
+        {
+            'id': row[0],
+            'text': row[1],
+            'index': row[2],
+            'start_char': row[3],
+            'end_char': row[4],
+            'type': row[5],
+            'metadata': json.loads(row[6])
+        }
+        for row in cursor.fetchall()
+    ]
+
+
+def mark_chunks_as_processed(db_connection, chunk_ids: List[int]):
+    cursor = db_connection.cursor()
+    cursor.executemany("""
+        UPDATE UnvectorizedMediaChunks
+        SET is_processed = TRUE, last_modified = ?
+        WHERE id = ?
+    """, [(datetime.now(), chunk_id) for chunk_id in chunk_ids])
+    db_connection.commit()
+
+
+# Usage example
+def process_media_chunks(db_connection, media_id: int, text: str):
+    chunk_ids = chunk_and_store_unvectorized(db_connection, media_id, text)
+    print(f"Stored {len(chunk_ids)} unvectorized chunks for media_id {media_id}")
+
+    # Later, when you want to process these chunks:
+    unprocessed_chunks = get_unvectorized_chunks(db_connection, media_id)
+    # Process chunks (e.g., vectorize them)
+    # ...
+    # After processing, mark them as processed
+    mark_chunks_as_processed(db_connection, [chunk['id'] for chunk in unprocessed_chunks])
+###########################################################################################################################################################################################################
+#
+# RAG System
+
+# To use this updated RAG system in your existing application:
+#
+# Install required packages:
+# pip install sentence-transformers psycopg2-binary scikit-learn transformers torch
+# Set up PostgreSQL with pgvector:
+#
+# Install PostgreSQL and the pgvector extension.
+# Create a new database for vector storage.
+#
+# Update your main application to use the RAG system:
+#
+# Import the RAGSystem class from this new file.
+# Initialize the RAG system with your SQLite and PostgreSQL configurations.
+# Use the vectorize_all_documents method to initially vectorize your existing documents.
+#
+#
+# Modify your existing PDF_Ingestion_Lib.py and Book_Ingestion_Lib.py:
+#
+# After successfully ingesting a document into SQLite, call the vectorization method from the RAG system.
+
+# Example modification for ingest_text_file in Book_Ingestion_Lib.py:
+# from RAG_Library import RAGSystem
+#
+# # Initialize RAG system (do this once in your main application)
+# rag_system = RAGSystem(sqlite_path, pg_config)
+#
+# def ingest_text_file(file_path, title=None, author=None, keywords=None):
+#     try:
+#         # ... (existing code)
+#
+#         # Add the text file to the database
+#         doc_id = add_media_with_keywords(
+#             url=file_path,
+#             title=title,
+#             media_type='document',
+#             content=content,
+#             keywords=keywords,
+#             prompt='No prompt for text files',
+#             summary='No summary for text files',
+#             transcription_model='None',
+#             author=author,
+#             ingestion_date=datetime.now().strftime('%Y-%m-%d')
+#         )
+#
+#         # Vectorize the newly added document
+#         rag_system.vectorize_document(doc_id, content)
+#
+#         return f"Text file '{title}' by {author} ingested and vectorized successfully."
+#     except Exception as e:
+#         logging.error(f"Error ingesting text file: {str(e)}")
+#         return f"Error ingesting text file: {str(e)}"
+
+
+
+# Setup logging
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+# Constants
+EMBEDDING_MODEL = 'all-MiniLM-L6-v2'
+VECTOR_DIM = 384  # Dimension of the chosen embedding model
+
+
+class RAGSystem:
+    def __init__(self, sqlite_path: str, pg_config: Dict[str, str], cache_size: int = 100):
+        self.sqlite_path = sqlite_path
+        self.pg_config = pg_config
+        self.model = SentenceTransformer(EMBEDDING_MODEL)
+        self.cache_size = cache_size
+
+        self._init_postgres()
+
+    def _init_postgres(self):
+        with psycopg2.connect(**self.pg_config) as conn:
+            with conn.cursor() as cur:
+                cur.execute("""
+                CREATE TABLE IF NOT EXISTS document_vectors (
+                    id SERIAL PRIMARY KEY,
+                    document_id INTEGER UNIQUE,
+                    vector vector(384)
+                )
+                """)
+            conn.commit()
+
+    @lru_cache(maxsize=100)
+    def _get_embedding(self, text: str) -> np.ndarray:
+        return self.model.encode([text])[0]
+
+    def vectorize_document(self, doc_id: int, content: str):
+        chunks = create_chunks(content, chunk_size=1000, overlap=100)
+        for chunk in chunks:
+            vector = self._get_embedding(chunk['text'])
+
+            with psycopg2.connect(**self.pg_config) as conn:
+                with conn.cursor() as cur:
+                    cur.execute("""
+                    INSERT INTO document_vectors (document_id, chunk_index, vector, metadata)
+                    VALUES (%s, %s, %s, %s)
+                    ON CONFLICT (document_id, chunk_index) DO UPDATE SET vector = EXCLUDED.vector
+                    """, (doc_id, chunk['index'], vector.tolist(), json.dumps(chunk)))
+                conn.commit()
+
+    def vectorize_all_documents(self):
+        with sqlite3.connect(self.sqlite_path) as sqlite_conn:
+            unprocessed_chunks = get_unvectorized_chunks(sqlite_conn, limit=1000)
+            for chunk in unprocessed_chunks:
+                self.vectorize_document(chunk['id'], chunk['text'])
+            mark_chunks_as_processed(sqlite_conn, [chunk['id'] for chunk in unprocessed_chunks])
+
+    def semantic_search(self, query: str, top_k: int = 5) -> List[Tuple[int, int, float]]:
+        query_vector = self._get_embedding(query)
+
+        with psycopg2.connect(**self.pg_config) as conn:
+            with conn.cursor() as cur:
+                cur.execute("""
+                SELECT document_id, chunk_index, 1 - (vector <-> %s) AS similarity
+                FROM document_vectors
+                ORDER BY vector <-> %s ASC
+                LIMIT %s
+                """, (query_vector.tolist(), query_vector.tolist(), top_k))
+                results = cur.fetchall()
+
+        return results
+
+    def get_document_content(self, doc_id: int) -> str:
+        with sqlite3.connect(self.sqlite_path) as conn:
+            cur = conn.cursor()
+            cur.execute("SELECT content FROM media WHERE id = ?", (doc_id,))
+            result = cur.fetchone()
+            return result[0] if result else ""
+
+    def bm25_search(self, query: str, top_k: int = 5) -> List[Tuple[int, float]]:
+        with sqlite3.connect(self.sqlite_path) as conn:
+            cur = conn.cursor()
+            cur.execute("SELECT id, content FROM media")
+            documents = cur.fetchall()
+
+        vectorizer = TfidfVectorizer(use_idf=True)
+        tfidf_matrix = vectorizer.fit_transform([doc[1] for doc in documents])
+
+        query_vector = vectorizer.transform([query])
+        doc_lengths = tfidf_matrix.sum(axis=1).A1
+        avg_doc_length = np.mean(doc_lengths)
+
+        k1, b = 1.5, 0.75
+        scores = []
+        for i, doc_vector in enumerate(tfidf_matrix):
+            score = np.sum(
+                ((k1 + 1) * query_vector.multiply(doc_vector)).A1 /
+                (k1 * (1 - b + b * doc_lengths[i] / avg_doc_length) + query_vector.multiply(doc_vector).A1)
+            )
+            scores.append((documents[i][0], score))
+
+        return sorted(scores, key=lambda x: x[1], reverse=True)[:top_k]
+
+    def combine_search_results(self, bm25_results: List[Tuple[int, float]], vector_results: List[Tuple[int, float]],
+                               alpha: float = 0.5) -> List[Tuple[int, float]]:
+        combined_scores = {}
+        for idx, score in bm25_results + vector_results:
+            if idx in combined_scores:
+                combined_scores[idx] += score * (alpha if idx in dict(bm25_results) else (1 - alpha))
+            else:
+                combined_scores[idx] = score * (alpha if idx in dict(bm25_results) else (1 - alpha))
+        return sorted(combined_scores.items(), key=lambda x: x[1], reverse=True)
+
+    def expand_query(self, query: str) -> str:
+        model = T5ForConditionalGeneration.from_pretrained("t5-small")
+        tokenizer = T5Tokenizer.from_pretrained("t5-small")
+
+        input_text = f"expand query: {query}"
+        input_ids = tokenizer.encode(input_text, return_tensors="pt")
+
+        outputs = model.generate(input_ids, max_length=50, num_return_sequences=1)
+        expanded_query = tokenizer.decode(outputs[0], skip_special_tokens=True)
+
+        return f"{query} {expanded_query}"
+
+    def cross_encoder_rerank(self, query: str, initial_results: List[Tuple[int, float]], top_k: int = 5) -> List[
+        Tuple[int, float]]:
+        from sentence_transformers import CrossEncoder
+        model = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
+
+        candidate_docs = [self.get_document_content(doc_id) for doc_id, _ in initial_results[:top_k * 2]]
+        pairs = [[query, doc] for doc in candidate_docs]
+        scores = model.predict(pairs)
+
+        reranked = sorted(zip(initial_results[:top_k * 2], scores), key=lambda x: x[1], reverse=True)
+        return [(idx, score) for (idx, _), score in reranked[:top_k]]
+
+    def rag_query(self, query: str, search_type: str = 'combined', top_k: int = 5, use_hyde: bool = False,
+                  rerank: bool = False, expand: bool = False) -> List[Dict[str, any]]:
+        try:
+            if expand:
+                query = self.expand_query(query)
+
+            if use_hyde:
+                # Implement HyDE if needed
+                pass
+            elif search_type == 'vector':
+                results = self.semantic_search(query, top_k)
+            elif search_type == 'bm25':
+                results = self.bm25_search(query, top_k)
+            elif search_type == 'combined':
+                bm25_results = self.bm25_search(query, top_k)
+                vector_results = self.semantic_search(query, top_k)
+                results = self.combine_search_results(bm25_results, vector_results)
+            else:
+                raise ValueError("Invalid search type. Choose 'vector', 'bm25', or 'combined'.")
+
+            if rerank:
+                results = self.cross_encoder_rerank(query, results, top_k)
+
+            enriched_results = []
+            for doc_id, score in results:
+                content = self.get_document_content(doc_id)
+                enriched_results.append({
+                    "document_id": doc_id,
+                    "score": score,
+                    "content": content[:500]  # Truncate content for brevity
+                })
+
+            return enriched_results
+        except Exception as e:
+            logger.error(f"An error occurred during RAG query: {str(e)}")
+            return []
+
+
+# Example usage
+if __name__ == "__main__":
+    sqlite_path = "path/to/your/sqlite/database.db"
+    pg_config = {
+        "dbname": "your_db_name",
+        "user": "your_username",
+        "password": "your_password",
+        "host": "localhost"
+    }
+
+    rag_system = RAGSystem(sqlite_path, pg_config)
+
+    # Vectorize all documents (run this once or periodically)
+    rag_system.vectorize_all_documents()
+
+    # Example query
+    query = "programming concepts for beginners"
+    results = rag_system.rag_query(query, search_type='combined', expand=True, rerank=True)
+
+    print(f"Search results for query: '{query}'\n")
+    for i, result in enumerate(results, 1):
+        print(f"Result {i}:")
+        print(f"Document ID: {result['document_id']}")
+        print(f"Score: {result['score']:.4f}")
+        print(f"Content snippet: {result['content']}")
+        print("---")

App_Function_Libraries/RAG/RAPTOR-Skeleton.py

@@ -0,0 +1,361 @@
+# Requirements
+# scikit-learn umap-learn
+from itertools import chain
+from typing import List, Dict
+
+from App_Function_Libraries.RAG.ChromaDB_Library import store_in_chroma, create_embedding, vector_search, chroma_client
+from App_Function_Libraries.Chunk_Lib import improved_chunking_process, recursive_summarize_chunks
+import logging
+from sklearn.mixture import GaussianMixture
+import umap
+from nltk.corpus import wordnet
+
+
+# Logging setup
+logging.basicConfig(filename='raptor.log', level=logging.DEBUG)
+
+# FIXME
+MAX_LEVELS = 3
+
+
+def log_and_summarize(text, prompt):
+    logging.debug(f"Summarizing text: {text[:100]} with prompt: {prompt}")
+    return dummy_summarize(text, prompt)
+
+# 1. Data Preparation
+def prepare_data(content: str, media_id: int, chunk_options: dict):
+    chunks = improved_chunking_process(content, chunk_options)
+    embeddings = [create_embedding(chunk['text']) for chunk in chunks]
+    return chunks, embeddings
+
+# 2. Recursive Summarization
+def recursive_summarization(chunks, summarize_func, custom_prompt):
+    summarized_chunks = recursive_summarize_chunks(
+        [chunk['text'] for chunk in chunks],
+        summarize_func=summarize_func,
+        custom_prompt=custom_prompt
+    )
+    return summarized_chunks
+
+# Initial gen
+# 3. Tree Organization
+#def build_tree_structure(chunks, embeddings, collection_name, level=0):
+#    if len(chunks) <= 1:
+#        return chunks  # Base case: if chunks are small enough, return as is
+
+    # Recursive case: cluster and summarize
+#    summarized_chunks = recursive_summarization(chunks, summarize_func=dummy_summarize, custom_prompt="Summarize:")
+#    new_chunks, new_embeddings = prepare_data(' '.join(summarized_chunks), media_id, chunk_options)
+
+    # Store in ChromaDB
+#    ids = [f"{media_id}_L{level}_chunk_{i}" for i in range(len(new_chunks))]
+#    store_in_chroma(collection_name, [chunk['text'] for chunk in new_chunks], new_embeddings, ids)
+
+    # Recursively build tree
+#    return build_tree_structure(new_chunks, new_embeddings, collection_name, level+1)
+
+# Second iteration
+def build_tree_structure(chunks, collection_name, level=0):
+    # Dynamic clustering
+    clustered_texts = dynamic_clustering([chunk['text'] for chunk in chunks])
+
+    # Summarize each cluster
+    summarized_clusters = {}
+    for cluster_id, cluster_texts in clustered_texts.items():
+        summary = dummy_summarize(' '.join(cluster_texts), custom_prompt="Summarize:")
+        summarized_clusters[cluster_id] = summary
+
+    # Store summaries at current level
+    ids = []
+    embeddings = []
+    summaries = []
+    for cluster_id, summary in summarized_clusters.items():
+        ids.append(f"{collection_name}_L{level}_C{cluster_id}")
+        embeddings.append(create_embedding(summary))
+        summaries.append(summary)
+
+    store_in_chroma(collection_name, summaries, embeddings, ids)
+
+    # Recursively build tree structure if necessary
+    if level < MAX_LEVELS:
+        for cluster_id, cluster_texts in clustered_texts.items():
+            build_tree_structure(cluster_texts, collection_name, level + 1)
+
+
+
+
+# Dummy summarize function (replace with actual summarization)
+def dummy_summarize(text, custom_prompt, temp=None, system_prompt=None):
+    return text  # Replace this with actual call to summarization model (like GPT-3.5-turbo)
+
+# 4. Retrieval
+def raptor_retrieve(query, collection_name, level=0):
+    results = vector_search(collection_name, query, k=5)
+    return results
+
+# Main function integrating RAPTOR
+def raptor_pipeline(media_id, content, chunk_options):
+    collection_name = f"media_{media_id}_raptor"
+    
+    # Step 1: Prepare Data
+    chunks, embeddings = prepare_data(content, media_id, chunk_options)
+    
+    # Step 2: Build Tree
+    build_tree_structure(chunks, embeddings, collection_name)
+    
+    # Step 3: Retrieve Information
+    query = "Your query here"
+    result = raptor_retrieve(query, collection_name)
+    print(result)
+    
+# Example usage
+content = "Your long document content here"
+chunk_options = {
+    'method': 'sentences',
+    'max_size': 300,
+    'overlap': 50
+}
+media_id = 1
+raptor_pipeline(media_id, content, chunk_options)
+
+
+#
+#
+###################################################################################################################
+#
+# Additions:
+
+
+def dynamic_clustering(texts, n_components=2):
+    # Step 1: Convert text to embeddings
+    embeddings = [create_embedding(text) for text in texts]
+    
+    # Step 2: Dimensionality reduction (UMAP)
+    reducer = umap.UMAP(n_components=n_components)
+    reduced_embeddings = reducer.fit_transform(embeddings)
+
+    # Step 3: Find optimal number of clusters using BIC
+    best_gmm = None
+    best_bic = float('inf')
+    n_clusters = range(2, 10)
+    for n in n_clusters:
+        gmm = GaussianMixture(n_components=n, covariance_type='full')
+        gmm.fit(reduced_embeddings)
+        bic = gmm.bic(reduced_embeddings)
+        if bic < best_bic:
+            best_bic = bic
+            best_gmm = gmm
+
+    # Step 4: Cluster the reduced embeddings
+    cluster_labels = best_gmm.predict(reduced_embeddings)
+    clustered_texts = {i: [] for i in range(best_gmm.n_components)}
+    for label, text in zip(cluster_labels, texts):
+        clustered_texts[label].append(text)
+
+    return clustered_texts
+
+
+def tree_traversal_retrieve(query, collection_name, max_depth=3):
+    logging.info(f"Starting tree traversal for query: {query}")
+    results = []
+    current_level = 0
+    current_nodes = [collection_name + '_L0']
+
+    while current_level <= max_depth and current_nodes:
+        next_level_nodes = []
+        for node_id in current_nodes:
+            documents = vector_search(node_id, query, k=5)
+            results.extend(documents)
+            next_level_nodes.extend([doc['id'] for doc in documents])  # Assuming your doc structure includes an 'id' field
+        current_nodes = next_level_nodes
+        current_level += 1
+
+    logging.info(f"Tree traversal completed with {len(results)} results")
+    return results
+
+
+def collapsed_tree_retrieve(query, collection_name):
+    all_layers = [f"{collection_name}_L{level}" for level in range(MAX_LEVELS)]
+    all_results = []
+
+    for layer in all_layers:
+        all_results.extend(vector_search(layer, query, k=5))
+
+    # Sort and rank results by relevance
+    sorted_results = sorted(all_results, key=lambda x: x['relevance'], reverse=True)  # Assuming 'relevance' is a key
+    return sorted_results[:5]  # Return top 5 results
+
+# Test collaped tree retrieval
+query = "Your broad query here"
+results = collapsed_tree_retrieve(query, collection_name=f"media_{media_id}_raptor")
+print(results)
+
+
+# Parallel processing
+# pip install joblib
+from joblib import Parallel, delayed
+
+def parallel_process_chunks(chunks):
+    return Parallel(n_jobs=-1)(delayed(create_embedding)(chunk['text']) for chunk in chunks)
+
+def build_tree_structure(chunks, collection_name, level=0):
+    clustered_texts = dynamic_clustering([chunk['text'] for chunk in chunks])
+
+    summarized_clusters = {}
+    for cluster_id, cluster_texts in clustered_texts.items():
+        summary = dummy_summarize(' '.join(cluster_texts), custom_prompt="Summarize:")
+        summarized_clusters[cluster_id] = summary
+
+    # Parallel processing of embeddings
+    embeddings = parallel_process_chunks([{'text': summary} for summary in summarized_clusters.values()])
+
+    ids = [f"{collection_name}_L{level}_C{cluster_id}" for cluster_id in summarized_clusters.keys()]
+    store_in_chroma(collection_name, list(summarized_clusters.values()), embeddings, ids)
+
+    if len(summarized_clusters) > 1 and level < MAX_LEVELS:
+        build_tree_structure(summarized_clusters.values(), collection_name, level + 1)
+
+# Asynchronous processing
+import asyncio
+
+async def async_create_embedding(text):
+    return create_embedding(text)  # Assuming create_embedding is now async
+
+async def build_tree_structure_async(chunks, collection_name, level=0):
+    clustered_texts = dynamic_clustering([chunk['text'] for chunk in chunks])
+
+    summarized_clusters = {}
+    for cluster_id, cluster_texts in clustered_texts.items():
+        summary = await async_create_embedding(' '.join(cluster_texts))
+        summarized_clusters[cluster_id] = summary
+
+    embeddings = await asyncio.gather(*[async_create_embedding(summary) for summary in summarized_clusters.values()])
+
+    ids = [f"{collection_name}_L{level}_C{cluster_id}" for cluster_id in summarized_clusters.keys()]
+    store_in_chroma(collection_name, list(summarized_clusters.values()), embeddings, ids)
+
+    if len(summarized_clusters) > 1 and level < MAX_LEVELS:
+        await build_tree_structure_async(summarized_clusters.values(), collection_name, level + 1)
+
+
+# User feedback Loop
+def get_user_feedback(results):
+    print("Please review the following results:")
+    for i, result in enumerate(results):
+        print(f"{i + 1}: {result['text'][:100]}...")
+
+    feedback = input("Enter the numbers of the results that were relevant (comma-separated): ")
+    relevant_indices = [int(i.strip()) - 1 for i in feedback.split(",")]
+    return relevant_indices
+
+
+def raptor_pipeline_with_feedback(media_id, content, chunk_options):
+    # ... Existing pipeline steps ...
+
+    query = "Your query here"
+    initial_results = tree_traversal_retrieve(query, collection_name=f"media_{media_id}_raptor")
+    relevant_indices = get_user_feedback(initial_results)
+
+    if relevant_indices:
+        relevant_results = [initial_results[i] for i in relevant_indices]
+        refined_query = " ".join([res['text'] for res in relevant_results])
+        try:
+            final_results = tree_traversal_retrieve(refined_query, collection_name=f"media_{media_id}_raptor")
+        except Exception as e:
+            logging.error(f"Error during retrieval: {str(e)}")
+            raise
+        print("Refined Results:", final_results)
+    else:
+        print("No relevant results were found in the initial search.")
+
+
+def identify_uncertain_results(results):
+    threshold = 0.5  # Define a confidence threshold
+    uncertain_results = [res for res in results if res['confidence'] < threshold]
+    return uncertain_results
+
+
+def raptor_pipeline_with_active_learning(media_id, content, chunk_options):
+    # ... Existing pipeline steps ...
+
+    query = "Your query here"
+    initial_results = tree_traversal_retrieve(query, collection_name=f"media_{media_id}_raptor")
+    uncertain_results = identify_uncertain_results(initial_results)
+
+    if uncertain_results:
+        print("The following results are uncertain. Please provide feedback:")
+        feedback_indices = get_user_feedback(uncertain_results)
+        # Use feedback to adjust retrieval or refine the query
+        refined_query = " ".join([uncertain_results[i]['text'] for i in feedback_indices])
+        final_results = tree_traversal_retrieve(refined_query, collection_name=f"media_{media_id}_raptor")
+        print("Refined Results:", final_results)
+    else:
+        print("No uncertain results were found.")
+
+
+# Query Expansion
+def expand_query_with_synonyms(query):
+    words = query.split()
+    expanded_query = []
+    for word in words:
+        synonyms = wordnet.synsets(word)
+        lemmas = set(chain.from_iterable([syn.lemma_names() for syn in synonyms]))
+        expanded_query.append(" ".join(lemmas))
+    return " ".join(expanded_query)
+
+
+def contextual_query_expansion(query, context):
+    # FIXME: Replace with actual contextual model
+    expanded_terms = some_contextual_model.get_expansions(query, context)
+    return query + " " + " ".join(expanded_terms)
+
+
+def raptor_pipeline_with_query_expansion(media_id, content, chunk_options):
+    # ... Existing pipeline steps ...
+
+    query = "Your initial query"
+    expanded_query = expand_query_with_synonyms(query)
+    initial_results = tree_traversal_retrieve(expanded_query, collection_name=f"media_{media_id}_raptor")
+    # ... Continue with feedback loop ...
+
+
+def generate_summary_with_citations(query: str, collection_name: str):
+    results = vector_search_with_citation(collection_name, query)
+    # FIXME
+    summary = summarize([res['text'] for res in results])
+    # Deduplicate sources
+    sources = list(set(res['source'] for res in results))
+    return f"{summary}\n\nCitations:\n" + "\n".join(sources)
+
+
+def vector_search_with_citation(collection_name: str, query: str, k: int = 10) -> List[Dict[str, str]]:
+    query_embedding = create_embedding(query)
+    collection = chroma_client.get_collection(name=collection_name)
+    results = collection.query(
+        query_embeddings=[query_embedding],
+        n_results=k
+    )
+    return [{'text': doc, 'source': meta['source']} for doc, meta in zip(results['documents'], results['metadatas'])]
+
+
+def generate_summary_with_footnotes(query: str, collection_name: str):
+    results = vector_search_with_citation(collection_name, query)
+    summary_parts = []
+    citations = []
+    for i, res in enumerate(results):
+        summary_parts.append(f"{res['text']} [{i + 1}]")
+        citations.append(f"[{i + 1}] {res['source']}")
+    return " ".join(summary_parts) + "\n\nFootnotes:\n" + "\n".join(citations)
+
+
+def generate_summary_with_hyperlinks(query: str, collection_name: str):
+    results = vector_search_with_citation(collection_name, query)
+    summary_parts = []
+    for res in results:
+        summary_parts.append(f'<a href="{res["source"]}">{res["text"][:100]}...</a>')
+    return " ".join(summary_parts)
+
+
+#
+# End of Additions
+############################################3############################################3##############################