Skip to content

Overview

This page provides information for using the Model Production Monitoring (MPM) Python SDK.

For more information on MPM, see Use MPM.

Install MPM SDK

To use the MPM SDK, downoad and install using:

pip install comet-mpm

The full reference documentation is available at MPM SDK reference

Use MPM SDK in your Inference Server code

Here are two examples, using popular frameworks.

Configure with Flask

If you are using Flask to serve your predictions, you simply need to initialize the MPM logger using CometMPM() and from there you can log predictions to MPM using .log_event(...). The MPM SDK includes a number of optimizations to make sure the logging overhead is kept at a minimum.

Here is a example of what an integration with Flask would look like:

from uuid import uuid4

from comet_mpm import CometMPM
from flask import Flask, jsonify, request

app = Flask(__name__)
MPM = CometMPM()

...

@app.post("/prediction")
def prediction():
    prediction_id = str(uuid4())

    input_features = get_input_features(request)
    prediction, probability = get_prediction()

    MPM.log_event(
        prediction_id=prediction_id,
        input_features=input_features,
        output_features={"value": prediction, "probability": probability}
    )

    return jsonify({"prediction": prediction, "probability": probability})

Configure with FastAPI/asyncio

Using the MPM SDK with FastAPI or other asyncio frameworks is very similar to usage with Flask. You do need to add two additional calls on startup and shutdown to make sure all the events are correctly sent to MPM.

Here is a example of what an integration with FastAPI would look like:

from uuid import uuid4

from comet_mpm import CometMPM
from fastapi import FastAPI, Request

app = FastAPI()
MPM = CometMPM(asyncio=True)

...

@app.on_event("startup")
async def startup_event():
    MPM.connect()


@app.post("/prediction")
async def prediction(request: Request):
    prediction_id = str(uuid4())

    input_features = get_input_features(request)
    prediction, probability = get_prediction()

    await MPM.log_event(
        prediction_id=prediction_id,
        input_features=input_features,
        output_features={"value": prediction, "probability": probability}
    )

    return {"prediction": prediction, "probability": probability}


@app.on_event("shutdown")
async def shutdown_mpm():
    await MPM.join()

Learn more

Mar. 27, 2024