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

> Learn to use LightRAG's apipeline_enqueue_documents for efficient incremental document indexing. Asynchronously add new content, deduplicate, and update your knowledge graph without full re-indexing.

- Repository: [✨Data Intelligence Lab@HKU✨/LightRAG](https://github.com/HKUDS/LightRAG)
- Tags: how-to-guide
- Published: 2026-03-23

---

**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`](https://github.com/HKUDS/LightRAG/blob/main/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`](https://github.com/HKUDS/LightRAG/blob/main/lightrag/types.py)** – Defines the `DocStatus` enum (`PENDING`, `PROCESSING`, `DONE`, `FAILED`) used to track document state throughout the pipeline.
- **[`lightrag/utils.py`](https://github.com/HKUDS/LightRAG/blob/main/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`](https://github.com/HKUDS/LightRAG/blob/main/qdrant_impl.py), [`postgres_impl.py`](https://github.com/HKUDS/LightRAG/blob/main/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:

```python
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:

```python

# 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`](https://github.com/HKUDS/LightRAG/blob/main/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.