Skip to content

CHHemant/nlp-data-pipeline

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

6 Commits
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

End-to-End AI Data Pipeline & NLP Automation System

BERT-based NLP pipeline processing 20,000+ records/day with anomaly detection, schema validation, and real-time performance monitoring.

Python BERT MongoDB React License


What this does

A production NLP pipeline built around three problems that kept breaking data-dependent ML systems in practice:

  1. Bad data goes in → bad model comes out — most pipelines have no anomaly detection step
  2. You can't fix what you can't see — no real-time visibility into data quality or model health
  3. Schema drift kills pipelines silently — a field changes upstream, everything breaks downstream

This system addresses all three with a BERT-based detection layer, automated schema validation, and a live React dashboard.

Results: 93% downstream model accuracy (up from baseline 74%), processing time reduced 65%, 20,000+ records/day throughput.


Architecture

Data Sources (APIs, files, streams)
          │
          ▼
┌─────────────────────────────────┐
│       Ingestion Layer           │
│   schema validation · typing    │
│   deduplication · normalisation │
└────────────┬────────────────────┘
             │
             ▼
┌─────────────────────────────────┐
│     BERT Processing Layer       │
│                                 │
│  ┌──────────┐  ┌─────────────┐  │
│  │ Anomaly  │  │    Text     │  │
│  │Detector  │  │Classifier   │  │
│  └──────────┘  └─────────────┘  │
│  ┌──────────────────────────┐   │
│  │  Entity / Relation       │   │
│  │  Extractor (spaCy+BERT)  │   │
│  └──────────────────────────┘   │
└────────────┬────────────────────┘
             │
             ▼
┌─────────────────────────────────┐
│       Storage Layer             │
│   MongoDB (documents)           │
│   PostgreSQL (metrics + logs)   │
└────────────┬────────────────────┘
             │
             ▼
┌─────────────────────────────────┐
│    REST API (FastAPI)           │
│    + WebSocket (live updates)   │
└────────────┬────────────────────┘
             │
             ▼
┌─────────────────────────────────┐
│   React + D3.js Dashboard       │
│   real-time model metrics       │
│   anomaly alerts · throughput   │
└─────────────────────────────────┘

Key results

Metric Result Baseline
Downstream model accuracy 93% 74%
Processing time -65% pre-optimisation
Daily throughput 20,000+ records
Anomaly detection precision 91%
Schema violation catch rate 99.2%
Pipeline uptime 99.7%

The anomaly detection approach

Vanilla anomaly detection (z-score, isolation forest) works for numerical data. It breaks on text. BERT gives us a way to detect semantic anomalies — records that are structurally valid but semantically wrong.

Three-layer detection:

Layer 1 — Structural (fast, runs on every record)

def validate_schema(record: dict, schema: Schema) -> ValidationResult:
    # type checking, required fields, value ranges
    # catches ~60% of bad records at near-zero cost

Layer 2 — Statistical (medium, runs on batches)

def detect_distribution_shift(batch: list, baseline: Stats) -> ShiftScore:
    # compare current batch distribution to rolling baseline
    # flags when field value distributions drift
    # catches schema drift from upstream changes

Layer 3 — Semantic (expensive, runs on flagged records)

def bert_anomaly_score(record: dict, context: str) -> float:
    # encode record text with BERT
    # compare embedding to cluster of known-good records
    # cosine distance > 0.4 = anomaly flag

Running all three on every record is too slow (tested — it is). The cascade approach — only escalate to BERT when layers 1+2 pass — keeps throughput at 20k+/day while catching 91% of real anomalies.


Ablation study — preprocessing strategies

The 65% processing time reduction came from testing which preprocessing steps actually matter:

Strategy Accuracy Time/record
No preprocessing 74% 12ms
Lowercasing + stopwords 76% 9ms
+ Lemmatisation 80% 14ms
+ Entity normalisation 86% 18ms
+ Semantic dedup 93% 4ms (batched)

Semantic deduplication was the biggest win — many records were near-duplicates that were confusing the downstream model. Removing them improved accuracy more than any other single step, and batched embedding comparison is faster than per-record processing.

Full analysis in notebooks/02_ablation_preprocessing.ipynb.


Dashboard

Real-time monitoring built in React + D3.js. Connects to the FastAPI WebSocket endpoint and updates every 5 seconds.

Panels:

  • Throughput — records/min rolling average
  • Anomaly rate — % flagged per time window with drill-down
  • Model accuracy — live accuracy on validation split
  • Schema violations — breakdown by field and violation type
  • Processing latency — p50/p90/p99 per pipeline stage

The dashboard is intentionally simple. It's built for a non-technical stakeholder who needs to know one thing: is the pipeline healthy right now? Green = yes, red = no.


Project structure

nlp-data-pipeline/
│
├── src/
│   ├── ingestion/
│   │   ├── loader.py             # multi-source data loading
│   │   ├── schema_validator.py   # structural + type validation
│   │   └── deduplicator.py       # semantic deduplication
│   │
│   ├── processing/
│   │   ├── bert_encoder.py       # BERT embedding wrapper
│   │   ├── anomaly_detector.py   # 3-layer detection cascade
│   │   ├── classifier.py         # text classification
│   │   └── entity_extractor.py   # NER + relation extraction
│   │
│   ├── storage/
│   │   ├── mongo_client.py       # document storage
│   │   └── metrics_store.py      # PostgreSQL metrics
│   │
│   ├── api/
│   │   ├── main.py               # FastAPI + WebSocket
│   │   └── routes.py
│   │
│   └── dashboard/                # React + D3.js frontend
│       ├── src/
│       └── public/
│
├── notebooks/
│   ├── 01_eda.ipynb              # exploratory data analysis
│   ├── 02_ablation_preprocessing.ipynb
│   └── 03_bert_finetuning.ipynb  # domain adaptation
│
├── tests/
├── docker-compose.yml
├── requirements.txt
└── README.md

Quickstart

git clone https://github.com/CHHemant/nlp-data-pipeline
cd nlp-data-pipeline

# backend
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
docker-compose up -d mongodb postgres
uvicorn src.api.main:app --reload

# dashboard
cd src/dashboard
npm install && npm start

Limitations

  • BERT latency — at 20k records/day the semantic layer runs fine. At 100k+/day you'd need to replace BERT with a lighter model (DistilBERT, or a fine-tuned smaller encoder) or move to async batch processing.
  • Cold start — the anomaly detector needs ~500 clean records to build a reliable baseline. First run on a new data source requires manual review.
  • English only — BERT model is bert-base-uncased. Works poorly on multilingual or code-mixed text.
  • MongoDB schema — currently schemaless by design for flexibility. In regulated environments you'd want strict schema enforcement at the DB level too.

About

Built by Hemant Chilkuri — B.Tech AI, G.H. Raisoni University (2024–2028).

hemant_189@outlook.com · linkedin.com/in/hemantchilkuri · github.com/CHHemant

About

No description, website, or topics provided.

Resources

Stars

0 stars

Watchers

0 watching

Forks

Releases

No releases published

Packages

 
 
 

Contributors