Implement RAPTOR

[rahulkrprajapati]

I have the following initialisation for my class:L
constructor code:

                self.redis_client = redis.from_url(
                    self.config.get("REDIS_VECTOR_STORE")
                )
                self.docstore = RedisDocumentStore.from_redis_client(
                    redis_client=self.redis_client, namespace=self.namespace
                )
                # TODO: UNcomment to use redis vector store
                self.vector_store = RedisVectorStore(
                    index_name=self.namespace,
                    redis_url=self.config.get("REDIS_VECTOR_STORE"),
                    metadata_fields=["namespace"],
                )

                self.storage_context = StorageContext.from_defaults(
                    docstore=self.docstore,
                    index_store=RedisIndexStore.from_redis_client(
                        redis_client=self.redis_client, namespace=self.namespace
                    ),
                    vector_store=self.vector_store,
                )

                try:
                    self.base_index = load_index_from_storage(self.storage_context)
                    print("[INFO] Index found at storage")
                except ValueError as e:
                    print("[INFO] No index found at storage")
                    self.base_index = VectorStoreIndex(
                        nodes=[],
                        storage_context=self.storage_context,
                        store_nodes_override=True,
                    )

                self.base_retriever = self.base_index.as_retriever(
                    similarity_top_k=self.similarity_top_k,
                    filters=MetadataFilters(
                        filters=[
                            ExactMatchFilter(key="namespace", value=self.namespace)
                        ]
                    ),
                )
                self.retriever = AutoMergingRetriever(
                    self.base_retriever, self.storage_context, verbose=verbose
                )
                self.recency_postprocessor = FixedRecencyPostprocessor(
                    top_k=recency, date_key="creation_date"
                )
                self.postprocessor = SentenceTransformerRerank(
                    model=self.config.get("RAG_RERANK_MODEL"),
                    top_n=similarity_top_k_reranker,
                )
                self.llm = OpenAI(
                    model=self.config.get("LLM_ANSWERING_MODEL"),
                    api_key=self.config.get("OPENAI_API_KEY"),
                )
                self.response_synthesizer = get_response_synthesizer(
                    response_mode=ResponseMode.COMPACT, llm=self.llm
                )

I also have the following helper functions:

    def create_document(self, text, filename):
        documents = []
        for idx, page in text.items():
            document = Document(text=page)
            current_date = datetime.now().strftime("%Y-%m-%d")
            document.metadata = {
                "filename": filename,
                "page_number": idx,
                "creation_date": current_date,
                "last_accessed_date": current_date,
                "last_modified_date": current_date,
                "namespace": self.namespace,
            }
            documents.append(document)
        return documents

    def return_nodes(self, docs):
        nodes = self.node_parser.get_nodes_from_documents(docs)
        return nodes

    def add_nodes_to_doc_store(self, all_nodes):
        try:
            self.docstore.add_documents(nodes=all_nodes)
        except Exception as e:
            print(f"An error occurred when adding documents to doc store: {e}")
            traceback.print_exc()
            raise Exception(
                f"An error occurred when adding documents to doc store: {e}"
            )

    def process_streamlit_fetch_query_results(self, query="", verbose=False):
        try:
            bm24_retriever = BM25Retriever.from_defaults(
                docstore=self.docstore, similarity_top_k=self.similarity_top_k
            )

            fusion_retriever = QueryFusionRetriever(
                [self.retriever, bm24_retriever],
                similarity_top_k=self.similarity_top_k,
                num_queries=1,  # set this to 1 to disable query generation
                mode="reciprocal_rerank",
                use_async=True,
                verbose=verbose,
            )
            query_bundle = QueryBundle(query_str=query)
            retrived_nodes = fusion_retriever.retrieve(query_bundle)
            recency_nodes = self.recency_postprocessor.postprocess_nodes(
                retrived_nodes, query_bundle=query_bundle
            )
            rerank_nodes = self.postprocessor.postprocess_nodes(
                nodes=recency_nodes, query_bundle=query_bundle
            )
            px.active_session().url
            return rerank_nodes
        except Exception as e:
            traceback.print_exc()
            raise Exception(f"An error occurred retrieving: {e}")

How can I modify it to use raptor ?

@dosu

[rahulkrprajapati]

@dosu how do I implement this making sure that I'm able to add more documents in the future if needed.

I have the following initialisation for my class:L
constructor code:

                self.redis_client = redis.from_url(
                    self.config.get("REDIS_VECTOR_STORE")
                )
                self.docstore = RedisDocumentStore.from_redis_client(
                    redis_client=self.redis_client, namespace=self.namespace
                )
                # TODO: UNcomment to use redis vector store
                self.vector_store = RedisVectorStore(
                    index_name=self.namespace,
                    redis_url=self.config.get("REDIS_VECTOR_STORE"),
                    metadata_fields=["namespace"],
                )

                self.storage_context = StorageContext.from_defaults(
                    docstore=self.docstore,
                    index_store=RedisIndexStore.from_redis_client(
                        redis_client=self.redis_client, namespace=self.namespace
                    ),
                    vector_store=self.vector_store,
                )

                try:
                    self.base_index = load_index_from_storage(self.storage_context)
                    print("[INFO] Index found at storage")
                except ValueError as e:
                    print("[INFO] No index found at storage")
                    self.base_index = VectorStoreIndex(
                        nodes=[],
                        storage_context=self.storage_context,
                        store_nodes_override=True,
                    )

                self.base_retriever = self.base_index.as_retriever(
                    similarity_top_k=self.similarity_top_k,
                    filters=MetadataFilters(
                        filters=[
                            ExactMatchFilter(key="namespace", value=self.namespace)
                        ]
                    ),
                )
                self.retriever = AutoMergingRetriever(
                    self.base_retriever, self.storage_context, verbose=verbose
                )
                self.recency_postprocessor = FixedRecencyPostprocessor(
                    top_k=recency, date_key="creation_date"
                )
                self.postprocessor = SentenceTransformerRerank(
                    model=self.config.get("RAG_RERANK_MODEL"),
                    top_n=similarity_top_k_reranker,
                )
                self.llm = OpenAI(
                    model=self.config.get("LLM_ANSWERING_MODEL"),
                    api_key=self.config.get("OPENAI_API_KEY"),
                )
                self.response_synthesizer = get_response_synthesizer(
                    response_mode=ResponseMode.COMPACT, llm=self.llm
                )

I also have the following helper functions:

    def create_document(self, text, filename):
        documents = []
        for idx, page in text.items():
            document = Document(text=page)
            current_date = datetime.now().strftime("%Y-%m-%d")
            document.metadata = {
                "filename": filename,
                "page_number": idx,
                "creation_date": current_date,
                "last_accessed_date": current_date,
                "last_modified_date": current_date,
                "namespace": self.namespace,
            }
            documents.append(document)
        return documents

    def return_nodes(self, docs):
        nodes = self.node_parser.get_nodes_from_documents(docs)
        return nodes

    def add_nodes_to_doc_store(self, all_nodes):
        try:
            self.docstore.add_documents(nodes=all_nodes)
        except Exception as e:
            print(f"An error occurred when adding documents to doc store: {e}")
            traceback.print_exc()
            raise Exception(
                f"An error occurred when adding documents to doc store: {e}"
            )

    def process_streamlit_fetch_query_results(self, query="", verbose=False):
        try:
            bm24_retriever = BM25Retriever.from_defaults(
                docstore=self.docstore, similarity_top_k=self.similarity_top_k
            )

            fusion_retriever = QueryFusionRetriever(
                [self.retriever, bm24_retriever],
                similarity_top_k=self.similarity_top_k,
                num_queries=1,  # set this to 1 to disable query generation
                mode="reciprocal_rerank",
                use_async=True,
                verbose=verbose,
            )
            query_bundle = QueryBundle(query_str=query)
            retrived_nodes = fusion_retriever.retrieve(query_bundle)
            recency_nodes = self.recency_postprocessor.postprocess_nodes(
                retrived_nodes, query_bundle=query_bundle
            )
            rerank_nodes = self.postprocessor.postprocess_nodes(
                nodes=recency_nodes, query_bundle=query_bundle
            )
            px.active_session().url
            return rerank_nodes
        except Exception as e:
            traceback.print_exc()
            raise Exception(f"An error occurred retrieving: {e}")

How can I modify it to use raptor ?

[dosubot[bot]]

To modify your class initialization and helper functions to use RAPTOR while ensuring that you can add more documents in the future, you can follow these steps:

Class Initialization

1. Import the necessary classes from the raptor package:

   from llama_index.packs.raptor import RaptorPack, RaptorRetriever
   from llama_index.packs.raptor.base import SummaryModule
   from llama_index.core.schema import BaseNode, TextNode

2. Modify the constructor to initialize RaptorPack and RaptorRetriever:

   class MyClass:
       def __init__(self, config, namespace, similarity_top_k, recency, similarity_top_k_reranker, verbose):
           self.config = config
           self.namespace = namespace
           self.similarity_top_k = similarity_top_k
   
           self.redis_client = redis.from_url(self.config.get("REDIS_VECTOR_STORE"))
           self.docstore = RedisDocumentStore.from_redis_client(redis_client=self.redis_client, namespace=self.namespace)
           self.vector_store = RedisVectorStore(index_name=self.namespace, redis_url=self.config.get("REDIS_VECTOR_STORE"), metadata_fields=["namespace"])
   
           self.storage_context = StorageContext.from_defaults(
               docstore=self.docstore,
               index_store=RedisIndexStore.from_redis_client(redis_client=self.redis_client, namespace=self.namespace),
               vector_store=self.vector_store,
           )
   
           llm = OpenAI(model=self.config.get("LLM_ANSWERING_MODEL"), api_key=self.config.get("OPENAI_API_KEY"))
           embed_model = "your_embedding_model"  # Replace with your embedding model
           summary_prompt = "As a professional summarizer, create a concise and comprehensive summary of the provided text, be it an article, post, conversation, or passage with as much detail as possible."
           summary_module = SummaryModule(llm=llm, summary_prompt=summary_prompt, num_workers=16)
   
           self.raptor_pack = RaptorPack([], llm=llm, embed_model=embed_model, summary_module=summary_module, vector_store=self.vector_store)
           self.raptor_retriever = self.raptor_pack.retriever
   
           self.recency_postprocessor = FixedRecencyPostprocessor(top_k=recency, date_key="creation_date")
           self.postprocessor = SentenceTransformerRerank(model=self.config.get("RAG_RERANK_MODEL"), top_n=similarity_top_k_reranker)
           self.response_synthesizer = get_response_synthesizer(response_mode=ResponseMode.COMPACT, llm=llm)

Helper Functions

1. Modify create_document to return TextNode instances:

   def create_document(self, text, filename):
       documents = []
       for idx, page in text.items():
           document = TextNode(text=page)
           current_date = datetime.now().strftime("%Y-%m-%d")
           document.metadata = {
               "filename": filename,
               "page_number": idx,
               "creation_date": current_date,
               "last_accessed_date": current_date,
               "last_modified_date": current_date,
               "namespace": self.namespace,
           }
           documents.append(document)
       return documents

2. Modify add_nodes_to_doc_store to use RaptorRetriever's insert method:

   def add_nodes_to_doc_store(self, all_nodes):
       try:
           asyncio.run(self.raptor_retriever.insert(all_nodes))
       except Exception as e:
           print(f"An error occurred when adding documents to doc store: {e}")
           traceback.print_exc()
           raise Exception(f"An error occurred when adding documents to doc store: {e}")

3. Modify process_streamlit_fetch_query_results to use RaptorRetriever's retrieve method:

   def process_streamlit_fetch_query_results(self, query="", verbose=False):
       try:
           query_bundle = QueryBundle(query_str=query)
           retrived_nodes = asyncio.run(self.raptor_retriever.retrieve(query_bundle.query_str, mode="collapsed"))
           recency_nodes = self.recency_postprocessor.postprocess_nodes(retrived_nodes, query_bundle=query_bundle)
           rerank_nodes = self.postprocessor.postprocess_nodes(nodes=recency_nodes, query_bundle=query_bundle)
           px.active_session().url
           return rerank_nodes
       except Exception as e:
           traceback.print_exc()
           raise Exception(f"An error occurred retrieving: {e}")

Example Usage

config = {
    "REDIS_VECTOR_STORE": "redis://localhost:6379/0",
    "LLM_ANSWERING_MODEL": "gpt-3.5-turbo",
    "OPENAI_API_KEY": "your_openai_api_key",
    "RAG_RERANK_MODEL": "sentence-transformers/all-MiniLM-L6-v2"
}
namespace = "example_namespace"
similarity_top_k = 5
recency = 3
similarity_top_k_reranker = 10
verbose = True

my_class_instance = MyClass(config, namespace, similarity_top_k, recency, similarity_top_k_reranker, verbose)
documents = my_class_instance.create_document({"1": "Document 1 content", "2": "Document 2 content"}, "example_file")
nodes = my_class_instance.return_nodes(documents)
my_class_instance.add_nodes_to_doc_store(nodes)
retrieved_docs = my_class_instance.process_streamlit_fetch_query_results("example query")
print(retrieved_docs)

This setup ensures that you can easily add more documents in the future by using the add_nodes_to_doc_store method and retrieve documents using the process_streamlit_fetch_query_results method. Additionally, it leverages the SummaryModule for configuring summaries and supports persistence with a vector store ^[1][2][3].

To continue talking to Dosu, mention @dosu.

[rahulkrprajapati]

@dosu you forgot the HierarchicalNodeParser should it be used for adding the nodes? Could you also use QueryFusionRetriever to merger raptor + bm25 in the generated code

[dosubot[bot]]

To incorporate HierarchicalNodeParser for adding nodes and use QueryFusionRetriever to merge RAPTOR and BM25 in your generated code, follow these steps:

Class Initialization

1. Import the necessary classes:

   from llama_index.packs.raptor import RaptorPack, RaptorRetriever
   from llama_index.packs.raptor.base import SummaryModule
   from llama_index.core.schema import BaseNode, TextNode
   from llama_index.core.node_parser import HierarchicalNodeParser, Node
   from llama_index.core.retrievers import QueryFusionRetriever, BM25Retriever
   from llama_index.llms.openai import OpenAI
   from llama_index.embeddings.openai import OpenAIEmbedding

2. Modify the constructor to initialize RaptorPack, RaptorRetriever, and HierarchicalNodeParser:

   class MyClass:
       def __init__(self, config, namespace, similarity_top_k, recency, similarity_top_k_reranker, verbose):
           self.config = config
           self.namespace = namespace
           self.similarity_top_k = similarity_top_k
   
           self.redis_client = redis.from_url(self.config.get("REDIS_VECTOR_STORE"))
           self.docstore = RedisDocumentStore.from_redis_client(redis_client=self.redis_client, namespace=self.namespace)
           self.vector_store = RedisVectorStore(index_name=self.namespace, redis_url=self.config.get("REDIS_VECTOR_STORE"), metadata_fields=["namespace"])
   
           self.storage_context = StorageContext.from_defaults(
               docstore=self.docstore,
               index_store=RedisIndexStore.from_redis_client(redis_client=self.redis_client, namespace=self.namespace),
               vector_store=self.vector_store,
           )
   
           llm = OpenAI(model=self.config.get("LLM_ANSWERING_MODEL"), api_key=self.config.get("OPENAI_API_KEY"))
           embed_model = OpenAIEmbedding(model="text-embedding-3-small")
           summary_prompt = "As a professional summarizer, create a concise and comprehensive summary of the provided text, be it an article, post, conversation, or passage with as much detail as possible."
           summary_module = SummaryModule(llm=llm, summary_prompt=summary_prompt, num_workers=16)
   
           self.raptor_pack = RaptorPack([], llm=llm, embed_model=embed_model, summary_module=summary_module, vector_store=self.vector_store)
           self.raptor_retriever = self.raptor_pack.retriever
   
           self.node_parser = HierarchicalNodeParser.from_defaults(chunk_sizes=[2048, 512, 128])
   
           self.recency_postprocessor = FixedRecencyPostprocessor(top_k=recency, date_key="creation_date")
           self.postprocessor = SentenceTransformerRerank(model=self.config.get("RAG_RERANK_MODEL"), top_n=similarity_top_k_reranker)
           self.response_synthesizer = get_response_synthesizer(response_mode=ResponseMode.COMPACT, llm=llm)

Helper Functions

1. Modify create_document to return TextNode instances:

   def create_document(self, text, filename):
       documents = []
       for idx, page in text.items():
           document = TextNode(text=page)
           current_date = datetime.now().strftime("%Y-%m-%d")
           document.metadata = {
               "filename": filename,
               "page_number": idx,
               "creation_date": current_date,
               "last_accessed_date": current_date,
               "last_modified_date": current_date,
               "namespace": self.namespace,
           }
           documents.append(document)
       return documents

2. Modify return_nodes to use HierarchicalNodeParser:

   def return_nodes(self, docs):
       nodes = self.node_parser.get_nodes_from_documents(docs)
       return nodes

3. Modify add_nodes_to_doc_store to use RaptorRetriever's insert method:

   def add_nodes_to_doc_store(self, all_nodes):
       try:
           asyncio.run(self.raptor_retriever.insert(all_nodes))
       except Exception as e:
           print(f"An error occurred when adding documents to doc store: {e}")
           traceback.print_exc()
           raise Exception(f"An error occurred when adding documents to doc store: {e}")

4. Modify process_streamlit_fetch_query_results to use QueryFusionRetriever to merge RAPTOR and BM25:

   def process_streamlit_fetch_query_results(self, query="", verbose=False):
       try:
           bm25_retriever = BM25Retriever.from_defaults(docstore=self.docstore, similarity_top_k=self.similarity_top_k)
   
           fusion_retriever = QueryFusionRetriever(
               [self.raptor_retriever, bm25_retriever],
               similarity_top_k=self.similarity_top_k,
               num_queries=1,  # set this to 1 to disable query generation
               mode="reciprocal_rerank",
               use_async=True,
               verbose=verbose,
           )
           query_bundle = QueryBundle(query_str=query)
           retrieved_nodes = asyncio.run(fusion_retriever.aretrieve(query_bundle.query_str))
           recency_nodes = self.recency_postprocessor.postprocess_nodes(retrieved_nodes, query_bundle=query_bundle)
           rerank_nodes = self.postprocessor.postprocess_nodes(nodes=recency_nodes, query_bundle=query_bundle)
           px.active_session().url
           return rerank_nodes
       except Exception as e:
           traceback.print_exc()
           raise Exception(f"An error occurred retrieving: {e}")

Example Usage

config = {
    "REDIS_VECTOR_STORE": "redis://localhost:6379/0",
    "LLM_ANSWERING_MODEL": "gpt-3.5-turbo",
    "OPENAI_API_KEY": "your_openai_api_key",
    "RAG_RERANK_MODEL": "sentence-transformers/all-MiniLM-L6-v2"
}
namespace = "example_namespace"
similarity_top_k = 5
recency = 3
similarity_top_k_reranker = 10
verbose = True

my_class_instance = MyClass(config, namespace, similarity_top_k, recency, similarity_top_k_reranker, verbose)
documents = my_class_instance.create_document({"1": "Document 1 content", "2": "Document 2 content"}, "example_file")
nodes = my_class_instance.return_nodes(documents)
my_class_instance.add_nodes_to_doc_store(nodes)
retrieved_docs = my_class_instance.process_streamlit_fetch_query_results("example query")
print(retrieved_docs)

This setup ensures that you can easily add more documents in the future by using the add_nodes_to_doc_store method and retrieve documents using the process_streamlit_fetch_query_results method. Additionally, it leverages the SummaryModule for configuring summaries and supports persistence with a vector store ^{[1][2][3][4][5]}.

To continue talking to Dosu, mention @dosu.

[rahulkrprajapati]

@dosu in the code below define my metadata filter properly:

self.raptor_pack = RaptorPack(
                    documents=[],
                    embed_model=Settings.embed_model,
                    llm=self.llm,
                    vector_store=self.vector_store,
                    similarity_top_k=self.similarity_top_k,
                    mode="collapsed",
                    summary_module=self.summary_module,
                )

                self.bm25_retriever = BM25Retriever.from_defaults(
                    docstore=self.docstore, similarity_top_k=self.similarity_top_k
                )

                self.fusion_retriever = QueryFusionRetriever(
                    [self.raptor_pack.retriever, self.bm25_retriever],
                    similarity_top_k=self.similarity_top_k,
                    num_queries=1,  # set this to 1 to disable query generation
                    mode="reciprocal_rerank",
                    use_async=True,
                    verbose=verbose,
                )

                self.node_parser = HierarchicalNodeParser.from_defaults()

                try:
                    self.base_index = load_index_from_storage(self.storage_context)
                    print("[INFO] Index found at storage")
                except ValueError as e:
                    print("[INFO] No index found at storage")
                    self.base_index = VectorStoreIndex(
                        nodes=[],
                        storage_context=self.storage_context,
                        store_nodes_override=True,
                    )

                self.base_retriever = self.base_index.as_retriever(
                    similarity_top_k=self.similarity_top_k,
                    filters=MetadataFilters(
                        filters=[
                            ExactMatchFilter(key="namespace", value=self.namespace)
                        ]
                    ),
                )
                self.retriever = AutoMergingRetriever(
                    self.base_retriever, self.storage_context, verbose=verbose
                )
                self.recency_postprocessor = FixedRecencyPostprocessor(
                    top_k=recency, date_key="creation_date"
                )
                self.postprocessor = SentenceTransformerRerank(
                    model=self.config.get("RAG_RERANK_MODEL"),
                    top_n=similarity_top_k_reranker,
                )
                self.response_synthesizer = get_response_synthesizer(
                    response_mode=ResponseMode.COMPACT, llm=self.llm
                )

the code fragment here does not take into account my metadata filter:

self.fusion_retriever = QueryFusionRetriever(
                    [self.raptor_pack.retriever, self.bm25_retriever],
                    similarity_top_k=self.similarity_top_k,
                    num_queries=1,  # set this to 1 to disable query generation
                    mode="reciprocal_rerank",
                    use_async=True,
                    verbose=verbose,
                )

the filter should be something like:

filters=MetadataFilters(
                        filters=[
                            ExactMatchFilter(key="namespace", value=self.namespace)
                        ]
                    ),

this is being used in the function as:

    def process_streamlit_fetch_query_results(self, query="", verbose=False):
        try:
            query_bundle = QueryBundle(query_str=query)
            retrived_nodes = self.fusion_retriever.retrieve(query_bundle)
            recency_nodes = self.recency_postprocessor.postprocess_nodes(
                retrived_nodes, query_bundle=query_bundle
            )
            rerank_nodes = self.postprocessor.postprocess_nodes(
                nodes=recency_nodes, query_bundle=query_bundle
            )
            px.active_session().url
            return rerank_nodes
        except Exception as e:
            traceback.print_exc()
            raise Exception(f"An error occurred retrieving: {e}")

[dosubot[bot]]

To properly define and incorporate a metadata filter in your QueryFusionRetriever, you need to pass the filters parameter when initializing the retriever. Here’s how you can modify your code to include the metadata filter:

Updated Code

1. Define the Metadata Filter:

   filters = MetadataFilters(
       filters=[
           ExactMatchFilter(key="namespace", value=self.namespace)
       ]
   )

2. Pass the Metadata Filter to QueryFusionRetriever:

   self.fusion_retriever = QueryFusionRetriever(
       [self.raptor_pack.retriever, self.bm25_retriever],
       similarity_top_k=self.similarity_top_k,
       num_queries=1,  # set this to 1 to disable query generation
       mode="reciprocal_rerank",
       use_async=True,
       verbose=verbose,
       filters=filters  # Add the filters here
   )

Full Code with Metadata Filter

class MyClass:
    def __init__(self, config, namespace, similarity_top_k, recency, similarity_top_k_reranker, verbose):
        self.config = config
        self.namespace = namespace
        self.similarity_top_k = similarity_top_k

        self.redis_client = redis.from_url(self.config.get("REDIS_VECTOR_STORE"))
        self.docstore = RedisDocumentStore.from_redis_client(redis_client=self.redis_client, namespace=self.namespace)
        self.vector_store = RedisVectorStore(index_name=self.namespace, redis_url=self.config.get("REDIS_VECTOR_STORE"), metadata_fields=["namespace"])

        self.storage_context = StorageContext.from_defaults(
            docstore=self.docstore,
            index_store=RedisIndexStore.from_redis_client(redis_client=self.redis_client, namespace=self.namespace),
            vector_store=self.vector_store,
        )

        llm = OpenAI(model=self.config.get("LLM_ANSWERING_MODEL"), api_key=self.config.get("OPENAI_API_KEY"))
        embed_model = OpenAIEmbedding(model="text-embedding-3-small")
        summary_prompt = "As a professional summarizer, create a concise and comprehensive summary of the provided text, be it an article, post, conversation, or passage with as much detail as possible."
        summary_module = SummaryModule(llm=llm, summary_prompt=summary_prompt, num_workers=16)

        self.raptor_pack = RaptorPack(
            documents=[],
            embed_model=embed_model,
            llm=llm,
            vector_store=self.vector_store,
            similarity_top_k=self.similarity_top_k,
            mode="collapsed",
            summary_module=summary_module,
        )

        self.bm25_retriever = BM25Retriever.from_defaults(
            docstore=self.docstore, similarity_top_k=self.similarity_top_k
        )

        filters = MetadataFilters(
            filters=[
                ExactMatchFilter(key="namespace", value=self.namespace)
            ]
        )

        self.fusion_retriever = QueryFusionRetriever(
            [self.raptor_pack.retriever, self.bm25_retriever],
            similarity_top_k=self.similarity_top_k,
            num_queries=1,  # set this to 1 to disable query generation
            mode="reciprocal_rerank",
            use_async=True,
            verbose=verbose,
            filters=filters  # Add the filters here
        )

        self.node_parser = HierarchicalNodeParser.from_defaults()

        try:
            self.base_index = load_index_from_storage(self.storage_context)
            print("[INFO] Index found at storage")
        except ValueError as e:
            print("[INFO] No index found at storage")
            self.base_index = VectorStoreIndex(
                nodes=[],
                storage_context=self.storage_context,
                store_nodes_override=True,
            )

        self.base_retriever = self.base_index.as_retriever(
            similarity_top_k=self.similarity_top_k,
            filters=MetadataFilters(
                filters=[
                    ExactMatchFilter(key="namespace", value=self.namespace)
                ]
            ),
        )
        self.retriever = AutoMergingRetriever(
            self.base_retriever, self.storage_context, verbose=verbose
        )
        self.recency_postprocessor = FixedRecencyPostprocessor(
            top_k=recency, date_key="creation_date"
        )
        self.postprocessor = SentenceTransformerRerank(
            model=self.config.get("RAG_RERANK_MODEL"),
            top_n=similarity_top_k_reranker,
        )
        self.response_synthesizer = get_response_synthesizer(
            response_mode=ResponseMode.COMPACT, llm=llm
        )

    def process_streamlit_fetch_query_results(self, query="", verbose=False):
        try:
            query_bundle = QueryBundle(query_str=query)
            retrieved_nodes = self.fusion_retriever.retrieve(query_bundle)
            recency_nodes = self.recency_postprocessor.postprocess_nodes(
                retrieved_nodes, query_bundle=query_bundle
            )
            rerank_nodes = self.postprocessor.postprocess_nodes(
                nodes=recency_nodes, query_bundle=query_bundle
            )
            px.active_session().url
            return rerank_nodes
        except Exception as e:
            traceback.print_exc()
            raise Exception(f"An error occurred retrieving: {e}")

This ensures that the QueryFusionRetriever uses the specified metadata filter when retrieving nodes ^[1].

To continue talking to Dosu, mention @dosu.

[rahulkrprajapati]

TypeError: QueryFusionRetriever.init() got an unexpected keyword argument 'filters

@dosu

[dosubot[bot]]

The QueryFusionRetriever class does not accept a filters keyword argument in its __init__ method. To apply metadata filters, you need to pass them through the individual retrievers or use a different approach.

Here’s how you can modify your code to apply metadata filters correctly:

1. Remove the filters argument from QueryFusionRetriever.
2. Apply the filters directly to the retrievers that support them.

Updated Code

1. Define the Metadata Filter:

   filters = MetadataFilters(
       filters=[
           ExactMatchFilter(key="namespace", value=self.namespace)
       ]
   )

2. Apply the Metadata Filter to the RaptorRetriever:

   self.raptor_pack = RaptorPack(
       documents=[],
       embed_model=Settings.embed_model,
       llm=self.llm,
       vector_store=self.vector_store,
       similarity_top_k=self.similarity_top_k,
       mode="collapsed",
       summary_module=self.summary_module,
   )
   
   self.raptor_retriever = self.raptor_pack.retriever
   self.raptor_retriever.filters = filters  # Apply filters here

3. Initialize BM25Retriever and QueryFusionRetriever without the filters argument:

   self.bm25_retriever = BM25Retriever.from_defaults(
       docstore=self.docstore, similarity_top_k=self.similarity_top_k
   )
   
   self.fusion_retriever = QueryFusionRetriever(
       [self.raptor_retriever, self.bm25_retriever],
       similarity_top_k=self.similarity_top_k,
       num_queries=1,  # set this to 1 to disable query generation
       mode="reciprocal_rerank",
       use_async=True,
       verbose=verbose
   )

Full Code with Metadata Filter

class MyClass:
    def __init__(self, config, namespace, similarity_top_k, recency, similarity_top_k_reranker, verbose):
        self.config = config
        self.namespace = namespace
        self.similarity_top_k = similarity_top_k

        self.redis_client = redis.from_url(self.config.get("REDIS_VECTOR_STORE"))
        self.docstore = RedisDocumentStore.from_redis_client(redis_client=self.redis_client, namespace=self.namespace)
        self.vector_store = RedisVectorStore(index_name=self.namespace, redis_url=self.config.get("REDIS_VECTOR_STORE"), metadata_fields=["namespace"])

        self.storage_context = StorageContext.from_defaults(
            docstore=self.docstore,
            index_store=RedisIndexStore.from_redis_client(redis_client=self.redis_client, namespace=self.namespace),
            vector_store=self.vector_store,
        )

        llm = OpenAI(model=self.config.get("LLM_ANSWERING_MODEL"), api_key=self.config.get("OPENAI_API_KEY"))
        embed_model = OpenAIEmbedding(model="text-embedding-3-small")
        summary_prompt = "As a professional summarizer, create a concise and comprehensive summary of the provided text, be it an article, post, conversation, or passage with as much detail as possible."
        summary_module = SummaryModule(llm=llm, summary_prompt=summary_prompt, num_workers=16)

        filters = MetadataFilters(
            filters=[
                ExactMatchFilter(key="namespace", value=self.namespace)
            ]
        )

        self.raptor_pack = RaptorPack(
            documents=[],
            embed_model=embed_model,
            llm=llm,
            vector_store=self.vector_store,
            similarity_top_k=self.similarity_top_k,
            mode="collapsed",
            summary_module=summary_module,
        )

        self.raptor_retriever = self.raptor_pack.retriever
        self.raptor_retriever.filters = filters  # Apply filters here

        self.bm25_retriever = BM25Retriever.from_defaults(
            docstore=self.docstore, similarity_top_k=self.similarity_top_k
        )

        self.fusion_retriever = QueryFusionRetriever(
            [self.raptor_retriever, self.bm25_retriever],
            similarity_top_k=self.similarity_top_k,
            num_queries=1,  # set this to 1 to disable query generation
            mode="reciprocal_rerank",
            use_async=True,
            verbose=verbose
        )

        self.node_parser = HierarchicalNodeParser.from_defaults()

        try:
            self.base_index = load_index_from_storage(self.storage_context)
            print("[INFO] Index found at storage")
        except ValueError as e:
            print("[INFO] No index found at storage")
            self.base_index = VectorStoreIndex(
                nodes=[],
                storage_context=self.storage_context,
                store_nodes_override=True,
            )

        self.base_retriever = self.base_index.as_retriever(
            similarity_top_k=self.similarity_top_k,
            filters=MetadataFilters(
                filters=[
                    ExactMatchFilter(key="namespace", value=self.namespace)
                ]
            ),
        )
        self.retriever = AutoMergingRetriever(
            self.base_retriever, self.storage_context, verbose=verbose
        )
        self.recency_postprocessor = FixedRecencyPostprocessor(
            top_k=recency, date_key="creation_date"
        )
        self.postprocessor = SentenceTransformerRerank(
            model=self.config.get("RAG_RERANK_MODEL"),
            top_n=similarity_top_k_reranker,
        )
        self.response_synthesizer = get_response_synthesizer(
            response_mode=ResponseMode.COMPACT, llm=llm
        )

    def process_streamlit_fetch_query_results(self, query="", verbose=False):
        try:
            query_bundle = QueryBundle(query_str=query)
            retrieved_nodes = self.fusion_retriever.retrieve(query_bundle)
            recency_nodes = self.recency_postprocessor.postprocess_nodes(
                retrieved_nodes, query_bundle=query_bundle
            )
            rerank_nodes = self.postprocessor.postprocess_nodes(
                nodes=recency_nodes, query_bundle=query_bundle
            )
            px.active_session().url
            return rerank_nodes
        except Exception as e:
            traceback.print_exc()
            raise Exception(f"An error occurred retrieving: {e}")

This ensures that the QueryFusionRetriever uses the specified metadata filter when retrieving nodes ^[1][2].

To continue talking to Dosu, mention @dosu.

[rahulkrprajapati]

@dosu

this does not work. Can we do it here in the function?

    def process_streamlit_fetch_query_results(self, query="", verbose=False):
        try:
            query_bundle = QueryBundle(query_str=query)
            retrived_nodes = self.fusion_retriever.retrieve(query_bundle)
            recency_nodes = self.recency_postprocessor.postprocess_nodes(
                retrived_nodes, query_bundle=query_bundle
            )
            rerank_nodes = self.postprocessor.postprocess_nodes(
                nodes=recency_nodes, query_bundle=query_bundle
            )
            px.active_session().url
            return rerank_nodes
        except Exception as e:
            traceback.print_exc()
            raise Exception(f"An error occurred retrieving: {e}")

[dosubot[bot]]

To apply the metadata filter directly within the process_streamlit_fetch_query_results function, you can modify the retrievers to include the filters before performing the query. Here’s how you can do it:

Updated process_streamlit_fetch_query_results Function

1. Define the Metadata Filter:

   filters = MetadataFilters(
       filters=[
           ExactMatchFilter(key="namespace", value=self.namespace)
       ]
   )

2. Apply the Metadata Filter to the Retrievers:

   def process_streamlit_fetch_query_results(self, query="", verbose=False):
       try:
           filters = MetadataFilters(
               filters=[
                   ExactMatchFilter(key="namespace", value=self.namespace)
               ]
           )
   
           # Apply filters to the Raptor retriever
           self.raptor_retriever.filters = filters
   
           # Initialize BM25 retriever with filters
           bm25_retriever = BM25Retriever.from_defaults(
               docstore=self.docstore, similarity_top_k=self.similarity_top_k
           )
           bm25_retriever.filters = filters
   
           # Initialize QueryFusionRetriever with the filtered retrievers
           fusion_retriever = QueryFusionRetriever(
               [self.raptor_retriever, bm25_retriever],
               similarity_top_k=self.similarity_top_k,
               num_queries=1,  # set this to 1 to disable query generation
               mode="reciprocal_rerank",
               use_async=True,
               verbose=verbose
           )
   
           query_bundle = QueryBundle(query_str=query)
           retrieved_nodes = fusion_retriever.retrieve(query_bundle)
           recency_nodes = self.recency_postprocessor.postprocess_nodes(
               retrieved_nodes, query_bundle=query_bundle
           )
           rerank_nodes = self.postprocessor.postprocess_nodes(
               nodes=recency_nodes, query_bundle=query_bundle
           )
           px.active_session().url
           return rerank_nodes
       except Exception as e:
           traceback.print_exc()
           raise Exception(f"An error occurred retrieving: {e}")

Full Code with Metadata Filter in Function

class MyClass:
    def __init__(self, config, namespace, similarity_top_k, recency, similarity_top_k_reranker, verbose):
        self.config = config
        self.namespace = namespace
        self.similarity_top_k = similarity_top_k

        self.redis_client = redis.from_url(self.config.get("REDIS_VECTOR_STORE"))
        self.docstore = RedisDocumentStore.from_redis_client(redis_client=self.redis_client, namespace=self.namespace)
        self.vector_store = RedisVectorStore(index_name=self.namespace, redis_url=self.config.get("REDIS_VECTOR_STORE"), metadata_fields=["namespace"])

        self.storage_context = StorageContext.from_defaults(
            docstore=self.docstore,
            index_store=RedisIndexStore.from_redis_client(redis_client=self.redis_client, namespace=self.namespace),
            vector_store=self.vector_store,
        )

        llm = OpenAI(model=self.config.get("LLM_ANSWERING_MODEL"), api_key=self.config.get("OPENAI_API_KEY"))
        embed_model = OpenAIEmbedding(model="text-embedding-3-small")
        summary_prompt = "As a professional summarizer, create a concise and comprehensive summary of the provided text, be it an article, post, conversation, or passage with as much detail as possible."
        summary_module = SummaryModule(llm=llm, summary_prompt=summary_prompt, num_workers=16)

        self.raptor_pack = RaptorPack(
            documents=[],
            embed_model=embed_model,
            llm=llm,
            vector_store=self.vector_store,
            similarity_top_k=self.similarity_top_k,
            mode="collapsed",
            summary_module=summary_module,
        )

        self.raptor_retriever = self.raptor_pack.retriever

        self.node_parser = HierarchicalNodeParser.from_defaults()

        try:
            self.base_index = load_index_from_storage(self.storage_context)
            print("[INFO] Index found at storage")
        except ValueError as e:
            print("[INFO] No index found at storage")
            self.base_index = VectorStoreIndex(
                nodes=[],
                storage_context=self.storage_context,
                store_nodes_override=True,
            )

        self.base_retriever = self.base_index.as_retriever(
            similarity_top_k=self.similarity_top_k,
            filters=MetadataFilters(
                filters=[
                    ExactMatchFilter(key="namespace", value=self.namespace)
                ]
            ),
        )
        self.retriever = AutoMergingRetriever(
            self.base_retriever, self.storage_context, verbose=verbose
        )
        self.recency_postprocessor = FixedRecencyPostprocessor(
            top_k=recency, date_key="creation_date"
        )
        self.postprocessor = SentenceTransformerRerank(
            model=self.config.get("RAG_RERANK_MODEL"),
            top_n=similarity_top_k_reranker,
        )
        self.response_synthesizer = get_response_synthesizer(
            response_mode=ResponseMode.COMPACT, llm=llm
        )

    def process_streamlit_fetch_query_results(self, query="", verbose=False):
        try:
            filters = MetadataFilters(
                filters=[
                    ExactMatchFilter(key="namespace", value=self.namespace)
                ]
            )

            # Apply filters to the Raptor retriever
            self.raptor_retriever.filters = filters

            # Initialize BM25 retriever with filters
            bm25_retriever = BM25Retriever.from_defaults(
                docstore=self.docstore, similarity_top_k=self.similarity_top_k
            )
            bm25_retriever.filters = filters

            # Initialize QueryFusionRetriever with the filtered retrievers
            fusion_retriever = QueryFusionRetriever(
                [self.raptor_retriever, bm25_retriever],
                similarity_top_k=self.similarity_top_k,
                num_queries=1,  # set this to 1 to disable query generation
                mode="reciprocal_rerank",
                use_async=True,
                verbose=verbose
            )

            query_bundle = QueryBundle(query_str=query)
            retrieved_nodes = fusion_retriever.retrieve(query_bundle)
            recency_nodes = self.recency_postprocessor.postprocess_nodes(
                retrieved_nodes, query_bundle=query_bundle
            )
            rerank_nodes = self.postprocessor.postprocess_nodes(
                nodes=recency_nodes, query_bundle=query_bundle
            )
            px.active_session().url
            return rerank_nodes
        except Exception as e:
            traceback.print_exc()
            raise Exception(f"An error occurred retrieving: {e}")

This approach ensures that the metadata filters are applied to the retrievers within the process_streamlit_fetch_query_results function, allowing you to filter the results based on the specified metadata criteria ^[1][2].

To continue talking to Dosu, mention @dosu.

[rahulkrprajapati]

@dosu what I meant is implement a metadata filter to filter based on this:

MetadataFilters(
                        filters=[
                            ExactMatchFilter(key="namespace", value=self.namespace)
                        ]
                    ),

    def process_streamlit_fetch_query_results(self, query="", verbose=False):
        try:
            query_bundle = QueryBundle(query_str=query)
            retrived_nodes = self.fusion_retriever.retrieve(query_bundle)
            recency_nodes = self.recency_postprocessor.postprocess_nodes(
                retrived_nodes, query_bundle=query_bundle
            )
            rerank_nodes = self.postprocessor.postprocess_nodes(
                nodes=recency_nodes, query_bundle=query_bundle
            )
            px.active_session().url
            return rerank_nodes
        except Exception as e:
            traceback.print_exc()
            raise Exception(f"An error occurred retrieving: {e}")

[dosubot[bot]]

To implement a metadata filter based on MetadataFilters(filters=[ExactMatchFilter(key="namespace", value=self.namespace)]) in the process_streamlit_fetch_query_results function, you can use the _build_metadata_filter_fn method provided in the llama_index library. Here is an example of how you can integrate this into your function:

def process_streamlit_fetch_query_results(self, query="", verbose=False):
    try:
        filters = MetadataFilters(
            filters=[
                ExactMatchFilter(key="namespace", value=self.namespace)
            ]
        )

        # Create the metadata filter function
        metadata_filter_fn = _build_metadata_filter_fn(lambda node_id: self.client.get_metadata(node_id), filters)

        # Initialize BM25 retriever with filters
        bm25_retriever = BM25Retriever.from_defaults(
            docstore=self.docstore, similarity_top_k=self.similarity_top_k
        )
        bm25_retriever.filters = filters

        # Initialize QueryFusionRetriever with the filtered retrievers
        fusion_retriever = QueryFusionRetriever(
            [self.raptor_retriever, bm25_retriever],
            similarity_top_k=self.similarity_top_k,
            num_queries=1,  # set this to 1 to disable query generation
            mode="reciprocal_rerank",
            use_async=True,
            verbose=verbose
        )

        query_bundle = QueryBundle(query_str=query)
        retrieved_nodes = fusion_retriever.retrieve(query_bundle)
        
        # Apply the metadata filter function to the retrieved nodes
        filtered_nodes = [node for node in retrieved_nodes if metadata_filter_fn(node['id'])]

        recency_nodes = self.recency_postprocessor.postprocess_nodes(
            filtered_nodes, query_bundle=query_bundle
        )
        rerank_nodes = self.postprocessor.postprocess_nodes(
            nodes=recency_nodes, query_bundle=query_bundle
        )
        px.active_session().url
        return rerank_nodes
    except Exception as e:
        traceback.print_exc()
        raise Exception(f"An error occurred retrieving: {e}")

In this example, _build_metadata_filter_fn(lambda node_id: self.client.get_metadata(node_id), filters) generates a filter function based on the provided MetadataFilters. This filter function is then used to filter the retrieved nodes based on the metadata.

This approach ensures that the metadata filters are applied to the retrievers within the process_streamlit_fetch_query_results function, allowing you to filter the results based on the specified metadata criteria ^{[1][2][3][4][5]}.

To continue talking to Dosu, mention @dosu.