监控预测分布与置信度分数
您将把预测概率记录到时间序列存储中，绘制滚动平均置信度，并在平均置信度低于部署阈值时发出标记。
监控预测分布与置信度分数 是 CoddyKit 上的免费 Machine Learning Academy 课时。 这是第 3 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 Machine Learning Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Machine Learning Academy 课程共包含 4 节课。
本课时的部分内容尚未翻译，以英文显示。
Why Monitor Predictions, Not Just Inputs?
Input feature monitoring detects data drift but requires monitoring every feature. Prediction monitoring provides a single integrated signal: if any combination of input changes causes the model to produce different outputs, it will show up in the prediction distribution — even if no individual feature passes its drift threshold. Monitoring predictions is complementary to input monitoring: it catches what input monitoring misses by looking at the model's integrated response to all inputs together.
Logging Predictions in Production
The foundation of prediction monitoring is a prediction log: every inference request's features, predicted label, predicted probabilities, and timestamp stored to a database or file. This log enables retrospective analysis when drift or degradation is detected. Design the schema to include request ID (for joining with ground-truth labels when they arrive), model version, and latency alongside the prediction details.
import datetime
import json
import os
import torch
import torch.nn.functional as F
PREDICTION_LOG = '/tmp/prediction_log.jsonl'
def predict_and_log(features, model, model_version='v1.2.3'):
import numpy as np
with torch.no_grad():
logits = model(torch.tensor(features, dtype=torch.float32).unsqueeze(0))
probs = F.softmax(logits, dim=1).squeeze().numpy()
pred_class = int(probs.argmax())
confidence = float(probs.max())
record = {
'timestamp': datetime.datetime.utcnow().isoformat(),
'model_version': model_version,
'predicted_class': pred_class,
'confidence': round(confidence, 4),
'probabilities': probs.tolist()
}
with open(PREDICTION_LOG, 'a') as f:
f.write(json.dumps(record) + '\n')
return pred_class, confidenceRolling Mean Confidence Score
Average prediction confidence is one of the most sensitive early-warning signals for concept drift. When the model encounters out-of-distribution data, its softmax probabilities spread more evenly across classes (closer to uniform), lowering the max probability. Plot the rolling mean confidence over time: a sustained drop below a deployment threshold (typically set at the 5th percentile of calibration confidence) signals the model is becoming uncertain about production inputs.
import numpy as np
import pandas as pd
np.random.seed(42)
# Simulate confidence scores over 12 weeks
# First 6 weeks: healthy (high confidence)
week1to6 = np.random.beta(8, 2, 600) # mean ~0.8
# Last 6 weeks: degrading (lower confidence)
week7to12 = np.random.beta(3, 3, 600) # mean ~0.5
all_scores = np.concatenate([week1to6, week7to12])
dates = pd.date_range('2024-01-01', periods=len(all_scores), freq='h')
df = pd.DataFrame({'timestamp': dates, 'confidence': all_scores})
df['week'] = df['timestamp'].dt.isocalendar().week
weekly_mean = df.groupby('week')['confidence'].mean()
threshold = weekly_mean.iloc[:6].quantile(0.05) # 5th percentile of healthy period
print(f'Alert threshold: {threshold:.4f}')
for week, mean_conf in weekly_mean.items():
status = 'ALERT' if mean_conf < threshold else 'OK'
print(f'Week {week}: mean confidence = {mean_conf:.4f} [{status}]')Predicted Class Distribution Monitoring
The distribution of predicted classes should remain stable over time if the model and real-world data are aligned. A shift in the predicted positive rate — e.g., a binary classifier predicting positive 30% in training but 5% in production — is a strong drift signal. Monitor the predicted class distribution weekly and alert when the proportion of any class deviates by more than a threshold from the training baseline, using a chi-squared test.
import numpy as np
from scipy.stats import chi2_contingency
# Baseline class distribution from training predictions
train_predictions = np.array([0, 1]).repeat([700, 300]) # 70% neg, 30% pos
np.random.shuffle(train_predictions)
# Current week production predictions (shifted distribution)
prod_predictions = np.random.choice([0, 1], p=[0.92, 0.08], size=500)
train_counts = [np.sum(train_predictions == 0), np.sum(train_predictions == 1)]
prod_counts = [np.sum(prod_predictions == 0), np.sum(prod_predictions == 1)]
chi2, p_val, _, _ = chi2_contingency([train_counts, prod_counts])
print(f'Train distribution: {train_counts[0]/len(train_predictions):.2%} neg, {train_counts[1]/len(train_predictions):.2%} pos')
print(f'Prod distribution: {prod_counts[0]/len(prod_predictions):.2%} neg, {prod_counts[1]/len(prod_predictions):.2%} pos')
print(f'Chi-squared: {chi2:.2f}, p-value: {p_val:.4f}')
print('Prediction distribution drift:', p_val < 0.05)Confidence Calibration: Are Probabilities Trustworthy?
Calibration measures whether a model's predicted probabilities match actual frequencies. A perfectly calibrated model that predicts 80% confidence should be correct 80% of the time. Uncalibrated models — common with gradient boosting and deep learning — produce overconfident or underconfident probabilities. Monitor calibration over time using a reliability diagram: bin predictions by confidence and compare predicted vs observed accuracy within each bin.
import numpy as np
from sklearn.calibration import calibration_curve
np.random.seed(42)
# Overconfident model: high probabilities but not that accurate
y_true = np.random.binomial(1, 0.6, 1000)
y_prob_uncalibrated = np.clip(
np.random.beta(5, 2, 1000) * (y_true * 0.6 + 0.2), 0, 1
)
fraction_of_positives, mean_predicted = calibration_curve(
y_true, y_prob_uncalibrated, n_bins=10
)
print('Bin | Predicted | Observed')
for pred, obs in zip(mean_predicted, fraction_of_positives):
gap = abs(pred - obs)
status = '*** MISCALIBRATED' if gap > 0.1 else 'OK'
print(f'{pred:.2f} | {obs:.2f} {status}')Expected Calibration Error
The Expected Calibration Error (ECE) is a single scalar that summarises calibration quality. It is the weighted average of the gap between predicted confidence and observed accuracy across all confidence bins, weighted by the number of examples in each bin. ECE near 0 means excellent calibration; ECE above 0.05 warrants investigation; ECE above 0.1 suggests the model's probabilities should not be trusted without recalibration.
import numpy as np
from sklearn.calibration import calibration_curve
def expected_calibration_error(y_true, y_prob, n_bins=10):
fraction_pos, mean_pred = calibration_curve(y_true, y_prob, n_bins=n_bins)
bin_sizes = []
bins = np.linspace(0, 1, n_bins + 1)
for i in range(n_bins):
in_bin = (y_prob >= bins[i]) & (y_prob < bins[i+1])
bin_sizes.append(in_bin.sum())
n = len(y_true)
ece = sum(
(bin_size / n) * abs(pred - obs)
for bin_size, pred, obs in zip(bin_sizes[:len(mean_pred)],
mean_pred, fraction_pos)
)
return round(ece, 4)
np.random.seed(42)
y_true = np.random.binomial(1, 0.5, 1000)
y_prob = np.random.beta(3, 3, 1000)
ece = expected_calibration_error(y_true, y_prob)
print(f'ECE: {ece}')
print('Calibration quality:', 'Good' if ece < 0.05 else 'Poor')Low-Confidence Request Flagging
In production, flag requests below a confidence threshold for human review or fallback handling. This is especially important in high-stakes domains: a credit application with 51% predicted default confidence should go to a human underwriter rather than being auto-rejected. Set the threshold at the point where your business can tolerate uncertainty, and log all flagged requests to a review queue with the full feature vector and prediction for analyst inspection.
import datetime
CONFIDENCE_THRESHOLD = 0.75
REVIEW_QUEUE = []
def predict_with_human_review(request_id, features, model_func):
pred_class, confidence = model_func(features)
result = {
'request_id': request_id,
'predicted_class': pred_class,
'confidence': confidence,
'timestamp': datetime.datetime.utcnow().isoformat(),
'needs_review': confidence < CONFIDENCE_THRESHOLD
}
if result['needs_review']:
REVIEW_QUEUE.append({**result, 'features': features})
print(f'Request {request_id} FLAGGED for review (conf={confidence:.2f})')
else:
print(f'Request {request_id} auto-decided: class={pred_class} (conf={confidence:.2f})')
return resultStoring Predictions in a Time-Series Database
For scalable monitoring, write prediction records to a time-series store like PostgreSQL with a timestamp index, InfluxDB, or a cloud analytics warehouse. Query aggregates (hourly/daily mean confidence, class distribution) using SQL window functions. Set up automated dashboards in Grafana or a BI tool that refresh every hour, so the team has real-time visibility without writing any monitoring scripts manually.
import psycopg2
import json
from datetime import datetime
# Schema (run once):
# CREATE TABLE prediction_log (
# id SERIAL PRIMARY KEY,
# request_id TEXT,
# model_version TEXT,
# predicted_class INT,
# confidence FLOAT,
# probabilities JSONB,
# timestamp TIMESTAMPTZ DEFAULT NOW()
# );
def log_prediction_to_db(conn, request_id, model_version, pred_class, confidence, probs):
cur = conn.cursor()
cur.execute(
'''INSERT INTO prediction_log
(request_id, model_version, predicted_class, confidence, probabilities)
VALUES (%s, %s, %s, %s, %s)''',
(request_id, model_version, pred_class, confidence, json.dumps(probs))
)
conn.commit()
print('DB logging function ready.')Setting Deployment Thresholds from Calibration
Choose the deployment confidence threshold using calibration data rather than intuition. Plot the ECE and accuracy at different confidence thresholds: at what confidence level does the model achieve 95% accuracy? That level becomes the auto-approval threshold. Below it, requests go to human review. This creates a principled confidence-based routing system that maintains high accuracy on auto-approved requests while routing uncertain cases to human judgment.
import numpy as np
np.random.seed(42)
# Calibrated model: high confidence = high accuracy
y_true = np.random.binomial(1, 0.6, 10000)
base_prob = np.where(y_true == 1, np.random.beta(7, 2, 10000),
np.random.beta(2, 7, 10000))
y_prob = np.clip(base_prob, 0, 1)
results = []
for threshold in np.arange(0.5, 0.96, 0.05):
mask = y_prob > threshold
if mask.sum() < 100:
break
acc = (y_true[mask] == (y_prob[mask] > 0.5)).mean()
coverage = mask.mean()
results.append({'threshold': round(threshold, 2), 'accuracy': round(acc, 3),
'coverage': round(coverage, 3)})
import pandas as pd
df = pd.DataFrame(results)
print(df.to_string(index=False))Alerting on Confidence Drops
Automate confidence monitoring by running a scheduled script that computes daily confidence statistics and triggers an alert if the rolling mean drops below the deployment threshold. Use a z-score test: compute how many standard deviations below the training baseline the current daily mean confidence is, and alert if it exceeds 2 standard deviations. This approach adapts automatically to natural variability in confidence without hard-coded thresholds.
import numpy as np
# Training period confidence stats (computed once during deployment)
train_conf_mean = 0.82
train_conf_std = 0.08
Z_THRESHOLD = 2.0 # alert if > 2 std below baseline
def check_confidence_alert(daily_confidences):
daily_mean = np.mean(daily_confidences)
z_score = (train_conf_mean - daily_mean) / train_conf_std
status = 'ALERT' if z_score > Z_THRESHOLD else 'OK'
print(f'Daily mean: {daily_mean:.4f}')
print(f'Z-score below baseline: {z_score:.2f}')
print(f'Status: {status}')
return status
# Healthy day
check_confidence_alert(np.random.beta(8, 2, 500)) # mean ~0.8
print()
# Degraded day
check_confidence_alert(np.random.beta(3, 3, 500)) # mean ~0.5Segment-Level Confidence Monitoring
Aggregate confidence monitoring can miss issues that only affect specific user segments. A model may maintain stable average confidence overall while its confidence for a newly acquired customer cohort drops sharply. Always break down monitoring by key dimensions: customer tenure bucket, product category, geography, and device type. A sudden confidence drop in a specific segment is often the earliest symptom of concept drift within that slice before it propagates to the aggregate.
import pandas as pd
import numpy as np
# Simulated prediction log
df = pd.DataFrame({
'segment': np.random.choice(['new_user', 'returning', 'premium'], 1000, p=[0.3, 0.5, 0.2]),
'confidence': np.random.beta(8, 2, 1000),
'week': np.random.choice(range(1, 9), 1000)
})
# Simulate drop for new_user segment in weeks 6-8
mask = (df['segment'] == 'new_user') & (df['week'] >= 6)
df.loc[mask, 'confidence'] *= 0.6
print(df.groupby(['week', 'segment'])['confidence'].mean().round(3).to_string())Quick Check
Test your understanding of Machine Learning with Python concepts from this lesson.
Lesson Recap
In this lesson you learned: logging predictions with timestamps creates an audit trail that enables both retrospective analysis and real-time monitoring, rolling mean confidence and predicted class distribution are label-free early warning signals for model degradation, and Expected Calibration Error measures how trustworthy predicted probabilities are and should be monitored over time. Next up we use the Evidently AI library to generate comprehensive automated drift reports and integrate them into alert pipelines.
常见问题解答
「监控预测分布与置信度分数」课时是免费的吗？
是的 — 「监控预测分布与置信度分数」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 Machine Learning Academy 课程的其余内容，请升级到 CoddyKit PRO。 Machine Learning Academy 课程共包含 4 节课。
「监控预测分布与置信度分数」这节课中我会学到什么？
您将把预测概率记录到时间序列存储中，绘制滚动平均置信度，并在平均置信度低于部署阈值时发出标记。 你通过在浏览器中直接运行的动手代码来练习 Machine Learning Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 Machine Learning Academy 需要有经验吗？
无需任何先前经验。CoddyKit 上的 Machine Learning Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 3 节课，共 4 节。
「监控预测分布与置信度分数」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 Machine Learning Academy 课中编写并运行代码吗？
能。每节 Machine Learning Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。