NLP Evaluation & Deployment

# ─── 1. Comprehensive NLP Evaluation ─────────────────────────
from sklearn.metrics import (classification_report, confusion_matrix,
                              roc_auc_score, average_precision_score)
import numpy as np

# Multi-class classification evaluation
y_true = [0, 1, 2, 0, 1, 2, 0, 1, 2, 1]
y_pred = [0, 1, 2, 0, 2, 2, 1, 1, 0, 1]
y_prob = np.array([
    [0.8, 0.1, 0.1], [0.1, 0.8, 0.1], [0.1, 0.1, 0.8],
    [0.7, 0.2, 0.1], [0.2, 0.3, 0.5], [0.1, 0.1, 0.8],
    [0.3, 0.5, 0.2], [0.1, 0.8, 0.1], [0.4, 0.2, 0.4],
    [0.1, 0.7, 0.2],
])

print("Classification Report:")
print(classification_report(y_true, y_pred,
      target_names=["positive", "negative", "neutral"]))

# AUC-ROC (one-vs-rest)
auc_ovr = roc_auc_score(y_true, y_prob, multi_class='ovr', average='weighted')
print(f"Weighted AUC-ROC (OvR): {auc_ovr:.4f}")

# ─── 2. NER Span-Level F1 ─────────────────────────────────────
# seqeval is the standard library for NER evaluation
# pip install seqeval
from seqeval.metrics import f1_score, precision_score, recall_score, classification_report as seq_report

y_true_ner = [
    ["B-PER", "I-PER", "O", "B-ORG", "O"],
    ["O", "B-GPE", "O", "O"],
]
y_pred_ner = [
    ["B-PER", "I-PER", "O", "B-ORG", "O"],  # perfect
    ["O", "B-GPE", "I-GPE", "O"],             # wrong: extra I-GPE
]

print("\nNER Span-level Evaluation:")
print(seq_report(y_true_ner, y_pred_ner))
print(f"  F1:        {f1_score(y_true_ner, y_pred_ner):.4f}")
print(f"  Precision: {precision_score(y_true_ner, y_pred_ner):.4f}")
print(f"  Recall:    {recall_score(y_true_ner, y_pred_ner):.4f}")

# ─── 3. BLEU, ROUGE, and BERTScore ───────────────────────────
import sacrebleu
from rouge_score import rouge_scorer

# Machine Translation
mt_refs = [["The cat sat on the mat.", "A cat was sitting on the mat."]]
mt_hyps = ["The cat is sitting on the mat."]
bleu = sacrebleu.corpus_bleu(mt_hyps, mt_refs)
chrf = sacrebleu.corpus_chrf(mt_hyps, mt_refs)
print(f"\nMT Evaluation:")
print(f"  BLEU: {bleu.score:.2f}, chrF: {chrf.score:.2f}")

# Summarization
rouge = rouge_scorer.RougeScorer(['rouge1', 'rouge2', 'rougeL'], use_stemmer=True)
ref_summary = "AI has transformed many industries including healthcare and finance."
hyp_summary = "Artificial intelligence has changed healthcare, finance, and other sectors."
scores = rouge.score(ref_summary, hyp_summary)
print(f"\nSummarization ROUGE:")
for k, v in scores.items():
    print(f"  {k}: P={v.precision:.3f} R={v.recall:.3f} F={v.fmeasure:.3f}")

# BERTScore (semantic similarity, better human correlation)
# pip install bert-score
from bert_score import score as bert_score
P, R, F1 = bert_score([hyp_summary], [ref_summary], lang="en", verbose=False)
print(f"  BERTScore F1: {F1.mean():.4f}")

# ─── 4. FastAPI NLP Serving ───────────────────────────────────
# Save as: nlp_api.py
# pip install fastapi uvicorn transformers torch
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from transformers import pipeline
import time

app = FastAPI(title="NLP Inference API", version="1.0.0")

# Load models once at startup
sentiment_pipe = pipeline("sentiment-analysis",
                           model="distilbert-base-uncased-finetuned-sst-2-english")
ner_pipe = pipeline("ner", model="dbmdz/bert-large-cased-finetuned-conll03-english",
                    aggregation_strategy="simple")

class TextRequest(BaseModel):
    text: str
    model: str = "sentiment"

class PredictionResponse(BaseModel):
    text: str
    predictions: list
    latency_ms: float
    model_version: str = "v1.0"

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: TextRequest):
    start = time.time()
    if len(request.text) > 10000:
        raise HTTPException(400, "Text exceeds maximum length of 10000 characters")

    if request.model == "sentiment":
        result = sentiment_pipe(request.text[:512])
    elif request.model == "ner":
        result = ner_pipe(request.text[:512])
    else:
        raise HTTPException(400, f"Unknown model: {request.model}")

    latency = (time.time() - start) * 1000
    return PredictionResponse(
        text=request.text[:100],
        predictions=result,
        latency_ms=round(latency, 2),
    )

@app.get("/health")
async def health():
    return {"status": "ok", "models": ["sentiment", "ner"]}

# Run: uvicorn nlp_api:app --host 0.0.0.0 --port 8000

# ─── 5. Monitoring: Data Drift Detection ──────────────────────
from scipy.stats import ks_2samp
from transformers import AutoTokenizer

def compute_text_stats(texts):
    """Compute distribution features for drift detection."""
    lengths = [len(t.split()) for t in texts]
    return {
        "mean_length": np.mean(lengths),
        "std_length": np.std(lengths),
        "max_length": np.max(lengths),
        "lengths": lengths,
    }

# Reference distribution (training data)
train_texts = ["This is great!", "Terrible product.", "Average quality.",
               "Works as expected.", "Would recommend.", "Disappointed."] * 100

# Production distribution (potentially drifted)
prod_texts  = ["Excelente producto!!!", "Sehr gut.", "Fantastique!", "훌륭해요",
               "Amazing 10/10 best ever", "Complete garbage waste of money"] * 100

train_stats = compute_text_stats(train_texts)
prod_stats  = compute_text_stats(prod_texts)

# Kolmogorov-Smirnov test for distribution shift
ks_stat, p_value = ks_2samp(train_stats["lengths"], prod_stats["lengths"])
print(f"\nDrift Detection (text length distribution):")
print(f"  KS statistic: {ks_stat:.4f}, p-value: {p_value:.4f}")
print(f"  Drift detected: {p_value < 0.05}")