ทะเบียนแบบจำลอง: การเตรียมใช้งาน การผลิต และการจัดเก็บถาวร

import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier

# Option 1: Register during logging
with mlflow.start_run():
    clf = RandomForestClassifier(n_estimators=100, random_state=42)
    # clf.fit(X_train, y_train)
    mlflow.sklearn.log_model(
        sk_model=clf,
        artifact_path='model',
        registered_model_name='SentimentClassifier'  # auto-registers
    )
    print('Model registered as SentimentClassifier v1')

from mlflow.tracking import MlflowClient

client = MlflowClient(tracking_uri='http://localhost:5000')

# Option 2: Register from an existing run artifact
run_id = 'abc123def456'  # get this from mlflow.last_active_run().info.run_id
model_uri = f'runs:/{run_id}/model'

model_version = mlflow.register_model(
    model_uri=model_uri,
    name='SentimentClassifier'
)
print('Version:', model_version.version)
print('Status:', model_version.status)  # PENDING_REGISTRATION -> READY

from mlflow.tracking import MlflowClient

client = MlflowClient()

# Transition version 1 to Staging
client.transition_model_version_stage(
    name='SentimentClassifier',
    version='1',
    stage='Staging',
    archive_existing_versions=False
)
print('Version 1 -> Staging')

# After testing, promote to Production (archives previous Production)
client.transition_model_version_stage(
    name='SentimentClassifier',
    version='1',
    stage='Production',
    archive_existing_versions=True  # auto-archives old Production
)
print('Version 1 -> Production')

from mlflow.tracking import MlflowClient

client = MlflowClient()

# Add description to version
client.update_model_version(
    name='SentimentClassifier',
    version='1',
    description=('RandomForest trained on IMDB v2 (50k reviews). '
                 'Test accuracy 0.924, F1 0.921. '
                 'Replaces rule-based baseline.')
)

# Add tags for filtering and search
client.set_model_version_tag(
    name='SentimentClassifier',
    version='1',
    key='dataset',
    value='imdb_v2'
)
client.set_model_version_tag('SentimentClassifier', '1', 'algorithm', 'random_forest')
print('Description and tags added.')

import mlflow.sklearn

# Load the current Production model by stage
model_name = 'SentimentClassifier'
stage = 'Production'
model_uri = f'models:/{model_name}/{stage}'

production_model = mlflow.sklearn.load_model(model_uri)
print('Loaded model from:', model_uri)

# Or load a specific version
version_uri = f'models:/{model_name}/1'
v1_model = mlflow.sklearn.load_model(version_uri)
print('Loaded specific version 1')

# Make predictions
# predictions = production_model.predict(X_new)

from mlflow.tracking import MlflowClient

client = MlflowClient()

# List all versions of a model
versions = client.search_model_versions("name='SentimentClassifier'")
for v in versions:
    print(f'Version {v.version}: stage={v.current_stage}, run_id={v.run_id[:8]}')

# Get the metric from the associated training run
for v in versions:
    run = client.get_run(v.run_id)
    acc = run.data.metrics.get('test_accuracy', 'N/A')
    print(f'  Version {v.version} accuracy: {acc}')

from mlflow.tracking import MlflowClient
import mlflow.sklearn

client = MlflowClient()

def get_metric(run_id, metric_name):
    return client.get_run(run_id).data.metrics.get(metric_name, 0)

def promote_if_better(model_name, challenger_version, metric='test_f1'):
    # Get current production version
    prod_versions = client.get_latest_versions(model_name, stages=['Production'])
    if not prod_versions:
        print('No production model found -- promoting challenger directly.')
        client.transition_model_version_stage(model_name, challenger_version, 'Production')
        return

    prod_v = prod_versions[0]
    prod_score = get_metric(prod_v.run_id, metric)
    chall_run_id = client.get_model_version(model_name, challenger_version).run_id
    chall_score = get_metric(chall_run_id, metric)

    print(f'Champion {metric}: {prod_score:.4f}  Challenger: {chall_score:.4f}')
    if chall_score > prod_score:
        client.transition_model_version_stage(model_name, challenger_version,
                                              'Production', archive_existing_versions=True)
        print('Challenger promoted to Production!')
    else:
        print('Champion retained.')

from mlflow.tracking import MlflowClient

client = MlflowClient()

# Manually archive a specific version
client.transition_model_version_stage(
    name='SentimentClassifier',
    version='1',
    stage='Archived'
)
print('Version 1 archived.')

# List only archived versions
archived = client.search_model_versions(
    "name='SentimentClassifier' and stage='Archived'"
)
for v in archived:
    print(f'Archived: v{v.version} created {v.creation_timestamp}')

# Serve the Production model as a local REST endpoint
# mlflow models serve -m 'models:/SentimentClassifier/Production' --port 8080

# Then call it with curl:
# curl -X POST http://localhost:8080/invocations \
#   -H 'Content-Type: application/json' \
#   -d '{"dataframe_records": [{"feature1": 0.5, "feature2": 1.2}]}'

# Or with Python requests:
import requests
data = {'dataframe_records': [{'feature1': 0.5, 'feature2': 1.2}]}
response = requests.post('http://localhost:8080/invocations', json=data)
print('Prediction:', response.json())

# Polling script (run on a schedule, e.g., cron every 5 minutes)
from mlflow.tracking import MlflowClient
import json
import os

client = MlflowClient()
state_file = '/tmp/model_registry_state.json'

def load_state():
    if os.path.exists(state_file):
        return json.load(open(state_file))
    return {}

def save_state(state):
    json.dump(state, open(state_file, 'w'))

state = load_state()
prod = client.get_latest_versions('SentimentClassifier', stages=['Production'])
if prod:
    current_prod = prod[0].version
    if state.get('production_version') != current_prod:
        print(f'ALERT: Production changed to version {current_prod}')
        # send_slack_notification(current_prod)
        state['production_version'] = current_prod
        save_state(state)