MLOps For NLP Systems

1. Log parameters, metrics, and artifacts (confusion matrices, prompts, tokenization stats).
2. Tag runs with dataset, commit SHA, and data/prompt version.
3. Establish a naming convention for runs and models.

1. Adopt a clear dataset schema and a versioning mechanism (e.g., SemVer like v1.2.0).
2. Store prompt templates as files with explicit version headers.
3. Link data and prompt versions to each training/eval run.

1. Push trained models to a registry with metadata (task, language, license, tags).
2. Define stages: Staging, Shadow, Production, Archived.
3. Automate stage transitions through evaluation gates.

1. Automate unit tests (tokenizer, pre/post-processing), data schema checks, and linting.
2. Run offline evaluations on key slices (short texts, long texts, domain-specific).
3. Deploy using blue/green or shadow traffic strategies.

1. Track latency, error rates, and throughput, plus NLP-specific quality metrics.
2. Monitor drift via embeddings and label distributions.
3. Define escalation paths, rollback steps, and communication templates.

import time, json, random
from sklearn.metrics import f1_score, classification_report

# Pseudo-train example
def train(params):
    time.sleep(0.2)
    # pretend predictions
    y_true = [0,1,1,0,1,0,1,1,0,0]
    y_pred = [0,1,0,0,1,0,1,1,0,1]
    f1 = f1_score(y_true, y_pred, average='macro')
    report = classification_report(y_true, y_pred, output_dict=True)
    return f1, report, {"model.bin":"/tmp/model.bin"}

# Minimal MLflow-like logging stub (replace with real MLflow/W&B)
class Run:
    def log_params(self, d):
        print("PARAMS:", d)
    def log_metrics(self, d):
        print("METRICS:", d)
    def log_artifact(self, path, name):
        print(f"ARTIFACT: {name} -> {path}")
    def set_tags(self, d):
        print("TAGS:", d)

run = Run()
params = {
    "model":"bert-base-uncased",
    "max_len":128,
    "lr":2e-5,
    "epochs":3
}
run.log_params(params)
run.set_tags({
    "task":"text-classification",
    "dataset_version":"reviews-v1.1.0",
    "prompt_version":"n/a",
    "commit":"abc1234"
})
f1, report, arts = train(params)
run.log_metrics({"val_f1_macro": f1, "latency_ms_p50": random.randint(20,40)})
run.log_artifact("/tmp/confusion_matrix.png", name="confusion_matrix.png")
run.log_artifact(arts["model.bin"], name="model.bin")
print(json.dumps(report, indent=2))

Key idea: always log params, metrics, tags, and artifacts together so a run is reproducible.

# Pseudocode representing a registry API
class Registry:
    def register(self, name, path, metadata):
        print(f"Registered {name} with metadata {metadata}")
        return {"name":name, "version": 1}
    def promote(self, name, version, stage):
        print(f"Promoted {name}:{version} to {stage}")

reg = Registry()
meta = {
  "task":"text-classification",
  "dataset_version":"reviews-v1.1.0",
  "val_f1_macro":0.82,
  "owner":"nlp-platform",
  "license":"internal"
}
model_ref = reg.register("sentiment-bert", "/tmp/model.bin", meta)
reg.promote(model_ref["name"], model_ref["version"], stage="Staging")
# After passing automated gates in CI/CD
reg.promote(model_ref["name"], model_ref["version"], stage="Production")

Always attach evaluation and data lineage metadata before promotion.

# dataset.yaml
name: product-reviews
version: 1.2.0
schema:
  text: string
  label: {enum: [negative, neutral, positive]}
quality_checks:
  - min_rows: 5000
  - max_null_text_pct: 0.5
splits:
  train: 0.8
  valid: 0.1
  test: 0.1

# prompt_v0.3.1.txt
version: 0.3.1
goal: Classify sentiment as negative, neutral, or positive.
rules:
  - Be concise.
  - Output one label only.
format: {"label": "negative|neutral|positive"}

Store these files in version control. Reference their versions in experiment tags and model metadata.

# ci-cd.yaml (conceptual)
name: nlp-ci-cd
on: [push]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Setup Python
        run: |
          python -m venv .venv
          . .venv/bin/activate
          pip install -r requirements.txt
      - name: Lint & unit tests
        run: |
          . .venv/bin/activate
          flake8 .
          pytest -q
      - name: Schema check
        run: |
          python tools/check_schema.py data/dataset.yaml
      - name: Offline evaluation gate
        run: |
          python tools/eval_gate.py --min_f1_macro 0.80 --max_latency_ms_p95 150
      - name: Package and push model
        if: ${{ success() }}
        run: |
          python tools/pack_model.py
          python tools/push_to_registry.py --stage Staging

Fail fast if checks do not meet thresholds. Only package and push after gates pass.

import sys, json, time
from statistics import mean

# Fake metrics for demo
metrics = {
  "val_f1_macro": 0.81,
  "latency_ms_p95": 140,
  "slice_f1": {
    "short_text": 0.83,
    "long_text": 0.78
  }
}

MIN_F1 = 0.80
MIN_LONG_TEXT_F1 = 0.75
MAX_P95_LAT_MS = 150

ok = True
reasons = []
if metrics["val_f1_macro"] < MIN_F1:
    ok = False; reasons.append("F1 macro below threshold")
if metrics["latency_ms_p95"] > MAX_P95_LAT_MS:
    ok = False; reasons.append("Latency p95 above threshold")
if metrics["slice_f1"]["long_text"] < MIN_LONG_TEXT_F1:
    ok = False; reasons.append("Long text slice under threshold")

print(json.dumps(metrics, indent=2))
if not ok:
    print("GATE: FAIL", reasons)
    sys.exit(1)
print("GATE: PASS")

Include slice-based checks to avoid regressions hidden by overall metrics.

import numpy as np
from numpy.linalg import norm

# Suppose we store reference mean embedding vector from last good window
ref_mean = np.array([0.12, -0.03, 0.22, 0.01])

# Current window embeddings mean (mock)
cur_mean = np.array([0.20, -0.05, 0.18, -0.02])

cosine_sim = np.dot(ref_mean, cur_mean) / (norm(ref_mean)*norm(cur_mean))
cosine_dist = 1 - cosine_sim
THRESHOLD = 0.08

if cosine_dist > THRESHOLD:
    print(f"ALERT: Embedding drift detected (cosine distance={cosine_dist:.3f})")
else:
    print("Embedding drift OK")

Track drift per domain slice (e.g., language, product category) to localize issues.

• Model: any small text classifier or a prompt-based classifier.
• Data: labeled reviews with a fixed schema and version file.
• CI/CD: tests, schema check, offline evaluation gates, packaging, and deployment to a staging environment.
• Registry: register model with metadata; promote Staging to Production only if gates pass.
• Monitoring: log latency, error rate, embedding drift; weekly drift report.
• Runbook: document rollback steps and on-call escalation.

• Every run has params, metrics, artifacts, and tags (dataset_version, prompt_version, commit).
• Automated gates enforce F1 ≥ target and p95 latency ≤ budget, including slice checks.
• One-click rollback or automated rollback on gate failure during canary.
• Weekly drift summary produced and stored as an artifact.
• Incident response playbook exists and is accessible to the team.