diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000000000000000000000000000000000000..029fa3974dcf97fecc247121f49b8479f83df2cc --- /dev/null +++ b/Dockerfile @@ -0,0 +1,32 @@ +FROM python:3.10-slim + +WORKDIR /app + +RUN apt-get update && apt-get install -y \ + curl \ + git \ + && rm -rf /var/lib/apt/lists/* + +RUN pip install uv --no-cache-dir + +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY src/ src/ +COPY api/ api/ +COPY app/ app/ +COPY configs/ configs/ +COPY app.py . + +RUN mkdir -p artifacts/models/distilbert \ + artifacts/vectorizers \ + artifacts/monitoring \ + data/raw + +ENV PYTHONPATH=/app +ENV PYTHONDONTWRITEBYTECODE=1 +ENV PYTHONUNBUFFERED=1 + +EXPOSE 7860 + +CMD ["python", "app.py"] diff --git a/README.md b/README.md index 4c0ec78a87136d97d8aab78ea4323034bb20c621..f5149c03b9120da689f93ef9ee8e7ffe0c32a913 100644 --- a/README.md +++ b/README.md @@ -1,10 +1,20 @@ --- title: Intent Classifier -emoji: 🏃 +emoji: 🧠 colorFrom: blue -colorTo: purple +colorTo: indigo sdk: docker +app_port: 7860 pinned: false --- -Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference +# Intent Classifier + +End-to-end intent classification system with full MLOps pipeline. + +Classifies user queries into 151 intent categories using CLINC150 dataset. +Compares 6 model architectures: LogReg, SVM, TextCNN, RNN, LSTM, DistilBERT. + +## Source + +[github.com/princ0301/intent-classifier](https://github.com/princ0301/intent-classifier) diff --git a/api/__init__.py b/api/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/api/__pycache__/__init__.cpython-312.pyc b/api/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..c51614c1fd3ee9c7d639c60aaa4010371d83d286 Binary files /dev/null and b/api/__pycache__/__init__.cpython-312.pyc differ diff --git a/api/__pycache__/main.cpython-312.pyc b/api/__pycache__/main.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..30195d3af729ce8aac18a36001739b916ca6c4df Binary files /dev/null and b/api/__pycache__/main.cpython-312.pyc differ diff --git a/api/__pycache__/schemas.cpython-312.pyc b/api/__pycache__/schemas.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..015916ed7213a71c53928d8a0663744f0323b01b Binary files /dev/null and b/api/__pycache__/schemas.cpython-312.pyc differ diff --git a/api/main.py b/api/main.py new file mode 100644 index 0000000000000000000000000000000000000000..1d2a1279f72f5fbb7cb6f9bb53a6f8d0358d5aac --- /dev/null +++ b/api/main.py @@ -0,0 +1,54 @@ +from contextlib import asynccontextmanager + +from fastapi import FastAPI, Request +from fastapi.middleware.cors import CORSMiddleware + +from api.routes import models, monitoring, predict +from api.schemas import HealthResponse +from src.serving.ab_router import ABRouter +from src.serving.predictor import Predictor + + +@asynccontextmanager +async def lifespan(app: FastAPI): + print("loading predictors...") + app.state.predictors = { + "classical": Predictor("classical"), + "svm": Predictor("svm"), + "transformer": Predictor("transformer"), + } + print("loading AB router...") + app.state.ab_router = ABRouter() + print("startup complete") + + yield + + app.state.predictors.clear() + + +app = FastAPI( + title="Intent Classifier API", + description="End-to-end intent classification with classical ML, neural networks, and fine-tuned transformers", + version="1.0.0", + lifespan=lifespan, +) + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +app.include_router(predict.router, tags=["predict"]) +app.include_router(models.router, tags=["models"]) +app.include_router(monitoring.router, tags=["monitoring"]) + + +@app.get("/health", response_model=HealthResponse) +def health(request: Request) -> HealthResponse: + return HealthResponse( + status="ok", + models_loaded=list(request.app.state.predictors.keys()), + ) diff --git a/api/routes/__init__.py b/api/routes/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/api/routes/__pycache__/__init__.cpython-312.pyc b/api/routes/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..615bcde6d368086d62d996659c42e6975a1a984b Binary files /dev/null and b/api/routes/__pycache__/__init__.cpython-312.pyc differ diff --git a/api/routes/__pycache__/models.cpython-312.pyc b/api/routes/__pycache__/models.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..c37b10cc48d7740dff1242c82efb5ef5304021d2 Binary files /dev/null and b/api/routes/__pycache__/models.cpython-312.pyc differ diff --git a/api/routes/__pycache__/monitoring.cpython-312.pyc b/api/routes/__pycache__/monitoring.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..a88175e980083a9814fa8c849a629742c9a2f7a5 Binary files /dev/null and b/api/routes/__pycache__/monitoring.cpython-312.pyc differ diff --git a/api/routes/__pycache__/predict.cpython-312.pyc b/api/routes/__pycache__/predict.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..3f2557c82a7d4313d6f48dd9a4c48c24036f1776 Binary files /dev/null and b/api/routes/__pycache__/predict.cpython-312.pyc differ diff --git a/api/routes/models.py b/api/routes/models.py new file mode 100644 index 0000000000000000000000000000000000000000..714d39af57ad3c37b3e476fd053b747b3a4f4409 --- /dev/null +++ b/api/routes/models.py @@ -0,0 +1,22 @@ +from fastapi import APIRouter, Request + +from api.schemas import ModelInfo, ModelsResponse + +router = APIRouter() + +REGISTERED_MODELS = [ + {"name": "intent-classifier-logreg", "alias": "challenger"}, + {"name": "intent-classifier-svm", "alias": "challenger"}, + {"name": "intent-classifier-distilbert", "alias": "champion"}, +] + + +@router.get("/models", response_model=ModelsResponse) +def list_models() -> ModelsResponse: + return ModelsResponse(models=[ModelInfo(name=m["name"], alias=m["alias"]) for m in REGISTERED_MODELS]) + + +@router.get("/models/loaded") +def list_loaded_models(request: Request) -> dict: + predictors = request.app.state.predictors + return {"loaded_models": list(predictors.keys())} diff --git a/api/routes/monitoring.py b/api/routes/monitoring.py new file mode 100644 index 0000000000000000000000000000000000000000..5f7c80fb130a2c4133f5c9f898c8a16371ded379 --- /dev/null +++ b/api/routes/monitoring.py @@ -0,0 +1,47 @@ +from fastapi import APIRouter, HTTPException, Request + +from api.schemas import ABStatsResponse, DriftResponse +from src.monitoring.drift import ( + get_confidence_drift, + get_drift_summary, + get_oos_rate_drift, + load_current_data, + load_reference_data, + run_drift_report, +) + +router = APIRouter() + + +@router.get("/monitoring/drift", response_model=DriftResponse) +def get_drift() -> DriftResponse: + try: + reference_df = load_reference_data() + current_df = load_current_data() + except FileNotFoundError as e: + raise HTTPException(status_code=404, detail=str(e)) + + snapshot = run_drift_report(reference_df, current_df) + drift_summary = get_drift_summary(snapshot) + confidence_drift = get_confidence_drift(reference_df, current_df) + oos_rate_drift = get_oos_rate_drift(reference_df, current_df) + + return DriftResponse( + drift_summary=drift_summary, + confidence_drift=confidence_drift, + oos_rate_drift=oos_rate_drift, + ) + + +@router.get("/monitoring/ab-stats", response_model=ABStatsResponse) +def get_ab_stats(request: Request) -> ABStatsResponse: + ab_router = request.app.state.ab_router + stats = ab_router.get_stats() + return ABStatsResponse(**stats) + + +@router.post("/monitoring/reset") +def reset_monitoring(request: Request) -> dict: + ab_router = request.app.state.ab_router + ab_router.reset_stats() + return {"status": "reset"} diff --git a/api/routes/predict.py b/api/routes/predict.py new file mode 100644 index 0000000000000000000000000000000000000000..62d9fa9a90f81871970a7cfec7ef36a5e0e8b875 --- /dev/null +++ b/api/routes/predict.py @@ -0,0 +1,78 @@ +from fastapi import APIRouter, HTTPException, Request + +from api.schemas import ( + BatchRequest, + BatchResponse, + PredictRequest, + PredictResponse, + TopIntent, +) +from src.monitoring.drift import append_current_data + +router = APIRouter() + + +def _to_predict_response(result, ab_variant: str | None = None) -> PredictResponse: + return PredictResponse( + intent=result.intent, + confidence=result.confidence, + top5=[TopIntent(**item) for item in result.top5], + latency_ms=result.latency_ms, + is_oos=result.is_oos, + model_used=result.model_used, + ab_variant=ab_variant, + ) + + +@router.post("/predict", response_model=PredictResponse) +def predict(payload: PredictRequest, request: Request) -> PredictResponse: + predictors = request.app.state.predictors + ab_router = request.app.state.ab_router + + if payload.model_type: + if payload.model_type not in predictors: + raise HTTPException(status_code=400, detail=f"unknown model_type: {payload.model_type}") + result = predictors[payload.model_type].predict(payload.text) + response = _to_predict_response(result) + else: + result_dict = ab_router.predict(payload.text) + response = PredictResponse( + intent=result_dict["intent"], + confidence=result_dict["confidence"], + top5=[TopIntent(**item) for item in result_dict["top5"]], + latency_ms=result_dict["latency_ms"], + is_oos=result_dict["is_oos"], + model_used=result_dict["model_used"], + ab_variant=result_dict["ab_variant"], + ) + + append_current_data( + texts=[payload.text], + predictions=[response.intent], + confidences=[response.confidence], + ) + + return response + + +@router.post("/predict/batch", response_model=BatchResponse) +def predict_batch(payload: BatchRequest, request: Request) -> BatchResponse: + predictors = request.app.state.predictors + + model_type = payload.model_type or "transformer" + if model_type not in predictors: + raise HTTPException(status_code=400, detail=f"unknown model_type: {model_type}") + + predictor = predictors[model_type] + results = predictor.predict_batch(payload.texts) + responses = [_to_predict_response(r) for r in results] + + total_latency = sum(r.latency_ms for r in responses) + + append_current_data( + texts=payload.texts, + predictions=[r.intent for r in responses], + confidences=[r.confidence for r in responses], + ) + + return BatchResponse(predictions=responses, total_latency_ms=round(total_latency, 3)) diff --git a/api/schemas.py b/api/schemas.py new file mode 100644 index 0000000000000000000000000000000000000000..815aa023bdbe4f5948bd8080bdd87f7602d69765 --- /dev/null +++ b/api/schemas.py @@ -0,0 +1,58 @@ +from pydantic import BaseModel, Field + + +class PredictRequest(BaseModel): + text: str = Field(..., min_length=1, max_length=500) + model_type: str | None = None + + +class TopIntent(BaseModel): + intent: str + confidence: float + + +class PredictResponse(BaseModel): + intent: str + confidence: float + top5: list[TopIntent] + latency_ms: float + is_oos: bool + model_used: str + ab_variant: str | None = None + + +class BatchRequest(BaseModel): + texts: list[str] = Field(..., min_length=1, max_length=100) + model_type: str | None = None + + +class BatchResponse(BaseModel): + predictions: list[PredictResponse] + total_latency_ms: float + + +class ModelInfo(BaseModel): + name: str + alias: str + version: str | None = None + + +class ModelsResponse(BaseModel): + models: list[ModelInfo] + + +class HealthResponse(BaseModel): + status: str + models_loaded: list[str] + + +class DriftResponse(BaseModel): + drift_summary: dict + confidence_drift: dict + oos_rate_drift: dict + + +class ABStatsResponse(BaseModel): + model_a: dict + model_b: dict + split: float diff --git a/app.py b/app.py new file mode 100644 index 0000000000000000000000000000000000000000..c53d3f4ed7b41e530b7a1bc2eb7f532bc750fcdc --- /dev/null +++ b/app.py @@ -0,0 +1,95 @@ +import os +import subprocess +import threading +import time + + +def start_api(): + subprocess.run( + [ + "uvicorn", + "api.main:app", + "--host", + "0.0.0.0", + "--port", + "8000", + ], + check=False, + ) + + +def wait_for_api(timeout: int = 120) -> bool: + import requests + + start = time.time() + while time.time() - start < timeout: + try: + requests.get("http://localhost:8000/health", timeout=2) + return True + except Exception: + time.sleep(2) + return False + + +def download_models(): + from src.storage.s3 import download_artifact + + print("downloading models from S3...") + + os.makedirs("artifacts/models/distilbert", exist_ok=True) + os.makedirs("artifacts/vectorizers", exist_ok=True) + + download_artifact("classical/logreg.pkl", "artifacts/models/logreg.pkl") + download_artifact("classical/svm.pkl", "artifacts/models/svm.pkl") + download_artifact("classical/tfidf.pkl", "artifacts/vectorizers/tfidf.pkl") + download_artifact("neural/vocab.pkl", "artifacts/models/vocab.pkl") + download_artifact("transformer/distilbert/config.json", "artifacts/models/distilbert/config.json") + download_artifact("transformer/distilbert/model.safetensors", "artifacts/models/distilbert/model.safetensors") + download_artifact("transformer/distilbert/tokenizer.json", "artifacts/models/distilbert/tokenizer.json") + download_artifact("transformer/distilbert/tokenizer_config.json", "artifacts/models/distilbert/tokenizer_config.json") + download_artifact("transformer/distilbert/special_tokens_map.json", "artifacts/models/distilbert/special_tokens_map.json") + download_artifact("transformer/distilbert/vocab.txt", "artifacts/models/distilbert/vocab.txt") + + print("models downloaded") + + +def generate_label_map(): + from src.data.loader import load_clinc150, save_splits + from src.data.preprocessor import preprocess + + label_map_path = "artifacts/label_map.json" + if os.path.exists(label_map_path): + print("label map already exists") + return + + print("generating label map...") + splits = load_clinc150("plus") + save_splits(splits) + preprocess(splits) + print("label map generated") + + +if __name__ == "__main__": + download_models() + generate_label_map() + + api_thread = threading.Thread(target=start_api, daemon=True) + api_thread.start() + + print("waiting for API to start...") + if wait_for_api(): + print("API ready") + else: + print("API startup timed out — continuing anyway") + + os.execv( + "/usr/bin/streamlit", + [ + "streamlit", + "run", + "app/streamlit_app.py", + "--server.port=7860", + "--server.address=0.0.0.0", + "--server.headless=true", + ], + ) \ No newline at end of file diff --git a/app/__pycache__/api_client.cpython-312.pyc b/app/__pycache__/api_client.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..7970fd66e3b19ec3f840bf6c1953abd8e8993c7c Binary files /dev/null and b/app/__pycache__/api_client.cpython-312.pyc differ diff --git a/app/api_client.py b/app/api_client.py new file mode 100644 index 0000000000000000000000000000000000000000..79a7a461aa44ec2eaa07b26bf12acfa416af79c7 --- /dev/null +++ b/app/api_client.py @@ -0,0 +1,62 @@ +import os + +import requests +import streamlit as st + +API_BASE_URL = os.getenv("API_BASE_URL", "http://localhost:8000") + + +def api_predict(text: str, model_type: str | None = None) -> dict: + payload = {"text": text} + if model_type: + payload["model_type"] = model_type + response = requests.post(f"{API_BASE_URL}/predict", json=payload, timeout=30) + response.raise_for_status() + return response.json() + + +def api_health() -> dict: + response = requests.get(f"{API_BASE_URL}/health", timeout=10) + response.raise_for_status() + return response.json() + + +def api_models() -> dict: + response = requests.get(f"{API_BASE_URL}/models", timeout=10) + response.raise_for_status() + return response.json() + + +def api_ab_stats() -> dict: + response = requests.get(f"{API_BASE_URL}/monitoring/ab-stats", timeout=10) + response.raise_for_status() + return response.json() + + +def api_drift() -> dict: + response = requests.get(f"{API_BASE_URL}/monitoring/drift", timeout=30) + response.raise_for_status() + return response.json() + + +def api_reset_monitoring() -> dict: + response = requests.post(f"{API_BASE_URL}/monitoring/reset", timeout=10) + response.raise_for_status() + return response.json() + + +def check_api_connection() -> bool: + try: + api_health() + return True + except requests.exceptions.RequestException: + return False + + +def require_api(): + if not check_api_connection(): + st.error( + f"Cannot connect to API at {API_BASE_URL}. " + "Start it with: `uv run uvicorn api.main:app --reload --port 8000`" + ) + st.stop() diff --git a/app/pages/compare.py b/app/pages/compare.py new file mode 100644 index 0000000000000000000000000000000000000000..cfc25a724192a8d3004fb113d2f43c6fcb933eff --- /dev/null +++ b/app/pages/compare.py @@ -0,0 +1,100 @@ +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +import pandas as pd +import streamlit as st + +from app.api_client import api_predict, require_api + +require_api() + +st.title("Compare Models") +st.caption("Side-by-side comparison of all trained models on CLINC150 test set.") + +RESULTS = { + "logreg": { + "accuracy": 0.8058, + "macro_f1": 0.8416, + "weighted_f1": 0.7985, + "latency_p50_ms": 0.12, + }, + "svm": { + "accuracy": 0.8002, + "macro_f1": 0.8474, + "weighted_f1": 0.7795, + "latency_p50_ms": 0.14, + }, + "textcnn": { + "accuracy": 0.7740, + "macro_f1": 0.8244, + "weighted_f1": 0.7646, + "latency_p50_ms": 97.6, + }, + "rnn": { + "accuracy": 0.1818, + "macro_f1": 0.0020, + "weighted_f1": 0.0560, + "latency_p50_ms": 124.5, + }, + "lstm": { + "accuracy": 0.7669, + "macro_f1": 0.8113, + "weighted_f1": 0.7555, + "latency_p50_ms": 249.2, + }, + "distilbert": { + "accuracy": 0.8765, + "macro_f1": 0.9019, + "weighted_f1": 0.8722, + "latency_p50_ms": 9.6, + }, +} + +df = pd.DataFrame(RESULTS).T +df.index.name = "model" + +st.subheader("Metrics Table") +st.dataframe( + df.style.format( + { + "accuracy": "{:.2%}", + "macro_f1": "{:.2%}", + "weighted_f1": "{:.2%}", + "latency_p50_ms": "{:.2f} ms", + } + ).background_gradient(subset=["accuracy", "macro_f1", "weighted_f1"], cmap="Greens"), + use_container_width=True, +) + +st.subheader("Accuracy & F1 Comparison") +chart_df = df[["accuracy", "macro_f1", "weighted_f1"]] +st.bar_chart(chart_df) + +st.subheader("Latency Comparison") +st.bar_chart(df[["latency_p50_ms"]]) + +st.divider() +st.subheader("Side-by-Side Prediction") +st.caption("Run the same input through multiple models at once.") + +text = st.text_input("Enter a query to compare", placeholder="e.g. cancel my hotel reservation") +selected_models = st.multiselect( + "Models to compare", + options=["classical", "svm", "transformer"], + default=["classical", "transformer"], +) + +if st.button("Compare", type="primary", disabled=not (text and selected_models)): + cols = st.columns(len(selected_models)) + for col, model_type in zip(cols, selected_models): + with col: + with st.spinner(f"running {model_type}..."): + result = api_predict(text, model_type) + st.markdown(f"**{model_type}**") + st.metric("Intent", result["intent"]) + st.metric("Confidence", f"{result['confidence']:.2%}") + st.metric("Latency", f"{result['latency_ms']:.1f} ms") + if result["is_oos"]: + st.warning("flagged OOS") diff --git a/app/pages/monitoring.py b/app/pages/monitoring.py new file mode 100644 index 0000000000000000000000000000000000000000..f8d4e8892f82772d0c1b1c82df5158f8049831c8 --- /dev/null +++ b/app/pages/monitoring.py @@ -0,0 +1,90 @@ +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +import streamlit as st + +from app.api_client import api_ab_stats, api_drift, api_reset_monitoring, require_api + +require_api() + +st.title("Monitoring Dashboard") +st.caption("Live A/B test stats and data drift detection.") + +st.subheader("A/B Test Stats") + +if st.button("Refresh stats"): + st.rerun() + +try: + stats = api_ab_stats() + + col1, col2 = st.columns(2) + + with col1: + st.markdown(f"### Model A — `{stats['model_a']['model_name']}`") + st.metric("Requests", stats["model_a"]["request_count"]) + st.metric("Avg Latency", f"{stats['model_a']['avg_latency_ms']:.2f} ms") + st.metric("Avg Confidence", f"{stats['model_a']['avg_confidence']:.2%}") + st.metric("OOS Rate", f"{stats['model_a']['oos_rate']:.2%}") + + with col2: + st.markdown(f"### Model B — `{stats['model_b']['model_name']}`") + st.metric("Requests", stats["model_b"]["request_count"]) + st.metric("Avg Latency", f"{stats['model_b']['avg_latency_ms']:.2f} ms") + st.metric("Avg Confidence", f"{stats['model_b']['avg_confidence']:.2%}") + st.metric("OOS Rate", f"{stats['model_b']['oos_rate']:.2%}") + + st.caption(f"Configured split: {stats['split']:.0%} to model B") + +except Exception as e: + st.warning(f"No A/B stats yet. Make some predictions on the Predict page first. ({e})") + +if st.button("Reset A/B stats"): + api_reset_monitoring() + st.success("Stats reset.") + st.rerun() + +st.divider() +st.subheader("Data Drift Report") +st.caption("Compares reference traffic (initial baseline) against recent production traffic.") + +if st.button("Run drift check"): + try: + with st.spinner("computing drift..."): + drift = api_drift() + + col1, col2, col3 = st.columns(3) + col1.metric( + "Drifted Columns", + int(drift["drift_summary"].get("drifted_columns_count", 0)), + ) + col2.metric( + "Confidence Drop", + f"{drift['confidence_drift']['confidence_drop']:.2%}", + delta=f"{-drift['confidence_drift']['confidence_drop']:.2%}", + ) + col3.metric( + "OOS Rate Increase", + f"{drift['oos_rate_drift']['oos_rate_increase']:.2%}", + ) + + if drift["confidence_drift"]["is_degraded"]: + st.error("Confidence has degraded significantly vs reference traffic.") + else: + st.success("Confidence is stable vs reference traffic.") + + if drift["oos_rate_drift"]["is_anomalous"]: + st.error("OOS rate has increased anomalously — possible topic drift in incoming queries.") + else: + st.success("OOS rate is within normal range.") + + with st.expander("Raw drift details"): + st.json(drift) + + except Exception as e: + st.warning( + "No reference/current data available yet. Run `verify_phase11.py` first to " + f"generate baseline monitoring data. ({e})" + ) diff --git a/app/pages/predict.py b/app/pages/predict.py new file mode 100644 index 0000000000000000000000000000000000000000..e11d3e6b5b7b3db278fa310650ed54d557540ff8 --- /dev/null +++ b/app/pages/predict.py @@ -0,0 +1,67 @@ +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +import pandas as pd +import streamlit as st + +from app.api_client import api_predict, require_api + +require_api() + +st.title("Predict Intent") +st.caption("Test the intent classifier live, switching between models or letting the A/B router decide.") + +col1, col2 = st.columns([3, 1]) + +with col1: + text = st.text_input("Enter a query", placeholder="e.g. what is my account balance") + +with col2: + model_choice = st.selectbox( + "Model", + options=["A/B Router", "Classical (LogReg)", "SVM", "Transformer (DistilBERT)"], + ) + +model_map = { + "A/B Router": None, + "Classical (LogReg)": "classical", + "SVM": "svm", + "Transformer (DistilBERT)": "transformer", +} + +if st.button("Predict", type="primary", disabled=not text): + with st.spinner("predicting..."): + result = api_predict(text, model_map[model_choice]) + + col_a, col_b, col_c = st.columns(3) + col_a.metric("Intent", result["intent"]) + col_b.metric("Confidence", f"{result['confidence']:.2%}") + col_c.metric("Latency", f"{result['latency_ms']:.1f} ms") + + if result["is_oos"]: + st.warning("This query was flagged as out-of-scope (low confidence).") + + if result.get("ab_variant"): + st.info(f"Served by A/B variant **{result['ab_variant']}** using model `{result['model_used']}`") + else: + st.info(f"Served by model `{result['model_used']}`") + + st.subheader("Top 5 Predictions") + df = pd.DataFrame(result["top5"]) + df["confidence"] = df["confidence"].astype(float) + st.bar_chart(df.set_index("intent")["confidence"], horizontal=True) + +st.divider() +st.caption("Sample queries to try:") +samples = [ + "what is my account balance", + "book a flight to new york", + "set an alarm for 7am", + "tell me about quantum physics", + "write me a poem about the ocean", +] +cols = st.columns(len(samples)) +for col, sample in zip(cols, samples): + col.code(sample, language=None) diff --git a/app/streamlit_app.py b/app/streamlit_app.py new file mode 100644 index 0000000000000000000000000000000000000000..6c8de064fd0c48130af7f204754b07eb33db6b7e --- /dev/null +++ b/app/streamlit_app.py @@ -0,0 +1,14 @@ +import streamlit as st + +st.set_page_config( + page_title="Intent Classifier", + page_icon=":material/psychology:", + layout="wide", +) + +predict_page = st.Page("pages/predict.py", title="Predict", icon=":material/chat:") +compare_page = st.Page("pages/compare.py", title="Compare Models", icon=":material/bar_chart:") +monitoring_page = st.Page("pages/monitoring.py", title="Monitoring", icon=":material/monitoring:") + +pg = st.navigation([predict_page, compare_page, monitoring_page]) +pg.run() diff --git a/artifacts/label_map.json b/artifacts/label_map.json new file mode 100644 index 0000000000000000000000000000000000000000..7c4dbcb1318efd28d640571dc244c74677a6c51e --- /dev/null +++ b/artifacts/label_map.json @@ -0,0 +1,153 @@ +{ + "accept_reservations": 0, + "account_blocked": 1, + "alarm": 2, + "application_status": 3, + "apr": 4, + "are_you_a_bot": 5, + "balance": 6, + "bill_balance": 7, + "bill_due": 8, + "book_flight": 9, + "book_hotel": 10, + "calculator": 11, + "calendar": 12, + "calendar_update": 13, + "calories": 14, + "cancel": 15, + "cancel_reservation": 16, + "car_rental": 17, + "card_declined": 18, + "carry_on": 19, + "change_accent": 20, + "change_ai_name": 21, + "change_language": 22, + "change_speed": 23, + "change_user_name": 24, + "change_volume": 25, + "confirm_reservation": 26, + "cook_time": 27, + "credit_limit": 28, + "credit_limit_change": 29, + "credit_score": 30, + "current_location": 31, + "damaged_card": 32, + "date": 33, + "definition": 34, + "direct_deposit": 35, + "directions": 36, + "distance": 37, + "do_you_have_pets": 38, + "exchange_rate": 39, + "expiration_date": 40, + "find_phone": 41, + "flight_status": 42, + "flip_coin": 43, + "food_last": 44, + "freeze_account": 45, + "fun_fact": 46, + "gas": 47, + "gas_type": 48, + "goodbye": 49, + "greeting": 50, + "how_busy": 51, + "how_old_are_you": 52, + "improve_credit_score": 53, + "income": 54, + "ingredient_substitution": 55, + "ingredients_list": 56, + "insurance": 57, + "insurance_change": 58, + "interest_rate": 59, + "international_fees": 60, + "international_visa": 61, + "jump_start": 62, + "last_maintenance": 63, + "lost_luggage": 64, + "make_call": 65, + "maybe": 66, + "meal_suggestion": 67, + "meaning_of_life": 68, + "measurement_conversion": 69, + "meeting_schedule": 70, + "min_payment": 71, + "mpg": 72, + "new_card": 73, + "next_holiday": 74, + "next_song": 75, + "no": 76, + "nutrition_info": 77, + "oil_change_how": 78, + "oil_change_when": 79, + "oos": 80, + "order": 81, + "order_checks": 82, + "order_status": 83, + "pay_bill": 84, + "payday": 85, + "pin_change": 86, + "play_music": 87, + "plug_type": 88, + "pto_balance": 89, + "pto_request": 90, + "pto_request_status": 91, + "pto_used": 92, + "recipe": 93, + "redeem_rewards": 94, + "reminder": 95, + "reminder_update": 96, + "repeat": 97, + "replacement_card_duration": 98, + "report_fraud": 99, + "report_lost_card": 100, + "reset_settings": 101, + "restaurant_reservation": 102, + "restaurant_reviews": 103, + "restaurant_suggestion": 104, + "rewards_balance": 105, + "roll_dice": 106, + "rollover_401k": 107, + "routing": 108, + "schedule_maintenance": 109, + "schedule_meeting": 110, + "share_location": 111, + "shopping_list": 112, + "shopping_list_update": 113, + "smart_home": 114, + "spelling": 115, + "spending_history": 116, + "sync_device": 117, + "taxes": 118, + "tell_joke": 119, + "text": 120, + "thank_you": 121, + "time": 122, + "timer": 123, + "timezone": 124, + "tire_change": 125, + "tire_pressure": 126, + "todo_list": 127, + "todo_list_update": 128, + "traffic": 129, + "transactions": 130, + "transfer": 131, + "translate": 132, + "travel_alert": 133, + "travel_notification": 134, + "travel_suggestion": 135, + "uber": 136, + "update_playlist": 137, + "user_name": 138, + "vaccines": 139, + "w2": 140, + "weather": 141, + "what_are_your_hobbies": 142, + "what_can_i_ask_you": 143, + "what_is_your_name": 144, + "what_song": 145, + "where_are_you_from": 146, + "whisper_mode": 147, + "who_do_you_work_for": 148, + "who_made_you": 149, + "yes": 150 +} \ No newline at end of file diff --git a/configs/classical.yaml b/configs/classical.yaml new file mode 100644 index 0000000000000000000000000000000000000000..fa76a39bb5c9446edd259618c95abf549736e0e2 --- /dev/null +++ b/configs/classical.yaml @@ -0,0 +1,25 @@ +model: + type: logistic_regression + C: 1.0 + max-iter: 1000 + random_state: 42 + +tfidf: + max-features: 50000 + ngram_range: [1, 2] + min_df: 2 + +data: + dataset: clinc_oos + subset: plus + test_size: 0.1 + val_size: 0.1 + random_state: 42 + +mlflow: + experiment_name: intent-classifier + run_name: logistic_regression + +s3: + bucket: intent-classifier145 + prefix: classical \ No newline at end of file diff --git a/configs/neural.yaml b/configs/neural.yaml new file mode 100644 index 0000000000000000000000000000000000000000..0efeda200261e554b3a705e6d50bb9c53cd9aa2c --- /dev/null +++ b/configs/neural.yaml @@ -0,0 +1,41 @@ +model: + textcnn: + embedding_dim: 128 + num_filters: 128 + kernel_sizes: [2, 3, 4] + dropout: 0.3 + + rnn: + embedding_dim: 128 + hidden_dim: 256 + num_layers: 2 + dropout: 0.3 + + lstm: + embedding_dim: 128 + hidden_dim: 256 + num_layers: 2 + dropout: 0.3 + +training: + epochs: 50 + batch_size: 64 + learning_rate: 0.001 + random_state: 42 + patience: 5 + grad_clip: 1.0 + +data: + dataset: clinc_oos + subset: plus + test_size: 0.1 + val_size: 0.1 + random_state: 42 + +mlflow: + experiment_name: intent-classifier + run_name: textcnn + +s3: + bucket: intent-classifier145 + prefix: neural \ No newline at end of file diff --git a/configs/transformer.yaml b/configs/transformer.yaml new file mode 100644 index 0000000000000000000000000000000000000000..c0c2eb21fe0444f832e19062231f32547efaf246 --- /dev/null +++ b/configs/transformer.yaml @@ -0,0 +1,29 @@ +model: + type: distilbert + name: distilbert-base-uncased + num_labels: 151 + dropout: 0.1 + +training: + epochs: 5 + batch_size: 32 + learning_rate: 0.00002 + warmup_steps: 500 + weight_decay: 0.01 + random_state: 42 + +data: + dataset: clinc_oos + subset: plus + max_length: 128 + test_size: 0.1 + val_size: 0.1 + random_state: 42 + +mlflow: + experiment_name: intent-classifier + run_name: distilbert-finetuned + +s3: + bucket: intent-classifier145 + prefix: transformer \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000000000000000000000000000000000000..996d60acb174154d972b10d4d738912beb7c0235 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,19 @@ +torch==2.6.0 +transformers==4.51.3 +accelerate==1.5.2 +datasets==3.3.2 +numpy==2.2.3 +pandas==2.2.3 +mlflow==2.21.3 +fastapi==0.115.12 +uvicorn==0.34.0 +streamlit==1.43.2 +pyyaml==6.0.2 +python-dotenv==1.0.1 +boto3==1.37.38 +evidently==0.7.5 +httpx==0.28.1 +pydantic-settings==2.8.1 +matplotlib==3.10.1 +seaborn==0.13.2 +scikit-learn==1.6.1 \ No newline at end of file diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/src/__pycache__/__init__.cpython-312.pyc b/src/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..520fe948223d0eac1a455131746e71e3ffab3150 Binary files /dev/null and b/src/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/data/__init__.py b/src/data/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/src/data/__pycache__/__init__.cpython-312.pyc b/src/data/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..463118f1281e69cbe7b55c30fde8cbc0ffa04853 Binary files /dev/null and b/src/data/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/data/__pycache__/loader.cpython-312.pyc b/src/data/__pycache__/loader.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..74a8229f1ced32b0d451af916f60e968d5a6ca49 Binary files /dev/null and b/src/data/__pycache__/loader.cpython-312.pyc differ diff --git a/src/data/__pycache__/preprocessor.cpython-312.pyc b/src/data/__pycache__/preprocessor.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..69addb46638749eab7b664d4b2d5bde43e712c34 Binary files /dev/null and b/src/data/__pycache__/preprocessor.cpython-312.pyc differ diff --git a/src/data/loader.py b/src/data/loader.py new file mode 100644 index 0000000000000000000000000000000000000000..171c7b1845772001354ad9e7535efaf0475ac436 --- /dev/null +++ b/src/data/loader.py @@ -0,0 +1,45 @@ +from pathlib import Path + +import pandas as pd +from datasets import load_dataset + + +def load_clinc150(subset: str = "plus") -> dict[str, pd.DataFrame]: + dataset = load_dataset("clinc/clinc_oos", subset) + + intent_feature = dataset["train"].features["intent"] + + splits = {} + for split_name in ["train", "validation", "test"]: + records = [ + { + "text": item["text"], + "label": intent_feature.int2str(item["intent"]), + } + for item in dataset[split_name] + ] + splits[split_name] = pd.DataFrame(records) + + return splits + + +def save_splits(splits: dict[str, pd.DataFrame], save_dir: str = "data/raw") -> None: + path = Path(save_dir) + path.mkdir(parents=True, exist_ok=True) + + for split_name, df in splits.items(): + df.to_csv(path / f"{split_name}.csv", index=False) + + +def load_splits(load_dir: str = "data/raw") -> dict[str, pd.DataFrame]: + path = Path(load_dir) + split_names = ["train", "validation", "test"] + + splits = {} + for split_name in split_names: + file_path = path / f"{split_name}.csv" + if not file_path.exists(): + raise FileNotFoundError(f"Split not found: {file_path}") + splits[split_name] = pd.read_csv(file_path) + + return splits diff --git a/src/data/preprocessor.py b/src/data/preprocessor.py new file mode 100644 index 0000000000000000000000000000000000000000..5618c87d0b9fce0baac55f536c3b1cb5cf32b19f --- /dev/null +++ b/src/data/preprocessor.py @@ -0,0 +1,62 @@ +import json +import re +from pathlib import Path + +import pandas as pd + + +def clean_text(text: str) -> str: + text = text.lower().strip() + text = re.sub(r"\s+", " ", text) + return text + + +def build_label_map(train_df: pd.DataFrame, label_col: str = "label") -> dict[str, int]: + labels = sorted(train_df[label_col].unique().tolist()) + return {label: idx for idx, label in enumerate(labels)} + + +def encode_labels( + df: pd.DataFrame, + label_map: dict[str, int], + label_col: str = "label", +) -> pd.DataFrame: + df = df.copy() + df["label_id"] = df[label_col].map(label_map) + return df + + +def save_label_map(label_map: dict[str, int], save_path: str = "artifacts/label_map.json") -> None: + path = Path(save_path) + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "w") as f: + json.dump(label_map, f, indent=2) + + +def load_label_map(load_path: str = "artifacts/label_map.json") -> dict[str, int]: + path = Path(load_path) + if not path.exists(): + raise FileNotFoundError(f"Label map not found: {path}") + with open(path) as f: + return json.load(f) + + +def get_id_to_label(label_map: dict[str, int]) -> dict[int, str]: + return {idx: label for label, idx in label_map.items()} + + +def preprocess( + splits: dict[str, pd.DataFrame], +) -> tuple[dict[str, pd.DataFrame], dict[str, int]]: + label_map = build_label_map(splits["train"]) + + processed = {} + for split_name, df in splits.items(): + df = df.copy() + df["text"] = df["text"].apply(clean_text) + df = encode_labels(df, label_map) + processed[split_name] = df + + save_label_map(label_map) + + return processed, label_map diff --git a/src/evaluation/__init__.py b/src/evaluation/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/src/evaluation/__pycache__/__init__.cpython-312.pyc b/src/evaluation/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..501eee5c124e1b1e14564a03ce9510bb2ef577e2 Binary files /dev/null and b/src/evaluation/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/evaluation/__pycache__/metrics.cpython-312.pyc b/src/evaluation/__pycache__/metrics.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..243f779a84f0a72d4617df4e488367194c448b28 Binary files /dev/null and b/src/evaluation/__pycache__/metrics.cpython-312.pyc differ diff --git a/src/evaluation/evaluate_all.py b/src/evaluation/evaluate_all.py new file mode 100644 index 0000000000000000000000000000000000000000..6a052852483842e2a70a9623602f2157467a8406 --- /dev/null +++ b/src/evaluation/evaluate_all.py @@ -0,0 +1,417 @@ +import time +from pathlib import Path + +import matplotlib.pyplot as plt +import mlflow +import numpy as np +import pandas as pd +import torch +from sklearn.metrics import ConfusionMatrixDisplay, confusion_matrix +from torch.utils.data import DataLoader + +from src.data.loader import load_splits +from src.data.preprocessor import preprocess +from src.features.tfidf import load_vectorizer, transform +from src.models.classical import LogisticRegressionModel, SVMModel +from src.models.neural import IntentDatasetNN, LSTMModel, TextCNN, Vocabulary +from src.models.transformer import IntentDatasetHF, TransformerModel +from src.utils.config import load_config +from src.utils.mlflow_utils import get_or_create_experiment +from src.utils.settings import settings + +VECTORIZER_PATH = "artifacts/vectorizers/tfidf.pkl" +VOCAB_PATH = "artifacts/models/vocab.pkl" +LOGREG_PATH = "artifacts/models/logreg.pkl" +SVM_PATH = "artifacts/models/svm.pkl" +TEXTCNN_PATH = "artifacts/models/textcnn.pt" +RNN_PATH = "artifacts/models/rnn.pt" +LSTM_PATH = "artifacts/models/lstm.pt" +DISTILBERT_DIR = "artifacts/models/distilbert" +MAX_LENGTH_NN = 32 +MAX_LENGTH_HF = 128 +REPORT_DIR = Path("artifacts/evaluation") + + +def load_data(): + splits = load_splits("data/raw") + processed, label_map = preprocess(splits) + id_to_label = {v: k for k, v in label_map.items()} + label_names = [id_to_label[i] for i in range(len(label_map))] + return processed, label_map, label_names + + +def compute_metrics(y_true: np.ndarray, y_pred: np.ndarray) -> dict: + from sklearn.metrics import accuracy_score, f1_score + + return { + "accuracy": round(accuracy_score(y_true, y_pred), 4), + "macro_f1": round(f1_score(y_true, y_pred, average="macro", zero_division=0), 4), + "weighted_f1": round(f1_score(y_true, y_pred, average="weighted", zero_division=0), 4), + } + + +def measure_latency(predict_fn, input_data, n_runs: int = 50) -> dict: + latencies = [] + for _ in range(n_runs): + start = time.perf_counter() + predict_fn(input_data) + latencies.append((time.perf_counter() - start) * 1000) + latencies = np.array(latencies) + return { + "latency_mean_ms": round(float(np.mean(latencies)), 3), + "latency_p50_ms": round(float(np.percentile(latencies, 50)), 3), + "latency_p95_ms": round(float(np.percentile(latencies, 95)), 3), + "latency_p99_ms": round(float(np.percentile(latencies, 99)), 3), + } + + +def eval_logreg(processed: dict, label_names: list[str]) -> tuple[dict, np.ndarray]: + print(" evaluating logistic regression...") + vectorizer = load_vectorizer(VECTORIZER_PATH) + X_test = transform(vectorizer, processed["test"]["text"].tolist()) + y_test = processed["test"]["label_id"].values + + model = LogisticRegressionModel() + model.load(LOGREG_PATH) + + y_pred = model.predict(X_test) + metrics = compute_metrics(y_test, y_pred) + latency = measure_latency(model.predict, X_test[:100]) + return {**metrics, **latency}, y_pred + + +def eval_svm(processed: dict, label_names: list[str]) -> tuple[dict, np.ndarray]: + print(" evaluating svm...") + vectorizer = load_vectorizer(VECTORIZER_PATH) + X_test = transform(vectorizer, processed["test"]["text"].tolist()) + y_test = processed["test"]["label_id"].values + + model = SVMModel() + model.load(SVM_PATH) + + y_pred = model.predict(X_test) + metrics = compute_metrics(y_test, y_pred) + latency = measure_latency(model.predict, X_test[:100]) + return {**metrics, **latency}, y_pred + + +def eval_neural( + model_type: str, + model_path: str, + processed: dict, + label_map: dict, +) -> tuple[dict, np.ndarray]: + print(f" evaluating {model_type}...") + vocab = Vocabulary.load(VOCAB_PATH) + device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + num_classes = len(label_map) + + neural_config = load_config("neural") + model_cfg = neural_config["model"][model_type] + + if model_type == "textcnn": + model = TextCNN( + vocab_size=len(vocab), + embedding_dim=model_cfg["embedding_dim"], + num_filters=model_cfg["num_filters"], + kernel_sizes=model_cfg["kernel_sizes"], + num_classes=num_classes, + dropout=model_cfg["dropout"], + ) + elif model_type == "rnn": + from src.models.neural import RNNModel + + model = RNNModel( + vocab_size=len(vocab), + embedding_dim=model_cfg["embedding_dim"], + hidden_dim=model_cfg["hidden_dim"], + num_layers=model_cfg["num_layers"], + num_classes=num_classes, + dropout=model_cfg["dropout"], + ) + elif model_type == "lstm": + model = LSTMModel( + vocab_size=len(vocab), + embedding_dim=model_cfg["embedding_dim"], + hidden_dim=model_cfg["hidden_dim"], + num_layers=model_cfg["num_layers"], + num_classes=num_classes, + dropout=model_cfg["dropout"], + ) + + model.load(model_path) + model.to(device) + model.eval() + + dataset = IntentDatasetNN( + processed["test"]["text"].tolist(), + processed["test"]["label_id"].tolist(), + vocab, + MAX_LENGTH_NN, + ) + loader = DataLoader(dataset, batch_size=64, shuffle=False) + + all_preds = [] + y_test = processed["test"]["label_id"].values + + with torch.no_grad(): + for inputs, labels in loader: + inputs = inputs.to(device) + logits = model(inputs) + preds = logits.argmax(dim=1).cpu().numpy() + all_preds.extend(preds) + + y_pred = np.array(all_preds) + metrics = compute_metrics(y_test, y_pred) + + def predict_fn(loader): + with torch.no_grad(): + for inputs, _ in loader: + model(inputs.to(device)) + + latency = measure_latency(predict_fn, loader, n_runs=20) + return {**metrics, **latency}, y_pred + + +def eval_distilbert(processed: dict, label_map: dict) -> tuple[dict, np.ndarray]: + print(" evaluating distilbert...") + device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + transformer = TransformerModel( + model_name=DISTILBERT_DIR, + num_labels=len(label_map), + ) + transformer.model.to(device) + transformer.model.eval() + + dataset = IntentDatasetHF( + processed["test"]["text"].tolist(), + processed["test"]["label_id"].tolist(), + transformer.tokenizer, + MAX_LENGTH_HF, + ) + loader = DataLoader(dataset, batch_size=32, shuffle=False) + + all_preds = [] + y_test = processed["test"]["label_id"].values + + with torch.no_grad(): + for batch in loader: + input_ids = batch["input_ids"].to(device) + attention_mask = batch["attention_mask"].to(device) + outputs = transformer.model(input_ids=input_ids, attention_mask=attention_mask) + preds = outputs.logits.argmax(dim=1).cpu().numpy() + all_preds.extend(preds) + + y_pred = np.array(all_preds) + metrics = compute_metrics(y_test, y_pred) + + single_text = processed["test"]["text"].iloc[:1].tolist() + single_dataset = IntentDatasetHF( + single_text, + [0], + transformer.tokenizer, + MAX_LENGTH_HF, + ) + single_loader = DataLoader(single_dataset, batch_size=1) + + def predict_fn(loader): + with torch.no_grad(): + for batch in loader: + transformer.model( + input_ids=batch["input_ids"].to(device), + attention_mask=batch["attention_mask"].to(device), + ) + + latency = measure_latency(predict_fn, single_loader, n_runs=50) + return {**metrics, **latency}, y_pred + + +def plot_comparison(results: dict, save_path: str) -> None: + df = pd.DataFrame(results).T.reset_index() + df.columns = ["model"] + list(df.columns[1:]) + + fig, axes = plt.subplots(1, 3, figsize=(18, 6)) + + metrics = ["accuracy", "macro_f1", "weighted_f1"] + titles = ["Accuracy", "Macro F1", "Weighted F1"] + colors = ["#4C72B0", "#DD8452", "#55A868", "#C44E52", "#8172B2", "#937860"] + + for ax, metric, title in zip(axes, metrics, titles): + bars = ax.bar(df["model"], df[metric].astype(float), color=colors) + ax.set_title(title, fontsize=14) + ax.set_ylim(0, 1.1) + ax.set_xticks(range(len(df["model"]))) + ax.set_xticklabels(df["model"], rotation=30, ha="right") + for bar, val in zip(bars, df[metric].astype(float)): + ax.text( + bar.get_x() + bar.get_width() / 2, + bar.get_height() + 0.01, + f"{val:.3f}", + ha="center", + va="bottom", + fontsize=9, + ) + + plt.suptitle("Model Comparison — CLINC150", fontsize=16, y=1.02) + plt.tight_layout() + plt.savefig(save_path, dpi=100, bbox_inches="tight") + plt.close() + print(f" saved: {save_path}") + + +def plot_latency(results: dict, save_path: str) -> None: + models = list(results.keys()) + p50 = [results[m]["latency_p50_ms"] for m in models] + p95 = [results[m]["latency_p95_ms"] for m in models] + + x = np.arange(len(models)) + width = 0.35 + + fig, ax = plt.subplots(figsize=(12, 6)) + ax.bar(x - width / 2, p50, width, label="P50", color="#4C72B0") + ax.bar(x + width / 2, p95, width, label="P95", color="#DD8452") + ax.set_yscale("log") + ax.set_ylabel("Latency (ms) — log scale") + ax.set_title("Inference Latency Comparison") + ax.set_xticks(x) + ax.set_xticklabels(models, rotation=30, ha="right") + ax.legend() + plt.tight_layout() + plt.savefig(save_path, dpi=100, bbox_inches="tight") + plt.close() + print(f" saved: {save_path}") + + +def get_top_confused_pairs( + y_true: np.ndarray, + y_pred: np.ndarray, + label_names: list[str], + top_n: int = 20, +) -> pd.DataFrame: + cm = confusion_matrix(y_true, y_pred) + np.fill_diagonal(cm, 0) + + pairs = [] + for i in range(cm.shape[0]): + for j in range(cm.shape[1]): + if cm[i, j] > 0: + pairs.append( + { + "true_label": label_names[i], + "predicted_label": label_names[j], + "count": int(cm[i, j]), + } + ) + + df = pd.DataFrame(pairs).sort_values("count", ascending=False).head(top_n) + return df.reset_index(drop=True) + + +def plot_top_confused_pairs(df: pd.DataFrame, model_name: str, save_path: str) -> None: + fig, ax = plt.subplots(figsize=(10, max(6, len(df) * 0.35))) + labels = [f"{row.true_label} -> {row.predicted_label}" for row in df.itertuples()] + ax.barh(labels, df["count"], color="#C44E52") + ax.invert_yaxis() + ax.set_xlabel("Misclassification Count") + ax.set_title(f"Top Confused Pairs — {model_name}") + plt.tight_layout() + plt.savefig(save_path, dpi=100, bbox_inches="tight") + plt.close() + print(f" saved: {save_path}") + + +def plot_oos_binary_confusion( + y_true: np.ndarray, + y_pred: np.ndarray, + label_map: dict, + model_name: str, + save_path: str, +) -> None: + oos_id = label_map.get("oos") + y_true_binary = (y_true == oos_id).astype(int) + y_pred_binary = (y_pred == oos_id).astype(int) + + cm = confusion_matrix(y_true_binary, y_pred_binary) + fig, ax = plt.subplots(figsize=(5, 5)) + disp = ConfusionMatrixDisplay( + confusion_matrix=cm, + display_labels=["in-scope", "oos"], + ) + disp.plot(ax=ax, colorbar=False, cmap="Blues") + ax.set_title(f"OOS Detection — {model_name}") + plt.tight_layout() + plt.savefig(save_path, dpi=100, bbox_inches="tight") + plt.close() + print(f" saved: {save_path}") + + +def main(): + REPORT_DIR.mkdir(parents=True, exist_ok=True) + + print("loading data...") + processed, label_map, label_names = load_data() + y_test = processed["test"]["label_id"].values + + print("\nevaluating all models...") + results = {} + predictions = {} + + results["logreg"], predictions["logreg"] = eval_logreg(processed, label_names) + results["svm"], predictions["svm"] = eval_svm(processed, label_names) + results["textcnn"], predictions["textcnn"] = eval_neural("textcnn", TEXTCNN_PATH, processed, label_map) + results["rnn"], predictions["rnn"] = eval_neural("rnn", RNN_PATH, processed, label_map) + results["lstm"], predictions["lstm"] = eval_neural("lstm", LSTM_PATH, processed, label_map) + results["distilbert"], predictions["distilbert"] = eval_distilbert(processed, label_map) + + print("\ngenerating plots...") + plot_comparison(results, str(REPORT_DIR / "model_comparison.png")) + plot_latency(results, str(REPORT_DIR / "latency_comparison.png")) + + for model_name, y_pred in predictions.items(): + confused_df = get_top_confused_pairs(y_test, y_pred, label_names, top_n=20) + confused_csv_path = REPORT_DIR / f"{model_name}_top_confused_pairs.csv" + confused_df.to_csv(confused_csv_path, index=False) + print(f" saved: {confused_csv_path}") + + plot_top_confused_pairs( + confused_df, + model_name, + str(REPORT_DIR / f"{model_name}_top_confused_pairs.png"), + ) + + plot_oos_binary_confusion( + y_test, + y_pred, + label_map, + model_name, + str(REPORT_DIR / f"{model_name}_oos_binary.png"), + ) + + print("\nlogging to MLflow...") + mlflow.set_tracking_uri(settings.mlflow_tracking_uri) + experiment_id = get_or_create_experiment("intent-classifier") + + with mlflow.start_run(experiment_id=experiment_id, run_name="unified-evaluation"): + for model_name, metrics in results.items(): + for metric_name, value in metrics.items(): + mlflow.log_metric(f"{model_name}.{metric_name}", value) + + mlflow.log_artifact(str(REPORT_DIR / "model_comparison.png")) + mlflow.log_artifact(str(REPORT_DIR / "latency_comparison.png")) + + print("\nfinal results:") + print(f"{'model':<14} {'accuracy':<12} {'macro_f1':<12} {'weighted_f1':<14} {'p50_ms':<12} {'p95_ms'}") + print("-" * 78) + for model_name, metrics in results.items(): + print( + f"{model_name:<14} " + f"{metrics['accuracy']:<12} " + f"{metrics['macro_f1']:<12} " + f"{metrics['weighted_f1']:<14} " + f"{metrics['latency_p50_ms']:<12} " + f"{metrics['latency_p95_ms']}" + ) + + +if __name__ == "__main__": + main() diff --git a/src/evaluation/metrics.py b/src/evaluation/metrics.py new file mode 100644 index 0000000000000000000000000000000000000000..0f40feddd4ff47bb4434fb79fdfcc946510336c6 --- /dev/null +++ b/src/evaluation/metrics.py @@ -0,0 +1,47 @@ +import time + +import numpy as np +import pandas as pd +from sklearn.metrics import accuracy_score, classification_report, f1_score + + +def compute_classification_metrics( + y_true: np.ndarray, + y_pred: np.ndarray, +) -> dict: + return { + "accuracy": round(accuracy_score(y_true, y_pred), 4), + "macro_f1": round(f1_score(y_true, y_pred, average="macro", zero_division=0), 4), + "weighted_f1": round(f1_score(y_true, y_pred, average="weighted", zero_division=0), 4), + } + + +def compute_latency(predict_fn, inputs, n_runs: int = 100) -> dict: + latencies = [] + for _ in range(n_runs): + start = time.perf_counter() + predict_fn(inputs) + latencies.append((time.perf_counter() - start) * 1000) + + latencies = np.array(latencies) + return { + "latency_mean_ms": round(float(np.mean(latencies)), 3), + "latency_p50_ms": round(float(np.percentile(latencies, 50)), 3), + "latency_p95_ms": round(float(np.percentile(latencies, 95)), 3), + "latency_p99_ms": round(float(np.percentile(latencies, 99)), 3), + } + + +def get_classification_report( + y_true: np.ndarray, + y_pred: np.ndarray, + label_names: list[str], +) -> str: + return classification_report(y_true, y_pred, target_names=label_names, zero_division=0) + + +def compare_models(results: dict) -> pd.DataFrame: + rows = [] + for model_name, metrics in results.items(): + rows.append({"model": model_name, **metrics}) + return pd.DataFrame(rows).set_index("model") diff --git a/src/features/__init__.py b/src/features/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/src/features/__pycache__/__init__.cpython-312.pyc b/src/features/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..795277cd3f5c9d61c3c0559f73867a196150ed82 Binary files /dev/null and b/src/features/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/features/__pycache__/tfidf.cpython-312.pyc b/src/features/__pycache__/tfidf.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..f3d550190c3140916fabe3c05fa1bfb2bd7abcf2 Binary files /dev/null and b/src/features/__pycache__/tfidf.cpython-312.pyc differ diff --git a/src/features/tfidf.py b/src/features/tfidf.py new file mode 100644 index 0000000000000000000000000000000000000000..26fb80791adff0f4f5dc284de4412666b4b38a4c --- /dev/null +++ b/src/features/tfidf.py @@ -0,0 +1,50 @@ +import pickle +from pathlib import Path + +import scipy.sparse as sp +from sklearn.feature_extraction.text import TfidfVectorizer + + +def _get_tfidf_value(tfidf_config: dict, key: str): + kebab_key = key.replace("_", "-") + if key in tfidf_config: + return tfidf_config[key] + if kebab_key in tfidf_config: + return tfidf_config[kebab_key] + raise KeyError(key) + + +def fit_tfidf(train_texts: list[str], config: dict) -> TfidfVectorizer: + tfidf_config = config["tfidf"] + vectorizer = TfidfVectorizer( + max_features=_get_tfidf_value(tfidf_config, "max_features"), + ngram_range=tuple(_get_tfidf_value(tfidf_config, "ngram_range")), + min_df=_get_tfidf_value(tfidf_config, "min_df"), + ) + vectorizer.fit(train_texts) + return vectorizer + + +def transform(vectorizer: TfidfVectorizer, texts: list[str]) -> sp.csr_matrix: + return vectorizer.transform(texts) + + +def save_vectorizer(vectorizer: TfidfVectorizer, save_path: str = "artifacts/vectorizers/tfidf.pkl") -> None: + path = Path(save_path) + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "wb") as f: + pickle.dump(vectorizer, f) + + +def load_vectorizer( + load_path: str = "artifacts/vectorizers/tfidf.pkl", +) -> TfidfVectorizer: + path = Path(load_path) + if not path.exists(): + raise FileNotFoundError(f"Vectorizer not found: {path}") + with open(path, "rb") as f: + return pickle.load(f) + + +def get_feature_names(vectorizer: TfidfVectorizer) -> list[str]: + return vectorizer.get_feature_names_out().tolist() diff --git a/src/models/__init__.py b/src/models/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/src/models/__pycache__/__init__.cpython-312.pyc b/src/models/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..8fb2535fa21572c122f86ae58e3b7e14fb235261 Binary files /dev/null and b/src/models/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/models/__pycache__/classical.cpython-312.pyc b/src/models/__pycache__/classical.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..c8e0bd1b43a9153032cf1c1d8a7bd32e402e5b8f Binary files /dev/null and b/src/models/__pycache__/classical.cpython-312.pyc differ diff --git a/src/models/__pycache__/neural.cpython-312.pyc b/src/models/__pycache__/neural.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..c2e2f49165b6ca681766063f582a6c71cda5b10f Binary files /dev/null and b/src/models/__pycache__/neural.cpython-312.pyc differ diff --git a/src/models/__pycache__/transformer.cpython-312.pyc b/src/models/__pycache__/transformer.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..20478b2f22a0ea113b3cc777c5c3c14d25f484da Binary files /dev/null and b/src/models/__pycache__/transformer.cpython-312.pyc differ diff --git a/src/models/classical.py b/src/models/classical.py new file mode 100644 index 0000000000000000000000000000000000000000..60ece70e5c0432aa8eafddd6b8c2e98b6f59c43e --- /dev/null +++ b/src/models/classical.py @@ -0,0 +1,67 @@ +import pickle +from pathlib import Path + +import numpy as np +import scipy.sparse as sp +from sklearn.linear_model import LogisticRegression +from sklearn.svm import LinearSVC + + +class LogisticRegressionModel: + def __init__(self, C: float = 1.0, max_iter: int = 1000, random_state: int = 42): + self.model = LogisticRegression( + C=C, + max_iter=max_iter, + random_state=random_state, + n_jobs=-1, + ) + + def fit(self, X: sp.csc_matrix, y: np.ndarray) -> None: + self.model.fit(X, y) + + def predict(self, X: sp.csr_matrix) -> np.ndarray: + return self.model.predict(X) + + def predict_proba(self, X: sp.csr_matrix) -> np.ndarray: + return self.model.predict_proba(X) + + def save(self, save_path: str) -> None: + path = Path(save_path) + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "wb") as f: + pickle.dump(self.model, f) + + def load(self, load_path: str) -> None: + path = Path(load_path) + if not path.exists(): + raise FileNotFoundError(f"Model not found: {path}") + with open(path, "rb") as f: + self.model = pickle.load(f) + + +class SVMModel: + def __init__(self, C: float = 1.0, max_iter: int = 1000, random_state: int = 42): + self.model = LinearSVC( + C=C, + max_iter=max_iter, + random_state=random_state, + ) + + def fit(self, X: sp.csr_matrix, y: np.ndarray) -> None: + self.model.fit(X, y) + + def predict(self, X: sp.csr_matrix) -> np.ndarray: + return self.model.predict(X) + + def save(self, save_path: str) -> None: + path = Path(save_path) + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "wb") as f: + pickle.dump(self.model, f) + + def load(self, load_path: str) -> None: + path = Path(load_path) + if not path.exists(): + raise FileNotFoundError(f"Model not found: {path}") + with open(path, "rb") as f: + self.model = pickle.load(f) diff --git a/src/models/neural.py b/src/models/neural.py new file mode 100644 index 0000000000000000000000000000000000000000..b9cf469ae48c9a24bba24e472e33f8e30df814ca --- /dev/null +++ b/src/models/neural.py @@ -0,0 +1,230 @@ +import pickle +from pathlib import Path + +import numpy as np +import torch +import torch.nn as nn +from torch.utils.data import Dataset + + +class Vocabulary: + def __init__(self, min_freq: int = 1): + self.min_freq = min_freq + self.token2idx = {"": 0, "": 1} + self.idx2token = {0: "", 1: ""} + + def build(self, texts: list[str]) -> None: + freq = {} + for text in texts: + for token in text.split(): + freq[token] = freq.get(token, 0) + 1 + + for token, count in freq.items(): + if count >= self.min_freq and token not in self.token2idx: + idx = len(self.token2idx) + self.token2idx[token] = idx + self.idx2token[idx] = token + + def encode(self, text: str, max_length: int) -> list[int]: + tokens = text.split()[:max_length] + ids = [self.token2idx.get(t, 1) for t in tokens] + ids += [0] * (max_length - len(ids)) + return ids + + def __len__(self) -> int: + return len(self.token2idx) + + def save(self, save_path: str) -> None: + path = Path(save_path) + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "wb") as f: + pickle.dump(self, f) + + @staticmethod + def load(load_path: str) -> "Vocabulary": + path = Path(load_path) + if not path.exists(): + raise FileNotFoundError(f"Vocabulary not found: {path}") + with open(path, "rb") as f: + return pickle.load(f) + + +class IntentDatasetNN(Dataset): + def __init__( + self, + texts: list[str], + labels: list[int], + vocab: Vocabulary, + max_length: int = 32, + ): + self.labels = labels + self.encodings = [vocab.encode(text, max_length) for text in texts] + + def __len__(self) -> int: + return len(self.labels) + + def __getitem__(self, idx: int) -> tuple[torch.Tensor, torch.Tensor]: + return ( + torch.tensor(self.encodings[idx], dtype=torch.long), + torch.tensor(self.labels[idx], dtype=torch.long), + ) + + +class TextCNN(nn.Module): + def __init__( + self, + vocab_size: int, + embedding_dim: int, + num_filters: int, + kernel_sizes: list[int], + num_classes: int, + dropout: float, + pad_idx: int = 0, + ): + super().__init__() + + self.embedding = nn.Embedding(vocab_size, embedding_dim, padding_idx=pad_idx) + + self.convs = nn.ModuleList( + [ + nn.Conv1d( + in_channels=embedding_dim, + out_channels=num_filters, + kernel_size=k, + ) + for k in kernel_sizes + ] + ) + + self.dropout = nn.Dropout(dropout) + self.fc = nn.Linear(num_filters * len(kernel_sizes), num_classes) + + def forward(self, x: torch.Tensor) -> torch.Tensor: + embedded = self.embedding(x) + embedded = embedded.permute(0, 2, 1) + + pooled = [] + for conv in self.convs: + activated = torch.relu(conv(embedded)) + pool = torch.max(activated, dim=2).values + pooled.append(pool) + + concatenated = torch.cat(pooled, dim=1) + dropped = self.dropout(concatenated) + return self.fc(dropped) + + def save(self, save_path: str) -> None: + path = Path(save_path) + path.parent.mkdir(parents=True, exist_ok=True) + torch.save(self.state_dict(), path) + + def load(self, load_path: str) -> None: + path = Path(load_path) + if not path.exists(): + raise FileNotFoundError(f"Model not found: {path}") + self.load_state_dict(torch.load(path, map_location="cpu")) + + def predict_proba(self, x: torch.Tensor) -> np.ndarray: + self.eval() + with torch.no_grad(): + logits = self.forward(x) + probs = torch.softmax(logits, dim=1) + return probs.cpu().numpy() + + +class RNNModel(nn.Module): + def __init__( + self, + vocab_size: int, + embedding_dim: int, + hidden_dim: int, + num_layers: int, + num_classes: int, + dropout: float, + pad_idx: int = 0, + ): + super().__init__() + + self.embedding = nn.Embedding(vocab_size, embedding_dim, padding_idx=pad_idx) + self.rnn = nn.RNN( + input_size=embedding_dim, + hidden_size=hidden_dim, + num_layers=num_layers, + batch_first=True, + dropout=dropout if num_layers > 1 else 0.0, + ) + self.dropout = nn.Dropout(dropout) + self.fc = nn.Linear(hidden_dim, num_classes) + + def forward(self, x: torch.Tensor) -> torch.Tensor: + embedded = self.dropout(self.embedding(x)) + _, hidden = self.rnn(embedded) + out = self.dropout(hidden[-1]) + return self.fc(out) + + def save(self, save_path: str) -> None: + path = Path(save_path) + path.parent.mkdir(parents=True, exist_ok=True) + torch.save(self.state_dict(), path) + + def load(self, load_path: str) -> None: + path = Path(load_path) + if not path.exists(): + raise FileNotFoundError(f"Model not found: {path}") + self.load_state_dict(torch.load(path, map_location="cpu")) + + def predict_proba(self, x: torch.Tensor) -> np.ndarray: + self.eval() + with torch.no_grad(): + logits = self.forward(x) + probs = torch.softmax(logits, dim=1) + return probs.cpu().numpy() + + +class LSTMModel(nn.Module): + def __init__( + self, + vocab_size: int, + embedding_dim: int, + hidden_dim: int, + num_layers: int, + num_classes: int, + dropout: float, + pad_idx: int = 0, + ): + super().__init__() + + self.embedding = nn.Embedding(vocab_size, embedding_dim, padding_idx=pad_idx) + self.lstm = nn.LSTM( + input_size=embedding_dim, + hidden_size=hidden_dim, + num_layers=num_layers, + batch_first=True, + dropout=dropout if num_layers > 1 else 0.0, + ) + self.dropout = nn.Dropout(dropout) + self.fc = nn.Linear(hidden_dim, num_classes) + + def forward(self, x: torch.Tensor) -> torch.Tensor: + embedded = self.dropout(self.embedding(x)) + _, (hidden, _) = self.lstm(embedded) + out = self.dropout(hidden[-1]) + return self.fc(out) + + def save(self, save_path: str) -> None: + path = Path(save_path) + path.parent.mkdir(parents=True, exist_ok=True) + torch.save(self.state_dict(), path) + + def load(self, load_path: str) -> None: + path = Path(load_path) + if not path.exists(): + raise FileNotFoundError(f"Model not found: {path}") + self.load_state_dict(torch.load(path, map_location="cpu")) + + def predict_proba(self, x: torch.Tensor) -> np.ndarray: + self.eval() + with torch.no_grad(): + logits = self.forward(x) + probs = torch.softmax(logits, dim=1) + return probs.cpu().numpy() diff --git a/src/models/transformer.py b/src/models/transformer.py new file mode 100644 index 0000000000000000000000000000000000000000..dde2afbff2fa5a4452ccf3879ee5d6f147bae15e --- /dev/null +++ b/src/models/transformer.py @@ -0,0 +1,77 @@ +from pathlib import Path + +import numpy as np +import torch +from torch.utils.data import Dataset +from transformers import AutoModelForSequenceClassification, AutoTokenizer + + +class IntentDatasetHF(Dataset): + def __init__( + self, + texts: list[str], + labels: list[int], + tokenizer, + max_length: int = 128, + ): + self.labels = labels + self.encodings = tokenizer( + texts, + truncation=True, + padding="max_length", + max_length=max_length, + return_tensors="pt", + ) + + def __len__(self) -> int: + return len(self.labels) + + def __getitem__(self, idx: int) -> dict: + return { + "input_ids": self.encodings["input_ids"][idx], + "attention_mask": self.encodings["attention_mask"][idx], + "labels": torch.tensor(self.labels[idx], dtype=torch.long), + } + + +class TransformerModel: + def __init__(self, model_name: str, num_labels: int, dropout: float = 0.1): + self.model_name = model_name + self.num_labels = num_labels + self.model = AutoModelForSequenceClassification.from_pretrained( + model_name, + num_labels=num_labels, + seq_classif_dropout=dropout, + ) + self.tokenizer = AutoTokenizer.from_pretrained(model_name) + + def save(self, save_dir: str) -> None: + path = Path(save_dir) + path.mkdir(parents=True, exist_ok=True) + self.model.save_pretrained(save_dir) + self.tokenizer.save_pretrained(save_dir) + + def load(self, load_dir: str) -> None: + path = Path(load_dir) + if not path.exists(): + raise FileNotFoundError(f"Model directory not found: {path}") + self.model = AutoModelForSequenceClassification.from_pretrained(load_dir) + self.tokenizer = AutoTokenizer.from_pretrained(load_dir) + + def predict_proba(self, texts: list[str], max_length: int = 128) -> np.ndarray: + self.model.eval() + encodings = self.tokenizer( + texts, + truncation=True, + padding="max_length", + max_length=max_length, + return_tensors="pt", + ) + with torch.no_grad(): + outputs = self.model(**encodings) + probs = torch.softmax(outputs.logits, dim=1) + return probs.cpu().numpy() + + +def get_tokenizer(model_name: str): + return AutoTokenizer.from_pretrained(model_name) diff --git a/src/monitoring/__init__.py b/src/monitoring/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/src/monitoring/__pycache__/__init__.cpython-312.pyc b/src/monitoring/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..cba2d4b1d5c1586c86411dc33518a0a964b263c0 Binary files /dev/null and b/src/monitoring/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/monitoring/__pycache__/drift.cpython-312.pyc b/src/monitoring/__pycache__/drift.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..4e936e8645994dce1383f2453472a7ae39098dc4 Binary files /dev/null and b/src/monitoring/__pycache__/drift.cpython-312.pyc differ diff --git a/src/monitoring/drift.py b/src/monitoring/drift.py new file mode 100644 index 0000000000000000000000000000000000000000..653b0f6ee8b23820b92f72a3ef031a1ed3e7e551 --- /dev/null +++ b/src/monitoring/drift.py @@ -0,0 +1,146 @@ +from pathlib import Path + +import pandas as pd +from evidently import Report +from evidently.presets import DataDriftPreset + +REFERENCE_PATH = "artifacts/monitoring/reference_data.csv" +CURRENT_PATH = "artifacts/monitoring/current_data.csv" + + +def save_reference_data( + texts: list[str], + predictions: list[str], + confidences: list[float], + save_path: str = REFERENCE_PATH, +) -> None: + path = Path(save_path) + path.parent.mkdir(parents=True, exist_ok=True) + + df = pd.DataFrame( + { + "text": texts, + "text_length": [len(t) for t in texts], + "prediction": predictions, + "confidence": confidences, + } + ) + df.to_csv(path, index=False) + + +def append_current_data( + texts: list[str], + predictions: list[str], + confidences: list[float], + save_path: str = CURRENT_PATH, + window_size: int = 1000, +) -> None: + path = Path(save_path) + path.parent.mkdir(parents=True, exist_ok=True) + + new_df = pd.DataFrame( + { + "text": texts, + "text_length": [len(t) for t in texts], + "prediction": predictions, + "confidence": confidences, + } + ) + + if path.exists(): + existing_df = pd.read_csv(path) + combined = pd.concat([existing_df, new_df], ignore_index=True) + combined = combined.tail(window_size) + else: + combined = new_df + + combined.to_csv(path, index=False) + + +def load_reference_data(load_path: str = REFERENCE_PATH) -> pd.DataFrame: + path = Path(load_path) + if not path.exists(): + raise FileNotFoundError(f"Reference data not found: {path}") + return pd.read_csv(path) + + +def load_current_data(load_path: str = CURRENT_PATH) -> pd.DataFrame: + path = Path(load_path) + if not path.exists(): + raise FileNotFoundError(f"Current data not found: {path}") + return pd.read_csv(path) + + +def run_drift_report( + reference_df: pd.DataFrame, + current_df: pd.DataFrame, +) -> Report: + report = Report(metrics=[DataDriftPreset()]) + snapshot = report.run(reference_data=reference_df, current_data=current_df) + return snapshot + + +def get_drift_summary(snapshot) -> dict: + result_dict = snapshot.dict() + + summary = { + "drifted_columns": [], + "total_columns": 0, + "drift_share": 0.0, + } + + for metric in result_dict.get("metrics", []): + metric_id = metric.get("metric_id", "") + value = metric.get("value", {}) + + if "DriftedColumnsCount" in metric_id: + summary["drifted_columns_count"] = value.get("count", 0) + summary["drift_share"] = value.get("share", 0.0) + + if "ValueDrift" in metric_id and isinstance(value, (int, float)): + column_name = metric.get("metric_name", metric_id) + if value > 0.5: + summary["drifted_columns"].append( + { + "column": column_name, + "drift_score": round(value, 4), + } + ) + + return summary + + +def save_drift_report_html(snapshot, save_path: str = "artifacts/monitoring/drift_report.html") -> None: + path = Path(save_path) + path.parent.mkdir(parents=True, exist_ok=True) + snapshot.save_html(str(path)) + + +def get_confidence_drift(reference_df: pd.DataFrame, current_df: pd.DataFrame) -> dict: + ref_mean = reference_df["confidence"].mean() + cur_mean = current_df["confidence"].mean() + drop = ref_mean - cur_mean + + return { + "reference_avg_confidence": round(float(ref_mean), 4), + "current_avg_confidence": round(float(cur_mean), 4), + "confidence_drop": round(float(drop), 4), + "is_degraded": bool(drop > 0.1), + } + + +def get_oos_rate_drift( + reference_df: pd.DataFrame, + current_df: pd.DataFrame, + oos_threshold: float = 0.5, +) -> dict: + ref_oos_rate = (reference_df["confidence"] < oos_threshold).mean() + cur_oos_rate = (current_df["confidence"] < oos_threshold).mean() + increase = cur_oos_rate - ref_oos_rate + + return { + "reference_oos_rate": round(float(ref_oos_rate), 4), + "current_oos_rate": round(float(cur_oos_rate), 4), + "oos_rate_increase": round(float(increase), 4), + "is_anomalous": bool(increase > 0.15), + } diff --git a/src/serving/__init__.py b/src/serving/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/src/serving/__pycache__/__init__.cpython-312.pyc b/src/serving/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..919af3f3f3d9463b77dd7e6617067594ef546a30 Binary files /dev/null and b/src/serving/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/serving/__pycache__/ab_router.cpython-312.pyc b/src/serving/__pycache__/ab_router.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..5ff1caccf049aaa2c8c2b9aeab163c61c87fa2d3 Binary files /dev/null and b/src/serving/__pycache__/ab_router.cpython-312.pyc differ diff --git a/src/serving/__pycache__/predictor.cpython-312.pyc b/src/serving/__pycache__/predictor.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..46e062cec4965e7a0b80e485fee961f75407aade Binary files /dev/null and b/src/serving/__pycache__/predictor.cpython-312.pyc differ diff --git a/src/serving/ab_router.py b/src/serving/ab_router.py new file mode 100644 index 0000000000000000000000000000000000000000..168bf06c2eac8b5635a125a90ad627e76fb13a10 --- /dev/null +++ b/src/serving/ab_router.py @@ -0,0 +1,88 @@ +import random +from dataclasses import dataclass + +from src.serving.predictor import PredictionResult, Predictor +from src.utils.settings import settings + + +@dataclass +class ABStats: + model_name: str + request_count: int = 0 + total_latency_ms: float = 0.0 + oos_count: int = 0 + confidence_sum: float = 0.0 + + def record(self, result: PredictionResult) -> None: + self.request_count += 1 + self.total_latency_ms += result.latency_ms + self.confidence_sum += result.confidence + if result.is_oos: + self.oos_count += 1 + + def summary(self) -> dict: + if self.request_count == 0: + return { + "model_name": self.model_name, + "request_count": 0, + "avg_latency_ms": 0.0, + "oos_rate": 0.0, + "avg_confidence": 0.0, + } + return { + "model_name": self.model_name, + "request_count": self.request_count, + "avg_latency_ms": round(self.total_latency_ms / self.request_count, 3), + "oos_rate": round(self.oos_count / self.request_count, 4), + "avg_confidence": round(self.confidence_sum / self.request_count, 4), + } + + +class ABRouter: + def __init__( + self, + model_a: str | None = None, + model_b: str | None = None, + split: float | None = None, + ): + self.model_a_name = model_a or settings.ab_model_a + self.model_b_name = model_b or settings.ab_model_b + self.split = split if split is not None else settings.ab_split + + self.predictor_a = Predictor(self.model_a_name) + self.predictor_b = Predictor(self.model_b_name) + + self.stats_a = ABStats(model_name=self.model_a_name) + self.stats_b = ABStats(model_name=self.model_b_name) + + def route(self, text: str) -> tuple[PredictionResult, str]: + if random.random() < self.split: + result = self.predictor_b.predict(text) + self.stats_b.record(result) + return result, "B" + result = self.predictor_a.predict(text) + self.stats_a.record(result) + return result, "A" + + def predict(self, text: str) -> dict: + result, variant = self.route(text) + return { + "intent": result.intent, + "confidence": result.confidence, + "top5": result.top5, + "latency_ms": result.latency_ms, + "is_oos": result.is_oos, + "model_used": result.model_used, + "ab_variant": variant, + } + + def get_stats(self) -> dict: + return { + "model_a": self.stats_a.summary(), + "model_b": self.stats_b.summary(), + "split": self.split, + } + + def reset_stats(self) -> None: + self.stats_a = ABStats(model_name=self.model_a_name) + self.stats_b = ABStats(model_name=self.model_b_name) diff --git a/src/serving/predictor.py b/src/serving/predictor.py new file mode 100644 index 0000000000000000000000000000000000000000..716ebd333789e6663835787cf2176e4e396f2222 --- /dev/null +++ b/src/serving/predictor.py @@ -0,0 +1,158 @@ +import time +from dataclasses import dataclass, field + +import numpy as np +import torch + +from src.data.preprocessor import clean_text, load_label_map +from src.features.tfidf import load_vectorizer, transform +from src.models.classical import LogisticRegressionModel, SVMModel +from src.models.neural import LSTMModel, RNNModel, TextCNN, Vocabulary +from src.models.transformer import TransformerModel +from src.utils.config import load_config +from src.utils.settings import settings + +VECTORIZER_PATH = "artifacts/vectorizers/tfidf.pkl" +VOCAB_PATH = "artifacts/models/vocab.pkl" +LOGREG_PATH = "artifacts/models/logreg.pkl" +SVM_PATH = "artifacts/models/svm.pkl" +TEXTCNN_PATH = "artifacts/models/textcnn.pt" +RNN_PATH = "artifacts/models/rnn.pt" +LSTM_PATH = "artifacts/models/lstm.pt" +DISTILBERT_DIR = "artifacts/models/distilbert" +MAX_LENGTH_NN = 32 +MAX_LENGTH_HF = 128 + +SUPPORTED_MODELS = {"classical", "svm", "textcnn", "rnn", "lstm", "transformer"} + + +@dataclass +class PredictionResult: + intent: str + confidence: float + top5: list[dict] = field(default_factory=list) + latency_ms: float = 0.0 + is_oos: bool = False + model_used: str = "" + + +class Predictor: + def __init__(self, model_type: str): + if model_type not in SUPPORTED_MODELS: + raise ValueError(f"Unknown model_type: {model_type}, must be one of {SUPPORTED_MODELS}") + + self.model_type = model_type + self.label_map = load_label_map() + self.id_to_label = {v: k for k, v in self.label_map.items()} + self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + + self.vectorizer = None + self._vocab = None + self._model = None + self._tokenizer = None + + self.load() + + def load(self) -> None: + if self.model_type in ("classical", "svm"): + self._vectorizer = load_vectorizer(VECTORIZER_PATH) + if self.model_type == "classical": + self._model = LogisticRegressionModel() + self._model.load(LOGREG_PATH) + else: + self._model = SVMModel() + self._model.load(SVM_PATH) + + elif self.model_type in ("textcnn", "rnn", "lstm"): + self._vocab = Vocabulary.load(VOCAB_PATH) + neural_config = load_config("neural") + model_cfg = neural_config["model"][self.model_type] + num_classes = len(self.label_map) + + if self.model_type == "textcnn": + self._model = TextCNN( + vocab_size=len(self._vocab), + embedding_dim=model_cfg["embedding_dim"], + num_filters=model_cfg["num_filters"], + kernel_sizes=model_cfg["kernel_sizes"], + num_classes=num_classes, + dropout=model_cfg["dropout"], + ) + self._model.load(TEXTCNN_PATH) + elif self.model_type == "rnn": + self._model = RNNModel( + vocab_size=len(self._vocab), + embedding_dim=model_cfg["embedding_dim"], + hidden_dim=model_cfg["hidden_dim"], + num_layers=model_cfg["num_layers"], + num_classes=num_classes, + dropout=model_cfg["dropout"], + ) + self._model.load(RNN_PATH) + else: + self._model = LSTMModel( + vocab_size=len(self._vocab), + embedding_dim=model_cfg["embedding_dim"], + hidden_dim=model_cfg["hidden_dim"], + num_layers=model_cfg["num_layers"], + num_classes=num_classes, + dropout=model_cfg["dropout"], + ) + self._model.load(LSTM_PATH) + + self._model.to(self.device) + self._model.eval() + + elif self.model_type == "transformer": + self._model = TransformerModel( + model_name=DISTILBERT_DIR, + num_labels=len(self.label_map), + ) + self._model.model.to(self.device) + self._model.model.eval() + + def _predict_proba(self, text: str) -> np.ndarray: + cleaned = clean_text(text) + + if self.model_type in ("classical", "svm"): + X = transform(self._vectorizer, [cleaned]) + if self.model_type == "classical": + return self._model.predict_proba(X)[0] + scores = self._model.model.decision_function(X)[0] + exp_scores = np.exp(scores - np.max(scores)) + return exp_scores / exp_scores.sum() + + if self.model_type in ("textcnn", "rnn", "lstm"): + encoded = self._vocab.encode(cleaned, MAX_LENGTH_NN) + tensor = torch.tensor([encoded], dtype=torch.long).to(self.device) + return self._model.predict_proba(tensor)[0] + + if self.model_type == "transformer": + return self._model.predict_proba([cleaned], MAX_LENGTH_HF)[0] + + raise ValueError(f"unsupported model_type: {self.model_type}") + + def predict(self, text: str) -> PredictionResult: + start = time.perf_counter() + probs = self._predict_proba(text) + latency_ms = (time.perf_counter() - start) * 1000 + + top_indices = np.argsort(probs)[-5:][::-1] + top5 = [{"intent": self.id_to_label[idx], "confidence": round(float(probs[idx]), 4)} for idx in top_indices] + + best_idx = int(top_indices[0]) + intent = self.id_to_label[best_idx] + confidence = float(probs[best_idx]) + is_oos = confidence < settings.oos_threshold + + return PredictionResult( + intent=intent, + confidence=round(confidence, 4), + top5=top5, + latency_ms=round(latency_ms, 3), + is_oos=is_oos, + model_used=self.model_type, + ) + + def predict_batch(self, texts: list[str]) -> list[PredictionResult]: + return [self.predict(text) for text in texts] diff --git a/src/storage/__init__.py b/src/storage/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/src/storage/__pycache__/__init__.cpython-312.pyc b/src/storage/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..946e227ec30ef4fdf7d8fdbca3fbe1aa95328fa2 Binary files /dev/null and b/src/storage/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/storage/__pycache__/s3.cpython-312.pyc b/src/storage/__pycache__/s3.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..3d4ee2d2fc9dd64b08dfd56d7a75b0c6432f541b Binary files /dev/null and b/src/storage/__pycache__/s3.cpython-312.pyc differ diff --git a/src/storage/s3.py b/src/storage/s3.py new file mode 100644 index 0000000000000000000000000000000000000000..4c07605ce30829f3af7105d2c48a1131e0b2afe3 --- /dev/null +++ b/src/storage/s3.py @@ -0,0 +1,46 @@ +from pathlib import Path + +import boto3 +from botocore.exceptions import ClientError + +from src.utils.settings import settings + + +def _get_client(): + return boto3.client( + "s3", + aws_access_key_id=settings.aws_access_key_id, + aws_secret_access_key=settings.aws_secret_access_key, + region_name=settings.aws_region, + ) + + +def upload_artifact(local_path: str, s3_key: str) -> None: + client = _get_client() + client.upload_file(local_path, settings.s3_bucket, s3_key) + print(f"uploaded {local_path} -> s3://{settings.s3_bucket}/{s3_key}") + + +def download_artifact(s3_key: str, local_path: str) -> None: + client = _get_client() + path = Path(local_path) + path.parent.mkdir(parents=True, exist_ok=True) + client.download_file(settings.s3_bucket, s3_key, local_path) + print(f"downloaded s3://{settings.s3_bucket}/{s3_key} -> {local_path}") + + +def artifact_exists(s3_key: str) -> bool: + client = _get_client() + try: + client.head_object(Bucket=settings.s3_bucket, Key=s3_key) + return True + except ClientError: + return False + + +def list_artifacts(prefix: str) -> list[str]: + client = _get_client() + response = client.list_objects_v2(Bucket=settings.s3_bucket, Prefix=prefix) + if "Contents" not in response: + return [] + return [obj["Key"] for obj in response["Contents"]] diff --git a/src/training/__init__.py b/src/training/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/src/training/register_models.py b/src/training/register_models.py new file mode 100644 index 0000000000000000000000000000000000000000..ccf0a2273f40d66cc83dabeafe86e359c3f6230e --- /dev/null +++ b/src/training/register_models.py @@ -0,0 +1,92 @@ +import mlflow +import mlflow.sklearn +import mlflow.transformers + +# from src.data.preprocessor import load_label_map +from src.features.tfidf import load_vectorizer, transform +from src.models.classical import LogisticRegressionModel, SVMModel +from src.models.transformer import TransformerModel +from src.utils.mlflow_utils import get_or_create_experiment, register_model +from src.utils.settings import settings + +LOGREG_PATH = "artifacts/models/logreg.pkl" +SVM_PATH = "artifacts/models/svm.pkl" +DISTILBERT_DIR = "artifacts/models/distilbert" +VECTORIZER_PATH = "artifacts/vectorizers/tfidf.pkl" + +REGISTRY_NAME = "intent-classifier" + + +def register_sklearn_model(model_obj, model_type: str, sample_input, experiment_id: str) -> None: + with mlflow.start_run(experiment_id=experiment_id, run_name=f"register-{model_type}") as run: + mlflow.sklearn.log_model( + sk_model=model_obj.model, + artifact_path="model", + input_example=sample_input, + ) + mlflow.set_tag("model_type", model_type) + register_model( + run_id=run.info.run_id, + artifact_path="model", + model_name=f"{REGISTRY_NAME}-{model_type}", + alias="challenger", + ) + + +def register_transformer_model(experiment_id: str) -> None: + transformer = TransformerModel(model_name=DISTILBERT_DIR, num_labels=151) + + with mlflow.start_run(experiment_id=experiment_id, run_name="register-distilbert") as run: + mlflow.transformers.log_model( + transformers_model={ + "model": transformer.model, + "tokenizer": transformer.tokenizer, + }, + artifact_path="model", + task="text-classification", + ) + mlflow.set_tag("model_type", "distilbert") + register_model( + run_id=run.info.run_id, + artifact_path="model", + model_name=f"{REGISTRY_NAME}-distilbert", + alias="champion", + ) + + +def main(): + mlflow.set_tracking_uri(settings.mlflow_tracking_uri) + experiment_id = get_or_create_experiment("intent-classifier") + + # label_map = load_label_map() + + print("registering logistic regression...") + vectorizer = load_vectorizer(VECTORIZER_PATH) + logreg = LogisticRegressionModel() + logreg.load(LOGREG_PATH) + sample = transform(vectorizer, ["what is my balance"]) + register_sklearn_model(logreg, "logreg", sample, experiment_id) + + print("registering svm...") + svm = SVMModel() + svm.load(SVM_PATH) + register_sklearn_model(svm, "svm", sample, experiment_id) + + print("registering distilbert as champion...") + register_transformer_model(experiment_id) + + print("\nregistry summary:") + client = mlflow.MlflowClient() + for name in [ + f"{REGISTRY_NAME}-logreg", + f"{REGISTRY_NAME}-svm", + f"{REGISTRY_NAME}-distilbert", + ]: + versions = client.search_model_versions(f"name='{name}'") + for v in versions: + aliases = v.aliases if hasattr(v, "aliases") else [] + print(f" {name} v{v.version} aliases={aliases}") + + +if __name__ == "__main__": + main() diff --git a/src/training/train_classical.py b/src/training/train_classical.py new file mode 100644 index 0000000000000000000000000000000000000000..dd744fca32bbc61944c9a7d52a6a6a8387089e3a --- /dev/null +++ b/src/training/train_classical.py @@ -0,0 +1,215 @@ +import os + +import mlflow +import numpy as np + +from src.data.loader import load_clinc150, load_splits, save_splits +from src.data.preprocessor import preprocess +from src.evaluation.metrics import ( + compute_classification_metrics, + compute_latency, + get_classification_report, +) +from src.features.tfidf import fit_tfidf, load_vectorizer, save_vectorizer, transform +from src.models.classical import LogisticRegressionModel, SVMModel +from src.storage.s3 import upload_artifact +from src.utils.config import load_config +from src.utils.mlflow_utils import ( + get_or_create_experiment, + log_config, + log_confusion_matrix, + log_metrics, +) +from src.utils.settings import settings + +DATA_DIR = "data/raw" +LOGREG_PATH = "artifacts/models/logreg.pkl" +SVM_PATH = "artifacts/models/svm.pkl" +VECTORIZER_PATH = "artifacts/vectorizers/tfidf.pkl" + + +def get_config_value(config: dict, key: str): + kebab_key = key.replace("_", "-") + if key in config: + return config[key] + if kebab_key in config: + return config[kebab_key] + raise KeyError(key) + + +def load_or_download_data(config: dict) -> tuple: + train_path = os.path.join(DATA_DIR, "train.csv") + if os.path.exists(train_path): + print("loading data from disk...") + splits = load_splits(DATA_DIR) + else: + print("downloading CLINC150...") + splits = load_clinc150(config["data"]["subset"]) + save_splits(splits, DATA_DIR) + + processed, label_map = preprocess(splits) + return processed, label_map + + +def get_features(processed: dict, config: dict) -> tuple: + train_texts = processed["train"]["text"].tolist() + val_texts = processed["validation"]["text"].tolist() + test_texts = processed["test"]["text"].tolist() + + try: + print("loading existing vectorizer...") + vectorizer = load_vectorizer(VECTORIZER_PATH) + except FileNotFoundError: + print("fitting tfidf vectorizer...") + vectorizer = fit_tfidf(train_texts, config) + save_vectorizer(vectorizer, VECTORIZER_PATH) + + X_train = transform(vectorizer, train_texts) + X_val = transform(vectorizer, val_texts) + X_test = transform(vectorizer, test_texts) + + return X_train, X_val, X_test, vectorizer + + +def train_and_log( + model_class, + model_name: str, + save_path: str, + s3_prefix: str, + X_train, + y_train: np.ndarray, + X_val, + y_val: np.ndarray, + X_test, + y_test: np.ndarray, + label_names: list[str], + config: dict, +) -> dict: + mlflow.set_tracking_uri(settings.mlflow_tracking_uri) + experiment_id = get_or_create_experiment(config["mlflow"]["experiment_name"]) + + with mlflow.start_run( + experiment_id=experiment_id, + run_name=model_name, + ) as run: + log_config(config) + + print(f" training {model_name}...") + model_config = config["model"] + model = model_class( + C=get_config_value(model_config, "C"), + max_iter=get_config_value(model_config, "max_iter"), + random_state=get_config_value(model_config, "random_state"), + ) + model.fit(X_train, y_train) + + val_preds = model.predict(X_val) + val_metrics = compute_classification_metrics(y_val, val_preds) + val_metrics = {f"val_{k}": v for k, v in val_metrics.items()} + log_metrics(val_metrics) + + test_preds = model.predict(X_test) + test_metrics = compute_classification_metrics(y_test, test_preds) + test_metrics = {f"test_{k}": v for k, v in test_metrics.items()} + log_metrics(test_metrics) + + latency = compute_latency(model.predict, X_test[:100]) + log_metrics(latency) + + report = get_classification_report(y_test, test_preds, label_names) + report_path = f"artifacts/{model_name}_report.txt" + with open(report_path, "w") as f: + f.write(report) + mlflow.log_artifact(report_path) + + log_confusion_matrix( + y_test, + test_preds, + label_names, + save_path=f"artifacts/{model_name}_confusion_matrix.png", + ) + + model.save(save_path) + mlflow.log_artifact(save_path) + + upload_artifact(save_path, f"{s3_prefix}/{save_path.split('/')[-1]}") + upload_artifact(VECTORIZER_PATH, f"{s3_prefix}/tfidf.pkl") + + all_metrics = {**val_metrics, **test_metrics, **latency} + print(f" val accuracy : {val_metrics['val_accuracy']}") + print(f" test accuracy : {test_metrics['test_accuracy']}") + print(f" test macro_f1 : {test_metrics['test_macro_f1']}") + print(f" latency p50 : {latency['latency_p50_ms']}ms") + print(f" run id : {run.info.run_id}") + + return all_metrics + + +def main(): + config = load_config("classical") + + processed, label_map = load_or_download_data(config) + + X_train, X_val, X_test, vectorizer = get_features(processed, config) + + y_train = processed["train"]["label_id"].values + y_val = processed["validation"]["label_id"].values + y_test = processed["test"]["label_id"].values + + id_to_label = {v: k for k, v in label_map.items()} + label_names = [id_to_label[i] for i in range(len(label_map))] + + print("\ntraining logistic regression...") + logreg_metrics = train_and_log( + model_class=LogisticRegressionModel, + model_name="logistic-regression", + save_path=LOGREG_PATH, + s3_prefix=config["s3"]["prefix"], + X_train=X_train, + y_train=y_train, + X_val=X_val, + y_val=y_val, + X_test=X_test, + y_test=y_test, + label_names=label_names, + config=config, + ) + + config["model"]["type"] = "svm" + print("\ntraining svm...") + svm_metrics = train_and_log( + model_class=SVMModel, + model_name="svm", + save_path=SVM_PATH, + s3_prefix=config["s3"]["prefix"], + X_train=X_train, + y_train=y_train, + X_val=X_val, + y_val=y_val, + X_test=X_test, + y_test=y_test, + label_names=label_names, + config=config, + ) + + print("\nmodel comparison:") + print(f"{'model':<25} {'val_acc':<12} {'test_acc':<12} {'macro_f1':<12} {'p50_ms'}") + print("-" * 70) + print( + f"{'logistic-regression':<25} " + f"{logreg_metrics['val_accuracy']:<12} " + f"{logreg_metrics['test_accuracy']:<12} " + f"{logreg_metrics['test_macro_f1']:<12} " + f"{logreg_metrics['latency_p50_ms']}ms" + ) + print( + f"{'svm':<25} " + f"{svm_metrics['val_accuracy']:<12} " + f"{svm_metrics['test_accuracy']:<12} " + f"{svm_metrics['test_macro_f1']:<12} " + f"{svm_metrics['latency_p50_ms']}ms" + ) + + +if __name__ == "__main__": + main() diff --git a/src/training/train_neural.py b/src/training/train_neural.py new file mode 100644 index 0000000000000000000000000000000000000000..119ae8a4cb96750484b7936dda195612d4e14271 --- /dev/null +++ b/src/training/train_neural.py @@ -0,0 +1,333 @@ +import os + +import mlflow +import numpy as np +import torch +import torch.nn as nn +from torch.utils.data import DataLoader + +from src.data.loader import load_clinc150, load_splits, save_splits +from src.data.preprocessor import preprocess +from src.evaluation.metrics import ( + compute_classification_metrics, + compute_latency, + get_classification_report, +) +from src.models.neural import IntentDatasetNN, LSTMModel, RNNModel, TextCNN, Vocabulary +from src.storage.s3 import upload_artifact +from src.utils.config import load_config +from src.utils.mlflow_utils import ( + get_or_create_experiment, + log_confusion_matrix, + log_metrics, +) +from src.utils.settings import settings + +DATA_DIR = "data/raw" +VOCAB_PATH = "artifacts/models/vocab.pkl" +MAX_LENGTH = 32 + + +def load_or_download_data(config: dict) -> tuple: + train_path = os.path.join(DATA_DIR, "train.csv") + if os.path.exists(train_path): + print("loading data from disk...") + splits = load_splits(DATA_DIR) + else: + print("downloading CLINC150...") + splits = load_clinc150(config["data"]["subset"]) + save_splits(splits, DATA_DIR) + + processed, label_map = preprocess(splits) + return processed, label_map + + +def build_dataloaders( + processed: dict, + vocab: Vocabulary, + batch_size: int, +) -> tuple[DataLoader, DataLoader, DataLoader]: + train_dataset = IntentDatasetNN( + processed["train"]["text"].tolist(), + processed["train"]["label_id"].tolist(), + vocab, + MAX_LENGTH, + ) + val_dataset = IntentDatasetNN( + processed["validation"]["text"].tolist(), + processed["validation"]["label_id"].tolist(), + vocab, + MAX_LENGTH, + ) + test_dataset = IntentDatasetNN( + processed["test"]["text"].tolist(), + processed["test"]["label_id"].tolist(), + vocab, + MAX_LENGTH, + ) + + train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True) + val_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False) + test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False) + + return train_loader, val_loader, test_loader + + +def run_epoch( + model: nn.Module, + loader: DataLoader, + criterion: nn.CrossEntropyLoss, + optimizer: torch.optim.Optimizer | None, + device: torch.device, + training: bool, + grad_clip: float | None = None, +) -> tuple[float, float]: + model.train() if training else model.eval() + + total_loss = 0.0 + correct = 0 + total = 0 + + context = torch.enable_grad() if training else torch.no_grad() + with context: + for inputs, labels in loader: + inputs = inputs.to(device) + labels = labels.to(device) + + logits = model(inputs) + loss = criterion(logits, labels) + + if training and optimizer is not None: + optimizer.zero_grad() + loss.backward() + if grad_clip is not None: + nn.utils.clip_grad_norm_(model.parameters(), grad_clip) + optimizer.step() + + total_loss += loss.item() * len(labels) + correct += (logits.argmax(dim=1) == labels).sum().item() + total += len(labels) + + return total_loss / total, correct / total + + +def predict_all( + model: nn.Module, + loader: DataLoader, + device: torch.device, +) -> tuple[np.ndarray, np.ndarray]: + model.eval() + all_preds = [] + all_labels = [] + + with torch.no_grad(): + for inputs, labels in loader: + inputs = inputs.to(device) + logits = model(inputs) + preds = logits.argmax(dim=1).cpu().numpy() + all_preds.extend(preds) + all_labels.extend(labels.numpy()) + + return np.array(all_preds), np.array(all_labels) + + +def build_model(model_type: str, model_cfg: dict, vocab_size: int, num_classes: int) -> nn.Module: + if model_type == "textcnn": + return TextCNN( + vocab_size=vocab_size, + embedding_dim=model_cfg["embedding_dim"], + num_filters=model_cfg["num_filters"], + kernel_sizes=model_cfg["kernel_sizes"], + num_classes=num_classes, + dropout=model_cfg["dropout"], + ) + if model_type == "rnn": + return RNNModel( + vocab_size=vocab_size, + embedding_dim=model_cfg["embedding_dim"], + hidden_dim=model_cfg["hidden_dim"], + num_layers=model_cfg["num_layers"], + num_classes=num_classes, + dropout=model_cfg["dropout"], + ) + if model_type == "lstm": + return LSTMModel( + vocab_size=vocab_size, + embedding_dim=model_cfg["embedding_dim"], + hidden_dim=model_cfg["hidden_dim"], + num_layers=model_cfg["num_layers"], + num_classes=num_classes, + dropout=model_cfg["dropout"], + ) + raise ValueError(f"unknown model type: {model_type}") + + +def train_model( + model_type: str, + config: dict, + train_loader: DataLoader, + val_loader: DataLoader, + test_loader: DataLoader, + vocab: Vocabulary, + label_map: dict, + device: torch.device, +) -> dict: + num_classes = len(label_map) + model_cfg = config["model"][model_type] + save_path = f"artifacts/models/{model_type}.pt" + grad_clip = config["training"].get("grad_clip", None) + + model = build_model(model_type, model_cfg, len(vocab), num_classes).to(device) + print(f"\n{model_type} parameters: {sum(p.numel() for p in model.parameters()):,}") + + criterion = nn.CrossEntropyLoss() + optimizer = torch.optim.Adam(model.parameters(), lr=config["training"]["learning_rate"]) + + mlflow.set_tracking_uri(settings.mlflow_tracking_uri) + experiment_id = get_or_create_experiment(config["mlflow"]["experiment_name"]) + + with mlflow.start_run(experiment_id=experiment_id, run_name=model_type) as run: + flat_config = { + "model.type": model_type, + **{f"model.{k}": v for k, v in model_cfg.items()}, + **{f"training.{k}": v for k, v in config["training"].items()}, + } + mlflow.log_params(flat_config) + + epochs = config["training"]["epochs"] + patience = config["training"]["patience"] + best_val_loss = float("inf") + epochs_no_improve = 0 + + print(f"training {model_type} for up to {epochs} epochs (patience={patience}, grad_clip={grad_clip})...") + print(f"{'epoch':<8} {'train_loss':<14} {'train_acc':<14} {'val_loss':<14} {'val_acc'}") + print("-" * 65) + + for epoch in range(1, epochs + 1): + train_loss, train_acc = run_epoch( + model, + train_loader, + criterion, + optimizer, + device, + training=True, + grad_clip=grad_clip, + ) + val_loss, val_acc = run_epoch( + model, + val_loader, + criterion, + None, + device, + training=False, + ) + + log_metrics( + { + "train_loss": round(train_loss, 4), + "train_accuracy": round(train_acc, 4), + "val_loss": round(val_loss, 4), + "val_accuracy": round(val_acc, 4), + }, + step=epoch, + ) + + print(f"{epoch:<8} {train_loss:<14.4f} {train_acc:<14.4f} {val_loss:<14.4f} {val_acc:.4f}") + + if val_loss < best_val_loss: + best_val_loss = val_loss + epochs_no_improve = 0 + model.save(save_path) + else: + epochs_no_improve += 1 + if epochs_no_improve >= patience: + print(f"early stopping at epoch {epoch}") + break + + print("loading best checkpoint...") + model.load(save_path) + + id_to_label = {v: k for k, v in label_map.items()} + label_names = [id_to_label[i] for i in range(num_classes)] + + test_preds, test_labels = predict_all(model, test_loader, device) + test_metrics = compute_classification_metrics(test_labels, test_preds) + test_metrics_logged = {f"test_{k}": v for k, v in test_metrics.items()} + log_metrics(test_metrics_logged) + + def predict_fn(loader): + predict_all(model, loader, device) + + latency = compute_latency(predict_fn, test_loader, n_runs=50) + log_metrics(latency) + + report = get_classification_report(test_labels, test_preds, label_names) + report_path = f"artifacts/{model_type}_report.txt" + with open(report_path, "w") as f: + f.write(report) + mlflow.log_artifact(report_path) + + log_confusion_matrix( + test_labels, + test_preds, + label_names, + save_path=f"artifacts/{model_type}_confusion_matrix.png", + ) + + mlflow.log_artifact(save_path) + upload_artifact(save_path, f"{config['s3']['prefix']}/{model_type}.pt") + + print(f" test accuracy : {test_metrics['accuracy']}") + print(f" test macro_f1 : {test_metrics['macro_f1']}") + print(f" latency p50 : {latency['latency_p50_ms']}ms") + print(f" run id : {run.info.run_id}") + + return {**test_metrics_logged, **latency} + + +def main(): + config = load_config("neural") + torch.manual_seed(config["training"]["random_state"]) + + device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + print(f"device: {device}") + + processed, label_map = load_or_download_data(config) + + print("building vocabulary on train set only...") + vocab = Vocabulary() + vocab.build(processed["train"]["text"].tolist()) + vocab.save(VOCAB_PATH) + print(f"vocabulary size: {len(vocab)}") + + batch_size = config["training"]["batch_size"] + train_loader, val_loader, test_loader = build_dataloaders(processed, vocab, batch_size) + + all_results = {} + for model_type in ["textcnn", "rnn", "lstm"]: + metrics = train_model( + model_type=model_type, + config=config, + train_loader=train_loader, + val_loader=val_loader, + test_loader=test_loader, + vocab=vocab, + label_map=label_map, + device=device, + ) + all_results[model_type] = metrics + + print("\n\nfinal comparison:") + print(f"{'model':<12} {'test_acc':<12} {'macro_f1':<12} {'p50_ms'}") + print("-" * 50) + for model_type, metrics in all_results.items(): + print( + f"{model_type:<12} " + f"{metrics['test_accuracy']:<12} " + f"{metrics['test_macro_f1']:<12} " + f"{metrics['latency_p50_ms']}ms" + ) + + +if __name__ == "__main__": + main() diff --git a/src/training/train_transformer.py b/src/training/train_transformer.py new file mode 100644 index 0000000000000000000000000000000000000000..0c243968e7bf965cc41fe59aa7813655671128ed --- /dev/null +++ b/src/training/train_transformer.py @@ -0,0 +1,219 @@ +import os + +import mlflow +import numpy as np +import torch +from torch.utils.data import DataLoader +from transformers import EarlyStoppingCallback, Trainer, TrainingArguments + +from src.data.loader import load_clinc150, load_splits, save_splits +from src.data.preprocessor import preprocess +from src.evaluation.metrics import ( + compute_classification_metrics, + compute_latency, + get_classification_report, +) +from src.models.transformer import IntentDatasetHF, TransformerModel +from src.storage.s3 import upload_artifact +from src.utils.config import load_config +from src.utils.mlflow_utils import ( + get_or_create_experiment, + log_confusion_matrix, + log_metrics, +) +from src.utils.settings import settings + +DATA_DIR = "data/raw" +MODEL_DIR = "artifacts/models/distilbert" + + +def load_or_download_data(config: dict) -> tuple: + train_path = os.path.join(DATA_DIR, "train.csv") + if os.path.exists(train_path): + print("loading data from disk...") + splits = load_splits(DATA_DIR) + else: + print("downloading CLINC150...") + splits = load_clinc150(config["data"]["subset"]) + save_splits(splits, DATA_DIR) + + processed, label_map = preprocess(splits) + return processed, label_map + + +def compute_metrics_hf(eval_pred) -> dict: + logits, labels = eval_pred + preds = np.argmax(logits, axis=1) + metrics = compute_classification_metrics(labels, preds) + return { + "accuracy": metrics["accuracy"], + "macro_f1": metrics["macro_f1"], + } + + +def predict_all( + transformer: TransformerModel, + dataset: IntentDatasetHF, + batch_size: int, + device: torch.device, +) -> tuple[np.ndarray, np.ndarray]: + transformer.model.eval() + transformer.model.to(device) + + loader = DataLoader(dataset, batch_size=batch_size, shuffle=False) + all_preds = [] + all_labels = [] + + with torch.no_grad(): + for batch in loader: + input_ids = batch["input_ids"].to(device) + attention_mask = batch["attention_mask"].to(device) + labels = batch["labels"] + + outputs = transformer.model(input_ids=input_ids, attention_mask=attention_mask) + preds = outputs.logits.argmax(dim=1).cpu().numpy() + all_preds.extend(preds) + all_labels.extend(labels.numpy()) + + return np.array(all_preds), np.array(all_labels) + + +def main(): + config = load_config("transformer") + torch.manual_seed(config["training"]["random_state"]) + + device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + print(f"device: {device}") + + processed, label_map = load_or_download_data(config) + + model_cfg = config["model"] + training_cfg = config["training"] + max_length = config["data"]["max_length"] + + print(f"loading {model_cfg['name']}...") + transformer = TransformerModel( + model_name=model_cfg["name"], + num_labels=model_cfg["num_labels"], + dropout=model_cfg["dropout"], + ) + + print("tokenizing datasets...") + train_dataset = IntentDatasetHF( + processed["train"]["text"].tolist(), + processed["train"]["label_id"].tolist(), + transformer.tokenizer, + max_length, + ) + val_dataset = IntentDatasetHF( + processed["validation"]["text"].tolist(), + processed["validation"]["label_id"].tolist(), + transformer.tokenizer, + max_length, + ) + test_dataset = IntentDatasetHF( + processed["test"]["text"].tolist(), + processed["test"]["label_id"].tolist(), + transformer.tokenizer, + max_length, + ) + + mlflow.set_tracking_uri(settings.mlflow_tracking_uri) + experiment_id = get_or_create_experiment(config["mlflow"]["experiment_name"]) + + with mlflow.start_run( + experiment_id=experiment_id, + run_name=config["mlflow"]["run_name"], + ) as run: + mlflow.log_params( + { + "model.name": model_cfg["name"], + "model.num_labels": model_cfg["num_labels"], + "model.dropout": model_cfg["dropout"], + "training.epochs": training_cfg["epochs"], + "training.batch_size": training_cfg["batch_size"], + "training.learning_rate": training_cfg["learning_rate"], + "training.warmup_steps": training_cfg["warmup_steps"], + "training.weight_decay": training_cfg["weight_decay"], + "data.max_length": max_length, + } + ) + + training_args = TrainingArguments( + output_dir=MODEL_DIR, + num_train_epochs=training_cfg["epochs"], + per_device_train_batch_size=training_cfg["batch_size"], + per_device_eval_batch_size=training_cfg["batch_size"], + learning_rate=training_cfg["learning_rate"], + warmup_steps=training_cfg["warmup_steps"], + weight_decay=training_cfg["weight_decay"], + eval_strategy="epoch", + save_strategy="epoch", + load_best_model_at_end=True, + metric_for_best_model="macro_f1", + greater_is_better=True, + logging_steps=50, + report_to="none", + seed=training_cfg["random_state"], + ) + + trainer = Trainer( + model=transformer.model, + args=training_args, + train_dataset=train_dataset, + eval_dataset=val_dataset, + compute_metrics=compute_metrics_hf, + callbacks=[EarlyStoppingCallback(early_stopping_patience=2)], + ) + + print("\nfine-tuning distilbert...") + trainer.train() + + print("\nloading best checkpoint...") + transformer.save(MODEL_DIR) + + id_to_label = {v: k for k, v in label_map.items()} + label_names = [id_to_label[i] for i in range(len(label_map))] + + print("evaluating on test set...") + test_preds, test_labels = predict_all(transformer, test_dataset, training_cfg["batch_size"], device) + + test_metrics = compute_classification_metrics(test_labels, test_preds) + test_metrics_logged = {f"test_{k}": v for k, v in test_metrics.items()} + log_metrics(test_metrics_logged) + + def predict_fn(dataset): + predict_all(transformer, dataset, training_cfg["batch_size"], device) + + latency = compute_latency(predict_fn, test_dataset, n_runs=20) + log_metrics(latency) + + report = get_classification_report(test_labels, test_preds, label_names) + report_path = "artifacts/distilbert_report.txt" + with open(report_path, "w") as f: + f.write(report) + mlflow.log_artifact(report_path) + + log_confusion_matrix( + test_labels, + test_preds, + label_names, + save_path="artifacts/distilbert_confusion_matrix.png", + ) + + mlflow.log_artifact(MODEL_DIR) + + print("uploading to S3...") + for fname in os.listdir(MODEL_DIR): + local_path = os.path.join(MODEL_DIR, fname) + if os.path.isfile(local_path): + upload_artifact(local_path, f"{config['s3']['prefix']}/distilbert/{fname}") + + print(f"\ntest accuracy : {test_metrics['accuracy']}") + print(f"test macro_f1 : {test_metrics['macro_f1']}") + print(f"latency p50 : {latency['latency_p50_ms']}ms") + print(f"run id : {run.info.run_id}") + + +if __name__ == "__main__": + main() diff --git a/src/utils/__init__.py b/src/utils/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/src/utils/__pycache__/__init__.cpython-312.pyc b/src/utils/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..ae39c45721dd2b4f7b6c9db302f9d4103bfcd024 Binary files /dev/null and b/src/utils/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/utils/__pycache__/config.cpython-312.pyc b/src/utils/__pycache__/config.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..84fb5e1d7f3fb7056380f508bd37fb687d849ec1 Binary files /dev/null and b/src/utils/__pycache__/config.cpython-312.pyc differ diff --git a/src/utils/__pycache__/mlflow_utils.cpython-312.pyc b/src/utils/__pycache__/mlflow_utils.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..bd6312181a5f89f455c91c565f026bae84928d27 Binary files /dev/null and b/src/utils/__pycache__/mlflow_utils.cpython-312.pyc differ diff --git a/src/utils/__pycache__/settings.cpython-312.pyc b/src/utils/__pycache__/settings.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..b5234d3f5818d174270b28d55ecc9a23978d5b02 Binary files /dev/null and b/src/utils/__pycache__/settings.cpython-312.pyc differ diff --git a/src/utils/config.py b/src/utils/config.py new file mode 100644 index 0000000000000000000000000000000000000000..2ae3ee4f885a6dfd19bdb10c005f620864b0569e --- /dev/null +++ b/src/utils/config.py @@ -0,0 +1,11 @@ +from pathlib import Path + +import yaml + + +def load_config(config_name: str) -> dict: + config_path = Path("configs") / f"{config_name}.yaml" + if not config_path.exists(): + raise FileNotFoundError(f"Config not found: {config_path}") + with open(config_path) as f: + return yaml.safe_load(f) diff --git a/src/utils/mlflow_utils.py b/src/utils/mlflow_utils.py new file mode 100644 index 0000000000000000000000000000000000000000..0271aa9397a7e2ba096e8023687a91a480c22c9c --- /dev/null +++ b/src/utils/mlflow_utils.py @@ -0,0 +1,71 @@ +from pathlib import Path + +import matplotlib.pyplot as plt +import mlflow +import numpy as np +from sklearn.metrics import ConfusionMatrixDisplay, confusion_matrix + + +def get_or_create_experiment(name: str) -> str: + experiment = mlflow.get_experiment_by_name(name) + if experiment is None: + return mlflow.create_experiment(name) + return experiment.experiment_id + + +def log_config(config: dict) -> None: + flat = {} + for section, values in config.items(): + if isinstance(values, dict): + for k, v in values.items(): + flat[f"{section}.{k}"] = v + else: + flat[section] = values + mlflow.log_params(flat) + + +def log_metrics(metrics: dict, step: int | None = None) -> None: + mlflow.log_metrics(metrics, step=step) + + +def log_confusion_matrix( + y_true: np.ndarray, + y_pred: np.ndarray, + labels: list[str], + save_path: str = "artifacts/confusion_matrix.png", +) -> None: + path = Path(save_path) + path.parent.mkdir(parents=True, exist_ok=True) + + cm = confusion_matrix(y_true, y_pred) + fig, ax = plt.subplots(figsize=(20, 20)) + disp = ConfusionMatrixDisplay(confusion_matrix=cm, display_labels=labels) + disp.plot(ax=ax, xticks_rotation=90, colorbar=False) + ax.set_title("Confusion Matrix") + plt.tight_layout() + plt.savefig(path, dpi=100) + plt.close() + + mlflow.log_artifact(str(path)) + + +def register_model(run_id: str, artifact_path: str, model_name: str, alias: str = "champion") -> None: + model_uri = f"runs:/{run_id}/{artifact_path}" + result = mlflow.register_model(model_uri, model_name) + client = mlflow.MlflowClient() + client.set_registered_model_alias( + name=model_name, + alias=alias, + version=result.version, + ) + print(f" registered {model_name} v{result.version} with alias '{alias}'") + + +def load_registered_model(model_name: str, alias: str = "champion"): + model_uri = f"models:/{model_name}@{alias}" + return mlflow.pyfunc.load_model(model_uri) + + +def get_model_version_by_alias(model_name: str, alias: str = "champion"): + client = mlflow.MlflowClient() + return client.get_model_version_by_alias(model_name, alias) diff --git a/src/utils/settings.py b/src/utils/settings.py new file mode 100644 index 0000000000000000000000000000000000000000..1ac206fe301742e7e9298d27940fa05d47130e51 --- /dev/null +++ b/src/utils/settings.py @@ -0,0 +1,22 @@ +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8") + + aws_access_key_id: str = "" + aws_secret_access_key: str = "" + aws_region: str = "us-east-1" + s3_bucket: str = "intent-classifier-artifacts" + + mlflow_tracking_uri: str = "http://localhost:5000" + + model_type: str = "transformer" + ab_model_a: str = "classical" + ab_model_b: str = "transformer" + ab_split: float = 0.3 + + oos_threshold: float = 0.5 + + +settings = Settings()