How PrivateGPT's File Watcher Automates Document Ingestion in Real-Time

PrivateGPT employs a watchdog-based file monitoring system that detects new or modified documents in a watched directory and automatically triggers the Llama-Index ingestion pipeline to embed and store content in the vector database without requiring manual ingestion commands.

The private-gpt repository implements a lightweight watch-and-ingest pipeline that eliminates the need for manual document processing. By leveraging the watchdog library, the system continuously monitors designated folders and instantly processes files as they appear. This automation ensures that the knowledge base remains synchronized with the latest document changes, streamlining the workflow for users managing dynamic document collections.

Architecture of the File Watcher System

The automation relies on three cooperating components that form a complete detection-to-storage pipeline:

IngestWatcher – Located in private_gpt/server/ingest/ingest_watcher.py, this class wraps the watchdog observer to monitor filesystem events recursively and forward paths to a callback.

LocalIngestWorker – Defined in scripts/ingest_folder.py, this worker implements the callback logic that bridges file detection and processing through the ingest_on_watch method.

IngestService – Found in private_gpt/server/ingest/ingest_service.py, this service executes the heavy lifting of parsing documents, generating embeddings, and updating the vector store via Llama-Index.

Step-by-Step Automation Workflow

1. Watcher Initialization and Observer Setup

When the CLI is executed with the --watch flag, scripts/ingest_folder.py instantiates IngestWatcher, passing the target directory and the callback method. Inside IngestWatcher.__init__, a custom Handler subclass of FileSystemEventHandler overrides on_created and on_modified to capture filesystem events.

class Handler(FileSystemEventHandler):
    def on_modified(self, event):
        if isinstance(event, FileModifiedEvent):
            on_file_changed(Path(event.src_path))

    def on_created(self, event):
        if isinstance(event, FileCreatedEvent):
            on_file_changed(Path(event.src_path))

The handler registers with a watchdog.observers.Observer scheduled for recursive monitoring of the watch_path, keeping the process alive via observer.join(1).

2. Event Detection and Callback Execution

When a file is created or modified, the handler invokes the supplied callback—specifically LocalIngestWorker.ingest_on_watch—passing the Path of the changed file. This method logs the detection event and delegates to the internal _do_ingest_one method.

def ingest_on_watch(self, changed_path: Path) -> None:
    logger.info("Detected change at path=%s, ingesting", changed_path)
    self._do_ingest_one(changed_path)

3. Single-File Ingestion Processing

The _do_ingest_one method verifies the file still exists, then calls IngestService.ingest_file to process the document.

def _do_ingest_one(self, changed_path: Path) -> None:
    if changed_path.exists():
        self.ingest_service.ingest_file(changed_path.name, changed_path)

4. Vector Store Integration

The IngestService.ingest_file method forwards the request to the Llama-Index ingestion component. This pipeline parses the document content, creates nodes, computes embeddings using the configured EmbeddingComponent, and persists both the vectors and metadata to the vector store and doc store. The operation returns an IngestedDoc record confirming successful indexing.

Implementation Code Examples

Running the File Watcher from CLI

To activate the automated ingestion pipeline:

python scripts/ingest_folder.py /path/to/docs --watch

IngestWatcher Class Implementation

The complete watcher setup from private_gpt/server/ingest/ingest_watcher.py:

# private_gpt/server/ingest/ingest_watcher.py

class IngestWatcher:
    def __init__(self, watch_path: Path, on_file_changed: Callable[[Path], None]) -> None:
        self.watch_path = watch_path
        self.on_file_changed = on_file_changed

        class Handler(FileSystemEventHandler):
            def on_modified(self, event):
                if isinstance(event, FileModifiedEvent):
                    on_file_changed(Path(event.src_path))

            def on_created(self, event):
                if isinstance(event, FileCreatedEvent):
                    on_file_changed(Path(event.src_path))

        observer = Observer()
        observer.schedule(Handler(), str(watch_path), recursive=True)
        self._observer = observer

Worker Callback Integration

The ingestion trigger from scripts/ingest_folder.py:

# scripts/ingest_folder.py

def ingest_on_watch(self, changed_path: Path) -> None:
    logger.info("Detected change in at path=%s, ingesting", changed_path)
    self._do_ingest_one(changed_path)

def _do_ingest_one(self, changed_path: Path) -> None:
    if changed_path.exists():
        self.ingest_service.ingest_file(changed_path.name, changed_path)

Summary

• Continuous Monitoring: The IngestWatcher leverages watchdog to recursively monitor directories for file creation and modification events.
• Automatic Triggering: The system immediately invokes ingest_on_watch upon detection, eliminating manual ingestion steps.
• Pipeline Execution: The IngestService processes documents through Llama-Index, handling parsing, embedding generation, and vector store updates.
• Defensive Programming: Existence checks in _do_ingest_one prevent errors from race conditions where files are deleted before processing begins.

Frequently Asked Questions

What Python library does PrivateGPT use for file watching?

PrivateGPT uses the watchdog library, specifically the Observer class and FileSystemEventHandler base class to implement asynchronous filesystem monitoring.

Does the file watcher handle subdirectories recursively?

Yes. The IngestWatcher schedules the observer with recursive=True, ensuring that all nested directories within the specified watch path trigger ingestion events when files are added or modified.

Can the watcher detect modifications to existing files?

Yes. The custom Handler class overrides both on_created and on_modified methods, triggering the ingestion pipeline for both new files and updates to existing documents in the watched directory.

What happens if a file is deleted before ingestion completes?

The _do_ingest_one method includes a safety check using changed_path.exists() before calling the ingestion service, which prevents the system from attempting to process files that have been moved or deleted after detection.

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