Spaces:
Runtime error
Runtime error
File size: 3,004 Bytes
1130076 a00fee9 1130076 a00fee9 1130076 a00fee9 1130076 a00fee9 1130076 a00fee9 1130076 a00fee9 1130076 a00fee9 c539266 1130076 a00fee9 1130076 a00fee9 1130076 a00fee9 1130076 a00fee9 1130076 a00fee9 1130076 a00fee9 1130076 a00fee9 1130076 a00fee9 1130076 a00fee9 1130076 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 | import os
from datetime import datetime, timedelta
import mlflow
import pandas as pd
from evidently.metric_preset import DataDriftPreset, TextOverviewPreset
from evidently.report import Report
from sqlalchemy import create_engine
def main():
# Attempt to fetch database URL, fallback to sqlite for local tests
db_url = os.getenv("DATABASE_URL", "sqlite:///./test.db")
# We would normally load the reference data (e.g. from training data CSV)
# For this script, we'll assume a local path or create a dummy reference if missing
ref_path = "data/reference.csv"
if os.path.exists(ref_path):
ref_df = pd.read_csv(ref_path)
else:
print(f"Reference data not found at {ref_path}. Creating dummy reference data for testing.")
ref_df = pd.DataFrame(
{"text": ["This is great", "I hate this", "Neutral statement"], "language": ["en", "en", "en"]}
)
try:
# Load production data from the last 7 days
engine = create_engine(db_url)
seven_days_ago = datetime.now() - timedelta(days=7)
# Load directly from SQLAlchemy using pandas with parameterized query
query = "SELECT text, language FROM reviews WHERE created_at >= %(cutoff)s"
curr_df = pd.read_sql(query, engine, params={"cutoff": seven_days_ago})
except Exception as e:
print(f"Failed to fetch production data: {e}")
curr_df = pd.DataFrame(columns=["text", "language"])
if len(curr_df) < 50:
print(
f"Not enough data to run drift monitor "
f"(found {len(curr_df)} records, need at least 50). Exiting gracefully."
)
return
# Run Evidently report
print("Running Evidently drift report...")
report = Report(metrics=[DataDriftPreset(), TextOverviewPreset(column_name="text")])
report.run(reference_data=ref_df, current_data=curr_df)
# Create monitoring/reports dir if missing
os.makedirs("monitoring/reports", exist_ok=True)
report_path = f"monitoring/reports/drift_{datetime.now().strftime('%Y%m%d')}.html"
report.save_html(report_path)
print(f"Report saved to {report_path}")
# Extract drift metrics as a dict
report_dict = report.as_dict()
# Simplified check for drift (using Dataset Drift metric from DataDriftPreset)
dataset_drift = report_dict["metrics"][0]["result"]["dataset_drift"]
drift_share = report_dict["metrics"][0]["result"]["drift_share"]
if dataset_drift and drift_share > 0.3:
print(f"⚠️ Drift detected — consider retraining. Drift share: {drift_share:.2f}")
try:
mlflow.set_tracking_uri(os.getenv("MLFLOW_TRACKING_URI", "file:./mlruns"))
with mlflow.start_run(run_name="drift_monitoring"):
mlflow.log_metric("drift_share", drift_share)
mlflow.log_artifact(report_path)
except Exception as e:
print(f"Failed to log warning to MLflow: {e}")
if __name__ == "__main__":
main()
|