MLOps流水线设计与实现

# feature_store/definitions.py
from feast import Entity, Feature, FeatureView, FileSource, ValueType
from datetime import timedelta

# Define entity
user = Entity(
    name="user_id",
    value_type=ValueType.INT64,
    description="User identifier",
)

# Define feature source
user_features_source = FileSource(
    path="data/user_features.parquet",
    timestamp_field="event_timestamp",
)

# Define feature view
user_features = FeatureView(
    name="user_features",
    entities=[user],
    ttl=timedelta(days=1),
    schema=[
        Feature(name="total_purchases", dtype=ValueType.INT64),
        Feature(name="avg_order_value", dtype=ValueType.FLOAT),
        Feature(name="days_since_last_order", dtype=ValueType.INT64),
        Feature(name="preferred_category", dtype=ValueType.STRING),
    ],
    source=user_features_source,
    online=True,
)

# Training: get historical features
from feast import FeatureStore

store = FeatureStore(repo_path="./feature_store")

# Point-in-time correct feature retrieval
training_df = store.get_historical_features(
    entity_df=entity_df,  # DataFrame with user_id and event_timestamp
    features=[
        "user_features:total_purchases",
        "user_features:avg_order_value",
        "user_features:days_since_last_order",
    ],
).to_df()

# Serving: get online features
online_features = store.get_online_features(
    features=[
        "user_features:total_purchases",
        "user_features:avg_order_value",
    ],
    entity_rows=[{"user_id": 12345}],
).to_dict()

import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, f1_score, precision_score, recall_score

# Set tracking URI
mlflow.set_tracking_uri("http://mlflow-server:5000")
mlflow.set_experiment("fraud-detection")

# Training with experiment tracking
with mlflow.start_run(run_name="rf-v2-tuned") as run:
    # Log parameters
    params = {
        "n_estimators": 200,
        "max_depth": 15,
        "min_samples_split": 5,
        "class_weight": "balanced",
    }
    mlflow.log_params(params)

    # Train model
    model = RandomForestClassifier(**params, random_state=42)
    model.fit(X_train, y_train)

    # Evaluate
    y_pred = model.predict(X_test)
    metrics = {
        "accuracy": accuracy_score(y_test, y_pred),
        "f1": f1_score(y_test, y_pred),
        "precision": precision_score(y_test, y_pred),
        "recall": recall_score(y_test, y_pred),
    }
    mlflow.log_metrics(metrics)

    # Log model with signature
    from mlflow.models import infer_signature
    signature = infer_signature(X_test, y_pred)
    mlflow.sklearn.log_model(
        model,
        artifact_path="model",
        signature=signature,
        registered_model_name="fraud-detector",
    )

    # Log additional artifacts
    mlflow.log_artifact("feature_importance.png")
    mlflow.log_dict({"features": feature_names}, "feature_config.json")

    # Log dataset info
    mlflow.log_input(
        mlflow.data.from_pandas(X_train, name="training_data"),
        context="training",
    )

    print(f"Run ID: {run.info.run_id}")
    print(f"Metrics: {metrics}")

from mlflow.tracking import MlflowClient

client = MlflowClient()

# Transition model stage
client.transition_model_version_stage(
    name="fraud-detector",
    version=3,
    stage="Production",
    archive_existing_versions=True,  # Archive previous production version
)

# Load production model
model = mlflow.pyfunc.load_model("models:/fraud-detector/Production")
predictions = model.predict(new_data)

# Compare model versions
runs = mlflow.search_runs(
    experiment_ids=["1"],
    filter_string="metrics.f1 > 0.85",
    order_by=["metrics.f1 DESC"],
    max_results=5,
)
print(runs[["run_id", "params.n_estimators", "metrics.f1"]])

# Client request
import requests
import json

data = {
    "instances": [
        {"feature1": 0.5, "feature2": 1.2, "feature3": -0.3},
        {"feature1": 0.8, "feature2": 0.1, "feature3": 0.6},
    ]
}

response = requests.post(
    "http://localhost:8501/v1/models/fraud_detector:predict",
    json=data,
)

predictions = response.json()["predictions"]

# Triton client
import tritonclient.http as httpclient
import numpy as np

client = httpclient.InferenceServerClient(url="localhost:8000")

# Prepare input
input_data = np.array([[0.5, 1.2, -0.3, ...]], dtype=np.float32)
inputs = [httpclient.InferInput("features", input_data.shape, "FP32")]
inputs[0].set_data_from_numpy(input_data)

# Inference
result = client.infer(model_name="fraud_detector", inputs=inputs)
prediction = result.as_numpy("prediction")
print(f"Prediction: {prediction}")

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import mlflow
import numpy as np

app = FastAPI(title="ML Model Serving")

# Load model at startup
model = None

@app.on_event("startup")
async def load_model():
    global model
    model = mlflow.pyfunc.load_model("models:/fraud-detector/Production")

class PredictionRequest(BaseModel):
    features: list[list[float]]

class PredictionResponse(BaseModel):
    predictions: list[float]
    model_version: str

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    try:
        input_data = np.array(request.features)
        predictions = model.predict(input_data).tolist()
        return PredictionResponse(
            predictions=predictions,
            model_version="3",
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health():
    return {"status": "healthy", "model_loaded": model is not None}

from scipy import stats
import numpy as np

def detect_drift_ks(reference_data, production_data, threshold=0.05):
    """
    Kolmogorov-Smirnov test for data drift detection.
    """
    drift_results = {}

    for feature in reference_data.columns:
        statistic, p_value = stats.ks_2samp(
            reference_data[feature],
            production_data[feature],
        )
        is_drifted = p_value < threshold
        drift_results[feature] = {
            "statistic": statistic,
            "p_value": p_value,
            "is_drifted": is_drifted,
        }

    return drift_results

def calculate_psi(expected, actual, bins=10):
    """
    Population Stability Index (PSI) for drift detection.

    PSI < 0.1: No drift
    0.1 <= PSI < 0.2: Moderate drift
    PSI >= 0.2: Significant drift
    """
    expected_percents = np.histogram(expected, bins=bins)[0] / len(expected)
    actual_percents = np.histogram(actual, bins=bins)[0] / len(actual)

    # Avoid division by zero
    expected_percents = np.clip(expected_percents, 0.001, None)
    actual_percents = np.clip(actual_percents, 0.001, None)

    psi = np.sum(
        (actual_percents - expected_percents) *
        np.log(actual_percents / expected_percents)
    )

    return psi

# Usage
drift_results = detect_drift_ks(training_data, last_week_data)
for feature, result in drift_results.items():
    if result["is_drifted"]:
        print(f"DRIFT DETECTED in {feature}: p-value={result['p_value']:.4f}")

from evidently import ColumnMapping
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, DataQualityPreset
from evidently.metrics import (
    DatasetDriftMetric,
    DataDriftTable,
    ClassificationQualityMetric,
)

# Define column mapping
column_mapping = ColumnMapping(
    target="label",
    prediction="prediction",
    numerical_features=["feature_1", "feature_2", "feature_3"],
    categorical_features=["category"],
)

# Create drift report
report = Report(metrics=[
    DatasetDriftMetric(),
    DataDriftTable(),
    ClassificationQualityMetric(),
])

report.run(
    reference_data=training_data,
    current_data=production_data,
    column_mapping=column_mapping,
)

# Save report
report.save_html("drift_report.html")

# Get results programmatically
results = report.as_dict()
dataset_drift = results["metrics"][0]["result"]["dataset_drift"]
if dataset_drift:
    print("Dataset drift detected! Consider retraining.")