OVERVIEW

1) Architecture

The ingestion pipeline turns mixed file types into a single, traceable JSONL stream for downstream chunking/embedding. It supports parallel processing via a ThreadPoolExecutor-based engine and real-time progress streaming via Server-Sent Events (SSE).

Pipeline Flow

File discovery under an input root (--input), recursing into all subdirectories.
File classification: supported files are routed by extension; unsupported files (.exe, .dll, .bin, etc.) are flagged and reported.
Modality routing by extension:
- Text: PDF/DOCX/PPTX/MD/HTML/TXT via LangChain + Unstructured loaders.
- Images: OCR (Tesseract default, PaddleOCR optional).
- Audio: faster-whisper transcription with timestamps.
- Video: ffmpeg audio extraction, then faster-whisper.
- CSV: row-wise text rendering.
Minimal normalization (whitespace + encoding cleanup only).
Emit JSONL records with stable doc_id and modality metadata.

File Structure

01_data_ingestion/
├── ingest_multimodal.py          # CLI entry point + extraction functions
├── ingestion/
│   ├── __init__.py
│   ├── engine.py                 # IngestionEngine (parallel orchestrator)
│   ├── events.py                 # SSE event types + emitters
│   └── api.py                    # FastAPI router (upload, start, stream, cancel)
├── tests/
│   ├── conftest.py               # Shared fixtures
│   ├── test_engine.py            # Engine unit tests
│   ├── test_events.py            # Event system tests
│   ├── test_api.py               # API endpoint tests
│   └── test_integration.py       # End-to-end smoke tests
├── validate_ingestion.py         # Output schema validation
└── download_test_datasets.py     # Test data downloader

Output schema (one JSON object per line):

{
  "doc_id": "policies-handbook-2024-pdf__a1b2c3d4",
  "page": 4,
  "section": "page_4",
  "section_index": 7,
  "content": "Employees are entitled to parental leave...",
  "source": {
    "path": "policies/handbook_2024.pdf",
    "filename": "handbook_2024.pdf",
    "filetype": "pdf",
    "modality": "text",
    "loader": "UnstructuredPDFLoader",
    "record_id": "policies-handbook-2024-pdf__a1b2c3d4:000007",
    "prev_record_id": "policies-handbook-2024-pdf__a1b2c3d4:000006",
    "next_record_id": "policies-handbook-2024-pdf__a1b2c3d4:000008"
  }
}

2) Frontend Integration Guide

This section gives frontend developers everything needed to integrate with the ingestion backend. The server runs on http://localhost:8000 via uvicorn server:app --reload --port 8000.

Starting a Job

Option A — Server-side folder:

const res = await fetch("http://localhost:8000/ingestion/start", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ input_dir: "00_Data", workers: 4 }),
});
const { job_id, status } = await res.json();
// → { "job_id": "a1b2c3d4e5f6", "status": "running" }

Option B — File upload (supports .zip):

const formData = new FormData();
formData.append("files", fileInput.files[0]);  // or a .zip
formData.append("workers", "4");

const res = await fetch("http://localhost:8000/ingestion/upload", {
  method: "POST",
  body: formData,
});
const { job_id, status, uploaded_files } = await res.json();
// → { "job_id": "a1b2c3d4e5f6", "status": "running", "uploaded_files": 15 }

Streaming Progress (SSE)

Connect with EventSource to receive real-time events as files are processed:

const evtSource = new EventSource(
  `http://localhost:8000/ingestion/${jobId}/stream`
);

evtSource.addEventListener("ingestion_started", (e) => {
  const data = JSON.parse(e.data);
  console.log(`Processing ${data.total_files} files`);
  // data.skipped_files[] lists unsupported formats that were flagged
});

evtSource.addEventListener("file_started", (e) => {
  const { file_path, file_index, total_files, modality } = JSON.parse(e.data);
  updateUI(`Processing ${file_path} (${file_index + 1}/${total_files})`);
});

evtSource.addEventListener("file_completed", (e) => {
  const { file_path, records_emitted, duration_sec } = JSON.parse(e.data);
  // duration_sec = wall-clock time for this individual file
});

evtSource.addEventListener("file_failed", (e) => {
  const { file_path, error, modality } = JSON.parse(e.data);
  showError(`${file_path}: ${error}`);
});

evtSource.addEventListener("file_skipped", (e) => {
  const { file_path, reason } = JSON.parse(e.data);
  // Unsupported file type flagged here
});

evtSource.addEventListener("ingestion_progress", (e) => {
  const { files_completed, total_files, records_emitted, elapsed_sec, eta_sec } =
    JSON.parse(e.data);
  updateProgressBar(files_completed / total_files);
  showETA(eta_sec);  // estimated seconds remaining
});

evtSource.addEventListener("ingestion_completed", (e) => {
  const { stats } = JSON.parse(e.data);
  // stats.counts.records_emitted, stats.duration_sec, stats.skipped_files[], etc.
  evtSource.close();
});

Polling Fallback

If SSE drops or isn't supported, poll the status endpoint:

async function pollStatus(jobId) {
  const res = await fetch(`http://localhost:8000/ingestion/${jobId}/status`);
  const data = await res.json();
  // → {
  //     "job_id": "...",
  //     "status": "running" | "completed" | "failed" | "cancelled",
  //     "files_completed": 12,
  //     "total_files": 47,
  //     "records_emitted": 892,
  //     "stats": null | { ... },    // populated when completed
  //     "error": null | "..."       // populated when failed
  //   }
  return data;
}

Cancelling a Job

await fetch(`http://localhost:8000/ingestion/${jobId}/cancel`, {
  method: "POST",
});
// Workers finish their current file, then stop cleanly.
// → { "job_id": "...", "status": "cancelling" }

Response Schemas Reference

Endpoint	Method	Response Fields
`/ingestion/start`	POST	`job_id`, `status`
`/ingestion/upload`	POST	`job_id`, `status`, `uploaded_files`
`/ingestion/{id}/stream`	GET	SSE stream (9 event types above)
`/ingestion/{id}/status`	GET	`job_id`, `status`, `files_completed`, `total_files`, `records_emitted`, `stats`, `error`
`/ingestion/{id}/cancel`	POST	`job_id`, `status`

Job Lifecycle

POST /start or /upload  →  pending  →  running  →  completed
                                          │
                                          ├──→  failed  (unrecoverable error)
                                          │
                             POST /cancel →  cancelling  →  cancelled

Jobs auto-expire from server memory 1 hour after completion.

3) Parallel Ingestion Engine

Why Threads (not Processes)

The engine uses ThreadPoolExecutor rather than ProcessPoolExecutor because the heavy workloads — Whisper inference (C/C++ via ctranslate2), FFmpeg (subprocess), Tesseract (subprocess), and Unstructured PDF parsing (C libraries) — all release Python's GIL during their compute phase. This means threads achieve true parallelism for these operations without the overhead of inter-process serialization.

How It Works

┌─────────────────────────────────────────────────────┐
│                  IngestionEngine.run()               │
│                                                     │
│  1. Discover files (rglob)                          │
│  2. Classify → supported[] + skipped[]              │
│  3. Submit to ThreadPoolExecutor(max_workers=N)     │
└────────────────────┬────────────────────────────────┘
                     │
        ┌────────────┼────────────┐
        ▼            ▼            ▼
   ┌─────────┐ ┌─────────┐ ┌─────────┐
   │ Thread 1 │ │ Thread 2 │ │ Thread 3 │  ...N threads
   │          │ │          │ │          │
   │ PDF file │ │ CSV file │ │ image    │  ← run freely in parallel
   │ (no lock)│ │ (no lock)│ │ (no lock)│
   └─────────┘ └─────────┘ └─────────┘

   ┌─────────┐ ┌─────────┐
   │ Thread 4 │ │ Thread 5 │
   │          │ │          │
   │ audio    │ │ video    │  ← must acquire _whisper_lock
   │ (locked) │ │ (waits) │     only one Whisper job at a time
   └─────────┘ └─────────┘

The Whisper Lock

Audio and video files require the Whisper model for transcription. Since:

CPU mode: Whisper saturates all CPU cores — running two transcriptions in parallel would just thrash the CPU with no speed benefit.
GPU mode: The model lives in VRAM — concurrent access would cause memory contention or OOM errors.

A threading.Lock() (_whisper_lock) serializes Whisper access. While one thread is transcribing audio/video, other threads continue processing PDFs, images, CSVs, and text files in parallel. This means a batch of 20 PDFs and 5 videos processes all 20 PDFs concurrently while the 5 videos queue behind each other — the optimal strategy for both CPU and GPU.

Cooperative Cancellation

The engine uses a threading.Event (_cancel). When engine.cancel() is called:

The event is set.
Each thread checks _cancel.is_set() before starting a new file.
Threads finish their current file, then stop.
The executor shuts down with cancel_futures=True to drop queued work.

This gives clean shutdown — no half-written records, no orphaned temp files.

Deterministic Output

Files are submitted to the thread pool with their original sorted index. As threads complete in arbitrary order, results are collected as (index, FileResult) pairs. Before writing to JSONL, results are sorted by index. This guarantees the output is identical whether you use --workers 0 (sequential) or --workers 8 (parallel).

Worker Count Guidelines

`--workers`	Behavior
`0`	Sequential legacy mode (original `ingest()` function, for debugging)
`1`	Single thread through the new engine (useful for isolating issues)
`4` (default)	Good balance for most machines
`8+`	Benefits large batches of text/image files; Whisper is still serialized

4) SSE Event System

Event Types

Event	Fields	When
`ingestion_started`	`job_id`, `total_files`, `skipped_files[]`	After file discovery
`file_started`	`file_path`, `file_index`, `total_files`, `modality`, `filetype`	File begins processing
`file_progress`	`file_path`, `detail`, `percent`	During long operations (video chunks)
`file_completed`	`file_path`, `records_emitted`, `duration_sec`	File done
`file_failed`	`file_path`, `error`, `modality`	File errored
`file_skipped`	`file_path`, `reason`	Unsupported format
`file_empty`	`file_path`, `modality`, `reason`	No content extracted
`ingestion_progress`	`files_completed`, `total_files`, `records_emitted`, `elapsed_sec`, `eta_sec`	After each file
`ingestion_completed`	`stats`	All done

Emitter Architecture

Events flow from worker threads to consumers via an emitter protocol:

Worker threads ──emit()──→ QueueEmitter ──stream()──→ SSE endpoint (async)
                           (thread-safe      (yields SSE-formatted
                            queue.Queue)      strings to StreamingResponse)

QueueEmitter: Bridges sync threads to async HTTP. Workers call emit() which puts events on a queue.Queue. The async stream() generator polls the queue and yields SSE text.
LogEmitter: Prints human-readable progress to stdout (for CLI mode).
NullEmitter: No-op (zero overhead for programmatic use without progress).
MultiEmitter: Broadcasts to multiple emitters simultaneously.

SSE Wire Format

event: file_completed
data: {"event_type": "file_completed", "file_path": "docs/report.pdf", "records_emitted": 45, "duration_sec": 2.3}

event: ingestion_progress
data: {"event_type": "ingestion_progress", "files_completed": 12, "total_files": 47, "records_emitted": 892, "elapsed_sec": 34.5, "eta_sec": 98.2}

5) HTTP API

All endpoints are mounted under /ingestion/ on the FastAPI server.

`POST /ingestion/start`

Start ingestion from a server-side directory.

// Request
{"input_dir": "/path/to/folder", "workers": 4, "pdf_strategy": "auto"}
// Response
{"job_id": "a1b2c3d4e5f6", "status": "running"}

`POST /ingestion/upload`

Upload files (or a .zip) via multipart form data.

curl -F "files=@documents.zip" -F "workers=4" http://localhost:8000/ingestion/upload

// Response
{"job_id": "a1b2c3d4e5f6", "status": "running", "uploaded_files": 15}

`GET /ingestion/{job_id}/stream`

SSE endpoint for real-time progress. Connect with EventSource in JavaScript or curl -N.

`GET /ingestion/{job_id}/status`

Polling fallback if the SSE connection drops.

{"job_id": "...", "status": "running", "files_completed": 12, "total_files": 47, "records_emitted": 892}

`POST /ingestion/{job_id}/cancel`

Cancel a running job. Workers stop after their current file.

6) Decisions and Justifications

LangChain + Unstructured: widest coverage for document formats and metadata.
Tesseract OCR default: light, local, reliable baseline.
PaddleOCR optional: higher OCR fidelity on complex layouts; heavier install.
faster-whisper for ASR: strong local quality, good speed, timestamped segments.
Stable doc_id: slug of relative path + short hash; deterministic and traceable.
Minimal cleanup only: avoid semantic changes before chunking/embedding.
ThreadPoolExecutor over ProcessPoolExecutor: GIL is released by C extensions and subprocesses; threads avoid serialization overhead and share the Whisper model.
Whisper lock: serializes transcription to avoid CPU thrashing or GPU OOM.
Cooperative cancellation: clean shutdown without half-written output.

7) Reproduce Tests (Mac + Windows)

These steps reproduce the exact smoke tests used in this repo. They run each modality separately to keep runtime small.

Run the test suite

cd 01_data_ingestion
python -m pytest tests/ -v

macOS

System deps:

brew install tesseract poppler ffmpeg

Python 3.11 venv (required for faster-whisper on macOS):

brew install python@3.11
python3.11 -m venv Final_venv
source Final_venv/bin/activate
pip install -r 01_data_ingestion/requirements.txt
pip install datasets

Download datasets (lightweight video subset is default):

python 01_data_ingestion/download_test_datasets.py \
  --out 00_Data_test \
  --text --image --audio --video \
  --cord-max-items 50

Run ingestion (smoke test):

# Text
python 01_data_ingestion/ingest_multimodal.py \
  --input 00_Data_test/text \
  --output-dir 01_data_ingestion/output/test_text

# Images (OCR sample of 5)
python 01_data_ingestion/ingest_multimodal.py \
  --input 00_Data_test/image/cord-v2/images \
  --output-dir 01_data_ingestion/output/test_image \
  --max-files 5

# Audio (small subset)
mkdir -p 00_Data_test/audio_subset
find 00_Data_test/audio/librispeech -name "*.flac" | head -n 2 | xargs -I{} cp "{}" 00_Data_test/audio_subset/
python 01_data_ingestion/ingest_multimodal.py \
  --input 00_Data_test/audio_subset \
  --output-dir 01_data_ingestion/output/test_audio

# Video (2 files)
python 01_data_ingestion/ingest_multimodal.py \
  --input 00_Data_test/video/yt-ugc-subset \
  --output-dir 01_data_ingestion/output/test_video \
  --max-files 2

Validate outputs:

python 01_data_ingestion/validate_ingestion.py --input 01_data_ingestion/output/test_text/cleaned_documents.jsonl
python 01_data_ingestion/validate_ingestion.py --input 01_data_ingestion/output/test_image/cleaned_documents.jsonl
python 01_data_ingestion/validate_ingestion.py --input 01_data_ingestion/output/test_audio/cleaned_documents.jsonl
python 01_data_ingestion/validate_ingestion.py --input 01_data_ingestion/output/test_video/cleaned_documents.jsonl

Windows (PowerShell)

System deps (Chocolatey):

choco install tesseract poppler ffmpeg

Python 3.11 venv:

py -3.11 -m venv Final_venv
.\Final_venv\Scripts\Activate.ps1
pip install -r 01_data_ingestion/requirements.txt
pip install datasets

Download datasets:

python 01_data_ingestion\download_test_datasets.py `
  --out 00_Data_test `
  --text --image --audio --video `
  --cord-max-items 50

Run ingestion (smoke test):

# Text
python 01_data_ingestion\ingest_multimodal.py `
  --input 00_Data_test\text `
  --output-dir 01_data_ingestion\output\test_text

# Images (OCR sample of 5)
python 01_data_ingestion\ingest_multimodal.py `
  --input 00_Data_test\image\cord-v2\images `
  --output-dir 01_data_ingestion\output\test_image `
  --max-files 5

# Audio (small subset)
New-Item -ItemType Directory -Force -Path 00_Data_test\audio_subset | Out-Null
Get-ChildItem -Recurse -Path 00_Data_test\audio\librispeech -Filter *.flac | Select-Object -First 2 | Copy-Item -Destination 00_Data_test\audio_subset
python 01_data_ingestion\ingest_multimodal.py `
  --input 00_Data_test\audio_subset `
  --output-dir 01_data_ingestion\output\test_audio

# Video (2 files)
python 01_data_ingestion\ingest_multimodal.py `
  --input 00_Data_test\video\yt-ugc-subset `
  --output-dir 01_data_ingestion\output\test_video `
  --max-files 2

Validate outputs:

python 01_data_ingestion\validate_ingestion.py --input 01_data_ingestion\output\test_text\cleaned_documents.jsonl
python 01_data_ingestion\validate_ingestion.py --input 01_data_ingestion\output\test_image\cleaned_documents.jsonl
python 01_data_ingestion\validate_ingestion.py --input 01_data_ingestion\output\test_audio\cleaned_documents.jsonl
python 01_data_ingestion\validate_ingestion.py --input 01_data_ingestion\output\test_video\cleaned_documents.jsonl

Notes

Default video URLs are direct .webm files; no yt-dlp required.
Audio/video ingestion requires Python 3.10-3.12 on macOS due to onnxruntime.
OCR can return empty text; those files are skipped to avoid noisy records.
The --workers 0 flag falls back to the original sequential ingest() function for backward compatibility and debugging.