How to refresh and index document based on metadata?

[d3buggerdan]

So, lets say I have a user base of people with their own documents that get indexed with the metadata having a unique userid and docid for each document and embedding to filter through. Lets say one user has an updated document, meaning I only need to refresh the index related to the vectors related to the proper userid and docid in the metadata. Is there anyway to do this without needing to refresh the whole index because then it defeats the point. From my understanding the only way to refresh an index is to have your initial and new document and refresh the whole index, but this defeats the point if the same index contains documents from multiple users. Thanks.

[dosubot[bot]]

Hey @d3buggerdan! I'm a bot here to help you with any bugs, questions, or contributions. Let's tackle this issue together!

To refresh and index a document based on metadata without needing to refresh the entire index, you can use the refresh_ref_docs method. This method allows you to update only the documents that have changed in text or metadata, and it will also insert any documents that were not previously stored.

Here is the relevant code snippet:

def refresh_ref_docs(
        self, documents: Sequence[Document], **update_kwargs: Any
    ) -> List[bool]:
        """Refresh an index with documents that have changed."""
        with self._callback_manager.as_trace("refresh"):
            pipeline_id = self._get_pipeline_id()
            upserted_documents = self._client.pipelines.upsert_batch_pipeline_documents(
                pipeline_id=pipeline_id,
                request=[
                    CloudDocumentCreate(
                        text=doc.text,
                        metadata=doc.metadata,
                        excluded_embed_metadata_keys=doc.excluded_embed_metadata_keys,
                        excluded_llm_metadata_keys=doc.excluded_llm_metadata_keys,
                        id=doc.id_,
                    )
                    for doc in documents
                ],
            )
            doc_ids = [doc.id for doc in upserted_documents]
            self._wait_for_documents_ingestion(
                doc_ids, verbose=True, raise_on_error=True
            )
            return [True] * len(doc_ids)

This method checks if the document already exists in the document store and whether its content has changed. If the document is new or has changed, it updates the document in the index. This way, you can refresh and index documents based on their metadata without needing to refresh the entire index.

The refresh_ref_docs function will return a boolean array, indicating which documents in the input were refreshed or inserted. This allows you to confirm that only the new or updated documents are processed ^[1][2][3][4].

To continue talking to Dosu, mention @dosu.

[dosubot[bot]]

The original 11 chunks were processed and added to the dataset. When you refresh the index with a new document, the refresh_ref_docs method only updates or inserts the provided documents without affecting the existing ones unless they have the same doc_id and different content.

Here's a detailed explanation of how the chunks are handled:

1. Processing Chunks: The run method in the RAFTDatasetPack class retrieves the chunks using the get_chunks method and processes each chunk by calling the add_chunk_to_dataset method. This method generates questions and adds the chunk along with distractor documents to the dataset.

   def run(self) -> Any:
       """Run the pipeline."""
       chunks = self.get_chunks(self.file_path, self.chunk_size)
   
       logger.info(f"Number of chunks created: {len(chunks)}")
   
       self.num_distract_docs = (
           min(self.num_distract_docs, len(chunks)) - 1
       )  # should be less than number of chunks/ nodes created
   
       for index, chunk in enumerate(chunks):
           logger.info(f"Processing chunk: {index}")
           self.add_chunk_to_dataset(
               chunks, chunk, self.num_questions_per_chunk, self.num_distract_docs
           )
   
       return self.ds

2. Indexing Chunks: In the index_docset function, the chunks are loaded and processed to build separate maps of chunks and parents. The chunks are then used to build full document summaries and are associated with parent chunks. These chunks and summaries are then updated in the local index and populated into a vector store for retrieval.

   def index_docset(docset_id: str, name: str, overwrite=False):
       """
       Indexes the given docset.
       """
       print(f"Indexing {name} (ID: {docset_id})")
   
       loader = DocugamiReader(
           min_text_length=MIN_CHUNK_TEXT_LENGTH,
           max_text_length=MAX_CHUNK_TEXT_LENGTH,
           sub_chunk_tables=SUB_CHUNK_TABLES,
           include_xml_tags=INCLUDE_XML_TAGS,
           parent_hierarchy_levels=PARENT_HIERARCHY_LEVELS,
           parent_id_key=PARENT_DOC_ID_KEY,
           include_project_metadata_in_doc_metadata=False,  # not used, so lighten the vector index
       )
       chunks = loader.load_data(docset_id=docset_id)
   
       # Build separate maps of chunks, and parents
       parent_chunks_by_id: Dict[str, Document] = {}
       chunks_by_source: Dict[str, List[str]] = {}
       for chunk in chunks:
           chunk_id = str(chunk.metadata.get("id"))
           chunk_source = str(chunk.metadata.get(SOURCE_KEY))
           parent_chunk_id = chunk.metadata.get(loader.parent_id_key)
           if not parent_chunk_id:
               # parent chunk, we will use this (for expanded context) as our chunk
               parent_chunks_by_id[chunk_id] = chunk
           else:
               # child chunk, we will keep track of this to build up our
               # full document summary
               if chunk_source not in chunks_by_source:
                   chunks_by_source[chunk_source] = []
   
               chunks_by_source[chunk_source].append(chunk.text)
   
       # Build up the full docs by concatenating all the child chunks
       full_docs_by_id: Dict[str, Document] = {}
       full_doc_ids_by_source: Dict[str, str] = {}
       for source in chunks_by_source:
           chunks_from_source = chunks_by_source[source]
           full_doc_text = "\n".join(c for c in chunks_from_source)
           full_doc_id = hashlib.md5(full_doc_text.encode()).hexdigest()
           full_doc_ids_by_source[source] = full_doc_id
           full_docs_by_id[full_doc_id] = Document(
               text=full_doc_text, metadata={"id": full_doc_id}
           )
   
       # Associate parent chunks with full docs
       for parent_chunk_id in parent_chunks_by_id:
           parent_chunk = parent_chunks_by_id[parent_chunk_id]
           parent_chunk_source = parent_chunk.metadata.get(SOURCE_KEY)
           if parent_chunk_source:
               full_doc_id = full_doc_ids_by_source.get(parent_chunk_source)
               if full_doc_id:
                   parent_chunk.metadata[FULL_DOC_SUMMARY_ID_KEY] = full_doc_id
   
       full_doc_summaries_by_id = build_full_doc_summary_mappings(full_docs_by_id)
       chunk_summaries_by_id = build_chunk_summary_mappings(parent_chunks_by_id)
   
       direct_tool_function_name = docset_name_to_direct_retriever_tool_function_name(name)
       direct_tool_description = chunks_to_direct_retriever_tool_description(
           name, list(parent_chunks_by_id.values())
       )
       report_details = build_report_details(docset_id)
   
       if overwrite:
           state = Path(INDEXING_LOCAL_STATE_PATH)
           if state.is_file() and state.exists():
               os.remove(state)
   
       update_local_index(
           docset_id=docset_id,
           full_doc_summaries_by_id=full_doc_summaries_by_id,
           chunks_by_id=parent_chunks_by_id,  # we are using the parent chunks as chunks for expanded context
           direct_tool_function_name=direct_tool_function_name,
           direct_tool_description=direct_tool_description,
           report_details=report_details,
       )
   
       populate_vector_index(
           docset_id, chunks=list(chunk_summaries_by_id.values()), overwrite=overwrite
       )

3. Semantic Splitting: The SemanticSplitterNodeParser class is used to split documents into semantically related chunks. This process involves grouping sentences based on semantic similarity, calculating distances between these groups, and then forming chunks based on these distances.

   class SemanticSplitterNodeParser(NodeParser):
       """Semantic node parser.
   
       Splits a document into Nodes, with each node being a group of semantically related sentences.
   
       Args:
           buffer_size (int): number of sentences to group together when evaluating semantic similarity
           embed_model: (BaseEmbedding): embedding model to use
           sentence_splitter (Optional[Callable]): splits text into sentences
           include_metadata (bool): whether to include metadata in nodes
           include_prev_next_rel (bool): whether to include prev/next relationships
       """
   
       sentence_splitter: Callable[[str], List[str]] = Field(
           default_factory=split_by_sentence_tokenizer,
           description="The text splitter to use when splitting documents.",
           exclude=True,
       )
   
       embed_model: BaseEmbedding = Field(
           description="The embedding model to use to for semantic comparison",
       )
   
       buffer_size: int = Field(
           default=1,
           description=(
               "The number of sentences to group together when evaluating semantic similarity. "
               "Set to 1 to consider each sentence individually. "
               "Set to >1 to group sentences together."
           ),
       )
   
       breakpoint_percentile_threshold = Field(
           default=95,
           description=(
               "The percentile of cosine dissimilarity that must be exceeded between a "
               "group of sentences and the next to form a node.  The smaller this "
               "number is, the more nodes will be generated"
           ),
       )
   
       def _build_node_chunks(
           self, sentences: List[SentenceCombination], distances: List[float]
       ) -> List[str]:
           chunks = []
           if len(distances) > 0:
               breakpoint_distance_threshold = np.percentile(
                   distances, self.breakpoint_percentile_threshold
               )
   
               indices_above_threshold = [
                   i for i, x in enumerate(distances) if x > breakpoint_distance_threshold
               ]
   
               # Chunk sentences into semantic groups based on percentile breakpoints
               start_index = 0
   
               for index in indices_above_threshold:
                   end_index = index - 1
   
                   group = sentences[start_index : end_index + 1]
                   combined_text = "".join([d["sentence"] for d in group])
                   chunks.append(combined_text)
   
                   start_index = index
   
               if start_index < len(sentences):
                   combined_text = "".join(
                       [d["sentence"] for d in sentences[start_index:]]
                   )
                   chunks.append(combined_text)
           else:
               # If, for some reason we didn't get any distances (i.e. very, very small documents) just
               # treat the whole document as a single node
               chunks = [" ".join([s["sentence"] for s in sentences])]
   
           return chunks

The chunks are thus incorporated into the dataset and indexed for retrieval through this comprehensive process ^[1][2][3].

To continue talking to Dosu, mention @dosu.

[d3buggerdan]

@dosu how can i see all of the documents in the index

[dosubot[bot]]

To see all of the documents in the index, you can inspect the index.storage_context.docstore.docs attribute. Here is an example of how you can do this:

# Assuming `index` is your GPTSimpleVectorIndex instance
documents = index.storage_context.docstore.docs
for doc_id, doc in documents.items():
    print(f"Document ID: {doc_id}")
    print(f"Document Content: {doc['content']}")
    print(f"Document Metadata: {doc.get('metadata', {})}")
    print("\n")

This code will iterate through all the documents in the index and print their IDs, content, and metadata ^[1][2].

To continue talking to Dosu, mention @dosu.

[d3buggerdan]

@dosu

Is this the same as the doc id I provided? How can i reference these values for example can i put doc_id to 1 and it will know which hash it is?

Document ID: a45bd56b-01e8-4217-8a86-24f3b3a72842
Document ID: 70491b4e-d402-449c-9832-6f5821682abb
Document ID: 1b1ac4a4-234f-4a0b-b670-701c6db69faf
Document ID: 3ed1f785-f356-424c-8dc1-86aef928a5f0
Document ID: fde3f8ec-9fed-410c-913c-74d6eecc4741
Document ID: 3ee2a69c-d379-4f1b-9a92-813d92aac570
Document ID: 5f92aea0-ceda-4fbc-aba3-8146bba3d3ff
Document ID: 93e50da5-9c5e-4788-81db-57f2410ec094
Document ID: 56a0b5a7-b722-4823-aac1-b6c2163d6600
Document ID: 20389a3b-8990-433e-be49-4535488a7c49
Document ID: 34537db4-648c-445b-9dae-b4ea400d8071
Document ID: f3830502-ffbb-44fc-97f5-a98fb4baaa29
Document ID: 51c40145-0afe-41e2-a56f-8633787effd2
Document ID: 971e515c-6920-41eb-992c-7c8336c2f252
Document ID: 982e6428-44d9-4a5d-93af-9b9489d7820c
Document ID: f5ece504-2f49-496b-9c5d-8b0923625683
Document ID: f83b7c5a-18d1-43ed-bccd-d02d58981095
Document ID: e124546e-e3cc-4bf6-b5e5-ee226243574d
Document ID: 54172637-605d-422b-9226-3bfc81773ede
Document ID: b84c65fb-2907-4350-9b97-bad7b5b9055a
Document ID: 99ef1dcf-bb92-4858-8d59-39e297e565a2
Document ID: 50791bdb-c4ae-4a50-b360-3c1605b6778b
Document ID: 5615dbea-9226-483f-917e-e124ec7860d6
Document ID: b805e160-b120-4285-ba9c-5c5086b495b4

[dosubot[bot]]

No, you cannot directly reference the document IDs from your index if you set doc_id to a simple value like 1. The document ID you assign does not actually set the doc_id in the docstore. Instead, a generated ID is used.

Supporting evidence from the document:

def get_document_hash(self, doc_id: str) -> Optional[str]:
    """Get the stored hash for a document, if it exists."""
    metadata = self._kvstore.get(doc_id, collection=self._metadata_collection)
    if metadata is not None:
        return metadata.get("doc_hash", None)
    else:
        return None

This shows that the doc_id is not directly set by the user but is instead generated and referenced by a ref_doc_id ^[1].

If you want to use simpler or more human-readable document IDs, you can set the doc_id of any Document directly:

document.doc_id = "My new document id!"

Additionally, when using the SimpleDirectoryReader, you can automatically set the doc_id to be the full path to each document:

from llama_index.core import SimpleDirectoryReader

documents = SimpleDirectoryReader("./data", filename_as_id=True).load_data()
print([x.doc_id for x in documents])

This approach allows you to customize the document IDs to be more human-readable or meaningful based on your specific requirements ^[2][3].

To continue talking to Dosu, mention @dosu.