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()