Meta Ax 自适应实验实战：贝叶斯优化与多目标调参指南

import importlib, subprocess, sys
def _ensure(module, pip_name=None):
   try:
       importlib.import_module(module)
   except ImportError:
       print(f"Installing {pip_name or module} ...")
       subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", pip_name or module])
_ensure("ax", "ax-platform")
_ensure("sklearn", "scikit-learn")
import logging, warnings, time
import numpy as np
import matplotlib.pyplot as plt
warnings.filterwarnings("ignore")
logging.getLogger("ax").setLevel(logging.WARNING)
from ax.api.client import Client
from ax.api.configs import RangeParameterConfig, ChoiceParameterConfig
from sklearn.datasets import make_classification
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import StratifiedKFold, cross_val_score
np.random.seed(0)

X, y = make_classification(
   n_samples=1400, n_features=20, n_informative=8, n_redundant=4,
   n_classes=3, random_state=0,
)
CV = StratifiedKFold(n_splits=3, shuffle=True, random_state=0)
def evaluate(p):
   n_est, depth = int(p["n_estimators"]), int(p["max_depth"])
   clf = RandomForestClassifier(
       n_estimators=n_est,
       max_depth=depth,
       max_features=float(p["max_features"]),
       min_samples_leaf=int(p["min_samples_leaf"]),
       criterion=p["criterion"],
       ccp_alpha=float(p["ccp_alpha"]),
       n_jobs=-1,
       random_state=0,
   )
   accuracy = cross_val_score(clf, X, y, cv=CV, scoring="accuracy").mean()
   model_size = n_est * depth
   return {"accuracy": float(accuracy), "model_size": float(model_size)}
SEARCH_SPACE = [
   RangeParameterConfig(name="n_estimators",    bounds=(50, 300),     parameter_type="int"),
   RangeParameterConfig(name="max_depth",       bounds=(3, 24),       parameter_type="int"),
   RangeParameterConfig(name="max_features",    bounds=(0.2, 1.0),    parameter_type="float"),
   RangeParameterConfig(name="min_samples_leaf",bounds=(1, 12),       parameter_type="int"),
   RangeParameterConfig(name="ccp_alpha",       bounds=(1e-5, 1e-1),  parameter_type="float", scaling="log"),
   ChoiceParameterConfig(name="criterion", values=["gini", "entropy", "log_loss"],
                         parameter_type="str", is_ordered=False),
]
def run_study(client, total_trials, metric_keys, batch=4):
   records = []
   while len(records) < total_trials:
       trials = client.get_next_trials(max_trials=min(batch, total_trials - len(records)))
       if not trials:
           break
       for idx, params in trials.items():
           full = evaluate(params)
           raw = {k: full[k] for k in metric_keys}
           client.complete_trial(trial_index=idx, raw_data=raw)
           records.append({"trial": idx, "params": params, **full})
   return records

print("\n=== Study 1: constrained single-objective Bayesian optimization ===")
c1 = Client()
c1.configure_experiment(parameters=SEARCH_SPACE, name="rf_constrained")
c1.configure_optimization(objective="accuracy",
                         outcome_constraints=["model_size <= 2500"])
rec1 = run_study(c1, total_trials=24, metric_keys=["accuracy", "model_size"])
best_params, prediction, best_idx, best_arm = c1.get_best_parameterization()
print("\nBest feasible configuration found:")
for k, v in best_params.items():
   print(f"   {k:>16}: {v}")
print("   predicted:", prediction)
feasible = [(r["trial"], r["accuracy"]) for r in rec1 if r["model_size"] <= 2500]
best_so_far, cur = [], -np.inf
for _, acc in feasible:
   cur = max(cur, acc); best_so_far.append(cur)
plt.figure(figsize=(7, 4))
plt.plot(range(1, len(best_so_far) + 1), best_so_far, "o-")
plt.xlabel("feasible trial #"); plt.ylabel("best accuracy so far")
plt.title("Study 1 — convergence (subject to model_size <= 2500)")
plt.grid(alpha=0.3); plt.tight_layout(); plt.show()

print("\n=== Study 2: multi-objective (accuracy vs. model_size) ===")
c2 = Client()
c2.configure_experiment(parameters=SEARCH_SPACE, name="rf_multiobjective")
c2.configure_optimization(objective="accuracy, -model_size")
rec2 = run_study(c2, total_trials=28, metric_keys=["accuracy", "model_size"])
try:
   frontier = c2.get_pareto_frontier()
   print(f"Ax identified {len(frontier)} Pareto-optimal configurations.")
except Exception as e:
   frontier = None
   print("get_pareto_frontier unavailable in this version:", e)
acc = np.array([r["accuracy"] for r in rec2])
size = np.array([r["model_size"] for r in rec2])
order = np.argsort(size)
pareto_idx, best_acc = [], -np.inf
for i in order:
   if acc[i] > best_acc:
       best_acc = acc[i]; pareto_idx.append(i)
plt.figure(figsize=(7, 5))
plt.scatter(size, acc, c="lightgray", label="all trials")
plt.scatter(size[pareto_idx], acc[pareto_idx], c="crimson", zorder=3, label="Pareto front")
plt.plot(size[pareto_idx], acc[pareto_idx], "--", c="crimson", alpha=0.6)
plt.xlabel("model_size (lower = cheaper)"); plt.ylabel("accuracy (higher = better)")
plt.title("Study 2 — accuracy vs. model size trade-off")
plt.legend(); plt.grid(alpha=0.3); plt.tight_layout(); plt.show()