check if doc exists before inserting via IngestionPipeline

[johnhu33]

how could we check if a doc_id already exists before inserting via IngestionPipeline?

[dosubot[bot]]

I found a similar solved discussion on how to check if a document exists before inserting it into the index. You can use the _handle_upserts method from the IngestionPipeline class to check for existing document IDs and hashes before performing upserts.

Here is the relevant code snippet:

def _handle_upserts(
    self,
    nodes: Sequence[BaseNode],
    store_doc_text: bool = True,
) -> Sequence[BaseNode]:
    """Handle docstore upserts by checking hashes and ids."""
    assert self.docstore is not None

    doc_ids_from_nodes = set()
    deduped_nodes_to_run = {}
    for node in nodes:
        ref_doc_id = node.ref_doc_id if node.ref_doc_id else node.id_
        doc_ids_from_nodes.add(ref_doc_id)
        existing_hash = self.docstore.get_document_hash(ref_doc_id)
        if not existing_hash:
            # document doesn't exist, so add it
            deduped_nodes_to_run[ref_doc_id] = node
        elif existing_hash and existing_hash != node.hash:
            self.docstore.delete_ref_doc(ref_doc_id, raise_error=False)

            if self.vector_store is not None:
                self.vector_store.delete(ref_doc_id)

            deduped_nodes_to_run[ref_doc_id] = node
        else:
            continue  # document exists and is unchanged, so skip it

    nodes_to_run = list(deduped_nodes_to_run.values())
    self.docstore.add_documents(nodes_to_run, store_text=store_doc_text)

    return nodes_to_run

This method ensures that the document store is checked for existing document IDs and hashes before any new documents are added or updated ^[1][2].

To continue talking to Dosu, mention @dosu.

[johnhu33]

@dosu show me a complete code using IngestionPipeline to add doc, whilst checking if the doc already exists

[dosubot[bot]]

from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.core.ingestion import (
    DocstoreStrategy,
    IngestionPipeline,
    IngestionCache,
)
from llama_index.storage.kvstore.redis import RedisKVStore as RedisCache
from llama_index.storage.docstore.redis import RedisDocumentStore
from llama_index.core.node_parser import SentenceSplitter
from llama_index.vector_stores.redis import RedisVectorStore
from llama_index.core import SimpleDirectoryReader
from redisvl.schema import IndexSchema

# Set up environment variables
import os
os.environ["OPENAI_API_KEY"] = "sk-..."
os.environ["TOKENIZERS_PARALLELISM"] = "false"

# Create seed data
!rm -rf test_redis_data
!mkdir -p test_redis_data
!echo "This is a test file: one!" > test_redis_data/test1.txt
!echo "This is a test file: two!" > test_redis_data/test2.txt

# Load documents with deterministic IDs
documents = SimpleDirectoryReader(
    "./test_redis_data", filename_as_id=True
).load_data()

# Define embedding model
embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5")

# Define custom schema
custom_schema = IndexSchema.from_dict(
    {
        "index": {"name": "redis_vector_store", "prefix": "doc"},
        "fields": [
            {"type": "tag", "name": "id"},
            {"type": "tag", "name": "doc_id"},
            {"type": "text", "name": "text"},
            {
                "type": "vector",
                "name": "vector",
                "attrs": {
                    "dims": 384,
                    "algorithm": "hnsw",
                    "distance_metric": "cosine",
                },
            },
        ],
    }
)

# Set up the ingestion pipeline
pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(),
        embed_model,
    ],
    docstore=RedisDocumentStore.from_host_and_port(
        "localhost", 6379, namespace="document_store"
    ),
    vector_store=RedisVectorStore(
        schema=custom_schema,
        redis_url="redis://localhost:6379",
    ),
    cache=IngestionCache(
        cache=RedisCache.from_host_and_port("localhost", 6379),
        collection="redis_cache",
    ),
    docstore_strategy=DocstoreStrategy.UPSERTS,
)

# Run the pipeline to ingest documents
nodes = pipeline.run(documents=documents)
print(f"Ingested {len(nodes)} Nodes")

# Confirm documents are ingested
from llama_index.core import VectorStoreIndex

index = VectorStoreIndex.from_vector_store(
    pipeline.vector_store, embed_model=embed_model
)

response = index.as_query_engine(similarity_top_k=10).query(
    "What documents do you see?"
)

print(response)

for node in response.source_nodes:
    print(node.get_text())

# Add new data and update existing data
!echo "This is a test file: three!" > test_redis_data/test3.txt
!echo "This is a NEW test file: one!" > test_redis_data/test1.txt

# Load new documents
documents = SimpleDirectoryReader(
    "./test_redis_data", filename_as_id=True
).load_data()

# Run the pipeline again to ingest new and updated documents
nodes = pipeline.run(documents=documents)
print(f"Ingested {len(nodes)} Nodes")

# Confirm updated documents are ingested
index = VectorStoreIndex.from_vector_store(
    pipeline.vector_store, embed_model=embed_model
)

response = index.as_query_engine(similarity_top_k=10).query(
    "What documents do you see?"
)

print(response)

for node in response.source_nodes:
    print(node.get_text())

This code sets up an IngestionPipeline to add documents while checking if they already exist, using Redis for storage and HuggingFace for embeddings. The pipeline handles duplicates by using the DocstoreStrategy.UPSERTS strategy. The docstore actively looks for duplicate documents by storing a map of doc_id to document_hash, re-processing documents if the hash has changed, and skipping them if it hasn't ^[1][2][3][4].

To continue talking to Dosu, mention @dosu.