How to Use Pipeline Insert (`apipeline_enqueue_documents`) for Incremental Document Indexing in LightRAG

Call await rag.apipeline_enqueue_documents(input) to asynchronously enqueue new documents, automatically deduplicate content using MD5 hashes, and append them to your existing LightRAG knowledge graph without re-indexing previously processed text.

The apipeline_enqueue_documents method serves as the core pipeline insert API in the HKUDS/LightRAG repository, enabling incremental document indexing by managing the complete lifecycle of document ingestion—from input normalization to atomic storage—in a single asynchronous operation. Unlike bulk indexing operations that require rebuilding the entire graph, this method checks existing document status before writing, ensuring that only new content triggers downstream processing while duplicate attempts are logged for audit.

How the Incremental Pipeline Works

When you invoke apipeline_enqueue_documents in lightrag/lightrag.py, the method executes a ten-step workflow designed for safe, concurrent document ingestion:

1. Track ID Generation – If you omit the track_id parameter, the system generates a unique identifier with an "enqueue" prefix to monitor this specific batch (lines L0112-L0114).
2. Input Normalization – The method accepts strings or lists, wrapping single documents into lists and validating that optional ids and file_paths match the document count (lines L0115-L0120).
3. File Path Association – Each document is optionally linked to its source file; missing paths default to "unknown_source" (lines L0121-L0135).
4. Content Sanitization – Raw text is sanitized via sanitize_text_for_encoding and deduplicated in a single pass to remove exact duplicates within the batch (lines L0137-L0158).
5. ID Assignment – When ids are not provided, compute_mdhash_id generates MD5-based identifiers for each unique content string (lines L0159-L0174).
6. Status Record Creation – For every unique document, the pipeline constructs a DocStatus.PENDING entry containing a content summary, length, timestamps, file path, and the batch track ID (lines L0176-L0189).
7. Duplicate Filtering – The method queries self.doc_status.filter_keys to identify which document IDs already exist in the index, skipping re-processing of existing content (lines L0192-L0197).
8. Duplicate Audit Trail – For documents already present in the graph, the system creates a failed status record with a dup- prefixed ID, preserving an audit trail without modifying the existing index (lines L0198-L0243).
9. Atomic Persistence – Unique documents are written to self.full_docs (raw content) and self.doc_status (metadata) in a single atomic step (lines L0255-L0272).
10. Return Tracking ID – The method returns the track_id for downstream monitoring of the asynchronous processing queue (line L0273).

Implementation Details and Key Functions

The incremental indexing logic relies on several critical components defined across the LightRAG codebase:

• lightrag/types.py – Defines the DocStatus enum (PENDING, PROCESSING, DONE, FAILED) used to track document state throughout the pipeline.
• lightrag/utils.py – Provides helper functions including sanitize_text_for_encoding (encoding normalization), compute_mdhash_id (MD5 hash generation), generate_track_id (unique batch identifiers), and get_content_summary (content length and snippet extraction).
• lightrag/kg/*_impl.py – Concrete storage implementations (e.g., qdrant_impl.py, postgres_impl.py) handle the actual persistence of full_docs and doc_status data.

The pipeline requires asynchronous execution, allowing you to enqueue documents continuously as new data arrives without blocking the main application thread.

Basic Usage Example

The following example demonstrates how to initialize LightRAG and incrementally index a batch of documents:

import asyncio
from lightrag.lightrag import LightRAG

async def index_documents():
    # Initialize with your preferred storage backends

    rag = LightRAG(
        vector_store="qdrant",
        metadata_store="postgres",
        workspace="production_kb"
    )
    
    # Prepare new documents for ingestion

    new_docs = [
        "The quick brown fox jumps over the lazy dog.",
        "Artificial intelligence is transforming many industries."
    ]
    
    # Optional: track original file sources for citation

    source_paths = ["notes/fox.txt", "reports/ai_overview.pdf"]
    
    # Enqueue documents asynchronously

    track_id = await rag.apipeline_enqueue_documents(
        input=new_docs,
        file_paths=source_paths
    )
    
    print(f"Batch queued with track_id: {track_id}")

# Run the async operation

asyncio.run(index_documents())

Handling Duplicates and Idempotency

The pipeline insert API is idempotent by design. When you attempt to enqueue content that already exists in the knowledge graph, the system performs the following actions:

• Skips Re-indexing – Existing document IDs (identified by MD5 hash) are filtered out before any write operations occur, preventing unnecessary processing.
• Logs Duplicate Attempts – A failed status record is created with a dup-<original_id> identifier and stored in doc_status, providing a clear audit trail of duplicate detection events.
• Preserves Existing Data – The original document in full_docs and its status remain untouched, ensuring that existing knowledge graph relationships stay intact.

This behavior makes it safe to rerun the same ingestion scripts repeatedly—only new or modified content triggers expensive embedding and graph extraction operations.

Monitoring Indexing Progress with track_id

The track_id returned by apipeline_enqueue_documents enables granular monitoring of asynchronous processing:

# Poll the status of a specific batch

batch_status = await rag.doc_status.get_by_track_id(track_id)
print(batch_status)  # Shows documents in PENDING, PROCESSING, DONE, or FAILED states

You can correlate this tracking ID with logs in doc_status to identify which documents from a specific batch have completed processing or failed due to errors. This is particularly useful when ingesting large document sets where individual files may succeed or fail independently.

Summary

• apipeline_enqueue_documents in lightrag/lightrag.py is the core pipeline insert method for incremental document indexing in LightRAG.
• The method accepts input (string or list), optional file_paths, optional ids, and an optional track_id parameter.
• It automatically generates MD5-based document IDs via compute_mdhash_id and filters duplicates using doc_status.filter_keys before writing to storage.
• Duplicate attempts are logged as failed records with dup- prefixed IDs, ensuring auditability without modifying existing indexed content.
• Documents are written atomically to full_docs (raw text) and doc_status (metadata), returning a track_id for progress monitoring.

Frequently Asked Questions

What happens if I enqueue the same document twice?

When you enqueue duplicate content, apipeline_enqueue_documents detects the existing MD5 hash in doc_status and skips re-processing. It creates a failed status record with a dup- prefixed ID to log the attempt, but leaves the original document and its knowledge graph connections completely unchanged.

Can I specify my own document IDs instead of using MD5 hashes?

Yes. Pass a list of unique identifiers to the ids parameter when calling apipeline_enqueue_documents. The method validates that the length matches the document count and that IDs are unique within the batch. If omitted, the system automatically generates MD5 hashes via compute_mdhash_id.

How do I handle partial failures in a batch?

The track_id returned by the method allows you to query specific document statuses using rag.doc_status.get_by_track_id(). Documents that fail during the pipeline (including duplicates) receive a FAILED status with error details, while successful documents transition to DONE. You can filter by status to retry only failed items.

Is apipeline_enqueue_documents thread-safe for concurrent ingestion?

Yes. The method is designed for asynchronous execution and safely handles concurrent calls. The atomic write operations to full_docs and doc_status prevent race conditions, though individual storage backends (e.g., Qdrant, PostgreSQL) should be configured to handle your specific concurrency requirements.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}