How LMForge Handles Partial Failures in Batch Document Processing Pipelines

LMForge processes documents in isolated try/except blocks with chunked vector store writes, ensuring that a failure in one document or segment does not cascade to the entire batch.

LMForge is an open-source LLMOps platform designed for multi-model agent orchestration. When handling large-scale document ingestion, the system implements sophisticated fault-tolerance mechanisms to manage partial failures in batch document processing pipelines without compromising pipeline integrity or data consistency.

Batch Creation and Asynchronous Task Dispatch

The pipeline begins in api/internal/service/document_service.py, where the DocumentService.create_documents method generates a unique batch identifier and initiates asynchronous processing.

When a user uploads files, the system creates a batch ID using a timestamp and random suffix:

batch = time.strftime("%Y%m%d%H%M%S") + str(random.randint(100000, 999999))
process_rule = self.create(ProcessRule, account_id=account.id, dataset_id=dataset_id,
                         mode=process_type, rule=rule)

After creating Document rows in the database, the system dispatches the build_documents Celery task defined in api/internal/task/document_task.py. This asynchronous handoff allows the API to return immediately while the heavy processing occurs in the background.

Per-Document Error Isolation

The core resilience logic resides in api/internal/service/indexing_service.py. The IndexingService.build_documents method iterates over every document in the batch, wrapping each in an independent exception handler.

This isolation ensures that a parsing error, encoding issue, or processing failure in one document does not terminate the entire batch:

for document in documents:
    try:
        # parsing → splitting → indexing → completion

        ...
    except Exception as e:
        logging.exception(f"构建文档发生错误: {e}")
        self.update(document,
                    status=DocumentStatus.ERROR.value,
                    error=str(e),
                    stopped_at=datetime.now())

When an exception occurs, the system updates the document's status to ERROR, records the exception message, and continues processing the remaining documents in the batch.

Chunked Vector Store Writes with Partial Failure Recovery

The most failure-prone stage involves writing document segments to the vector database. LMForge implements chunked writes with granular error handling to prevent partial corruption of the vector store.

Processing Segments in Batches of 10

Within IndexingService._completed, the system processes segments in chunks of 10 to balance transaction size and failure granularity:

for i in range(0, len(lc_segments), 10):
    chunks = lc_segments[i:i + 10]
    ids = [c.metadata["node_id"] for c in chunks]

Each chunk receives a unique set of node IDs, enabling precise tracking of which segments succeeded or failed.

Isolating Vector Store Failures

The system wraps each chunk write in a nested try/except block. If the vector store write fails, only that specific chunk is marked as failed:

try:
    self.vector_database_service.vector_store.add_documents(documents=chunks, ids=ids)
    self.db.session.query(Segment).filter(Segment.node_id.in_(ids)).update({
        "status": SegmentStatus.COMPLETED.value,
        "completed_at": datetime.now(),
        "enabled": True,
    })
except Exception:
    self.db.session.query(Segment).filter(Segment.node_id.in_(ids)).update({
        "status": SegmentStatus.ERROR.value,
        "completed_at": None,
        "stopped_at": datetime.now(),
        "enabled": False,
    })
    raise

The inner exception handler updates the segment rows to ERROR status, disables them, and re-raises the exception. An outer try/except block in build_documents catches this re-raised exception to mark the parent document as failed, while previously committed chunks remain in COMPLETED status.

Transactional Cleanup and Concurrency Controls

Beyond processing failures, LMForge implements safeguards for deletion and concurrent modifications.

When removing documents, the delete_document method in indexing_service.py removes segment rows, vector-store entries, and keyword-table rows within a single database transaction. Any exception triggers a rollback, preventing partial deletion states that could orphan vector embeddings.

For status updates, the system uses Redis distributed locking via LOCK_DOCUMENT_UPDATE_ENABLED. When enabling or disabling a document, the code attempts to acquire this lock; if another process holds it, the request fails immediately with a FailException, preventing race conditions that could corrupt document states during concurrent batch operations.

Summary

LMForge handles partial failures in batch document processing pipelines through several architectural patterns:

• Isolated per-document processing with individual try/except blocks prevents single document failures from cascading to the entire batch.
• Chunked vector store writes process segments in groups of 10 with granular error handling, ensuring only failed chunks are marked as errors while successful commits persist.
• Hierarchical error propagation uses nested exception handlers to mark specific segments, then documents, as failed while allowing the batch iteration to continue.
• Transactional cleanup and distributed locking prevent partial states during deletion or concurrent status updates.

Frequently Asked Questions

What happens when a single document fails in a batch?

When IndexingService.build_documents encounters an exception while processing a specific document, it catches the error, updates the document's status to ERROR in the database, records the exception message, and continues iterating through the remaining documents in the batch. The Celery task completes successfully even if individual documents fail, ensuring pipeline stability.

How does LMForge prevent vector store corruption during partial failures?

The system writes document segments to the vector database in chunks of 10, with each chunk wrapped in its own transaction. If a vector store write fails, the code catches the exception, marks only those specific segment rows as ERROR and disabled, then re-raises the exception. Previously committed chunks remain in COMPLETED status, preventing partial corruption of the vector index.

Can failed documents be retried individually?

Yes. Because each document maintains an independent status field (DocumentStatus.ERROR), operators can identify specific failed documents through the database or API. The modular design of IndexingService.build_documents allows re-running the pipeline for individual document IDs without reprocessing the entire batch, though the specific retry mechanism would depend on the operational wrapper or UI implementation.

What concurrency controls protect document status updates?

LMForge uses Redis distributed locking via the LOCK_DOCUMENT_UPDATE_ENABLED key when updating a document's enabled status. Before modifying the status, the system attempts to acquire this lock; if another process currently holds it, the request immediately raises a FailException. This prevents race conditions where concurrent batch operations might otherwise leave documents in inconsistent partial states.

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