Repository: VinciGit00/Scrapegraph-ai
File: scrapegraphai/nodes/rag_node.py (Python)

Code before:
        # Fetching data from the state based on the input keys
        input_data = [state[key] for key in input_keys]

        user_prompt = input_data[0]
        doc = input_data[1]

        chunked_docs = []

        for i, chunk in enumerate(doc):
            doc = Document(
                page_content=chunk,
                metadata={
                    "chunk": i + 1,
                },
            )
            chunked_docs.append(doc)

        self.logger.info("--- (updated chunks metadata) ---")

        # check if embedder_model is provided, if not use llm_model
        self.embedder_model = (
            self.embedder_model if self.embedder_model else self.llm_model
        )
        embeddings = self.embedder_model

        folder_name = "cache"

        if self.node_config.get("cache", False) and not os.path.exists(folder_name):
            index = FAISS.from_documents(chunked_docs, embeddings)
            os.makedirs(folder_name)

            index.save_local(folder_name)
        if self.node_config.get("cache", False) and os.path.exists(folder_name):
            index = FAISS.load_local(folder_path=folder_name, embeddings=embeddings)
        else:
            index = FAISS.from_documents(chunked_docs, embeddings)

        retriever = index.as_retriever()

        redundant_filter = EmbeddingsRedundantFilter(embeddings=embeddings)
        # similarity_threshold could be set, now k=20
        relevant_filter = EmbeddingsFilter(embeddings=embeddings)
        pipeline_compressor = DocumentCompressorPipeline(
            transformers=[redundant_filter, relevant_filter]
        )
        # redundant + relevant filter compressor
        compression_retriever = ContextualCompressionRetriever(
            base_compressor=pipeline_compressor, base_retriever=retriever
        )

        # relevant filter compressor only

Review comment (performance):
that's the only thing I'd change.

as it stands, each RAG node with the cache config would overwrite the very same folder, effectively nullifying the cache, unless a RAG node on the very same input state is invoked subsequently.

I would allow for more flexibility by either 1): allowing the user to specify the cache folder in the config for that specific RAG node or 2) automatically compute a unique identifier for that specific input state (e.g., the filename, the endpoint or a some hashing function on the documents) so that the user may choose what to keep and what not to keep in the cache

Code after:

        # Fetching data from the state based on the input keys
        input_data = [state[key] for key in input_keys]

        user_prompt = input_data[0]
        doc = input_data[1]

        chunked_docs = []

        for i, chunk in enumerate(doc):
            doc = Document(
                page_content=chunk,
                metadata={
                    "chunk": i + 1,
                },
            )
            chunked_docs.append(doc)

        self.logger.info("--- (updated chunks metadata) ---")

        # check if embedder_model is provided, if not use llm_model
        self.embedder_model = (
            self.embedder_model if self.embedder_model else self.llm_model
        )
        embeddings = self.embedder_model

        folder_name = self.node_config.get("cache_path", "cache")

        if self.node_config.get("cache_path", False) and not os.path.exists(folder_name):
            index = FAISS.from_documents(chunked_docs, embeddings)
            os.makedirs(folder_name)
            index.save_local(folder_name)
            self.logger.info("--- (indexes saved to cache) ---")

        elif self.node_config.get("cache_path", False) and os.path.exists(folder_name):
            index = FAISS.load_local(folder_path=folder_name,
                                     embeddings=embeddings,
                                     allow_dangerous_deserialization=True)
            self.logger.info("--- (indexes loaded from cache) ---")

        else:
            index = FAISS.from_documents(chunked_docs, embeddings)

        retriever = index.as_retriever()

        redundant_filter = EmbeddingsRedundantFilter(embeddings=embeddings)
        # similarity_threshold could be set, now k=20
        relevant_filter = EmbeddingsFilter(embeddings=embeddings)
        pipeline_compressor = DocumentCompressorPipeline(
            transformers=[redundant_filter, relevant_filter]
        )