MlflowChart

MlflowChart

MlflowChart(*, tracking_uri: str | None = None, experiment_id: str | None = None, runs: Sequence[Any] | None = None, metric_key: str = '', poll_seconds: int | None = 5, smoothing_kind: str = 'gaussian', smoothing_param: float | None = None, show_slider: bool = True, x_axis: str = 'step', width: int = 700, height: int = 300, **kwargs: Any)

Bases: AnyWidget

Live line chart that polls MLflow for metric data.

Renders a Canvas-based chart that can auto-update while runs are active. Supports multiple runs for side-by-side comparison and optional rolling smoothing (rolling mean, exponential moving average, or gaussian).

Examples:

import mlflow
from mlflow_widgets import MlflowChart

with mlflow.start_run() as run:
    for step in range(100):
        mlflow.log_metric("loss", 1.0 / (step + 1), step=step)

    chart = MlflowChart(
        tracking_uri="http://localhost:5000",
        runs=[run],
        metric_key="loss",
    )
    chart

Create a MlflowChart widget.

Parameters:

Name	Type	Description	Default
tracking_uri	str | None	MLflow tracking server URI. Defaults to MLFLOW_TRACKING_URI env var or http://localhost:5000.	None
experiment_id	str | None	MLflow experiment ID. If provided and runs is not given, all runs in this experiment are charted.	None
runs	Sequence[Any] | None	A list of MLflow Run objects, run-ID strings, or dicts with id and label keys.	None
metric_key	str	The metric key to chart (e.g. "loss").	''
poll_seconds	int | None	Seconds between polling updates, or None to disable auto-polling and show a manual refresh button instead.	5
smoothing_kind	str	"rolling", "exponential", or "gaussian".	'gaussian'
smoothing_param	float | None	Smoothing parameter, or None for no smoothing.	None
show_slider	bool	Whether to show the smoothing slider in the UI.	True
x_axis	str	X-axis mode: "step", "wall", or "relative".	'step'
width	int	Chart width in pixels.	700
height	int	Chart height in pixels.	300

Source code in src/mlflow_widgets/chart.py

def __init__(
    self,
    *,
    tracking_uri: str | None = None,
    experiment_id: str | None = None,
    runs: Sequence[Any] | None = None,
    metric_key: str = "",
    poll_seconds: int | None = 5,
    smoothing_kind: str = "gaussian",
    smoothing_param: float | None = None,
    show_slider: bool = True,
    x_axis: str = "step",
    width: int = 700,
    height: int = 300,
    **kwargs: Any,
) -> None:
    """Create a MlflowChart widget.

    Args:
        tracking_uri: MLflow tracking server URI. Defaults to
            ``MLFLOW_TRACKING_URI`` env var or ``http://localhost:5000``.
        experiment_id: MLflow experiment ID. If provided and ``runs`` is
            not given, all runs in this experiment are charted.
        runs: A list of MLflow Run objects, run-ID strings, or dicts
            with ``id`` and ``label`` keys.
        metric_key: The metric key to chart (e.g. ``"loss"``).
        poll_seconds: Seconds between polling updates, or ``None`` to
            disable auto-polling and show a manual refresh button instead.
        smoothing_kind: ``"rolling"``, ``"exponential"``, or ``"gaussian"``.
        smoothing_param: Smoothing parameter, or ``None`` for no smoothing.
        show_slider: Whether to show the smoothing slider in the UI.
        x_axis: X-axis mode: ``"step"``, ``"wall"``, or ``"relative"``.
        width: Chart width in pixels.
        height: Chart height in pixels.
    """
    self._tracking_uri = tracking_uri or os.environ.get(
        "MLFLOW_TRACKING_URI", "http://localhost:5000"
    )
    self._experiment_id = experiment_id
    self._mlflow = _try_import_mlflow()
    self._client = None
    if self._mlflow is not None:
        self._client = self._mlflow.MlflowClient(self._tracking_uri)

    self._run_specs: list[dict] = []
    if runs is not None:
        self._run_specs = self._normalize_runs(runs)
    elif experiment_id is not None:
        self._run_specs = self._discover_runs(experiment_id)

    self._poll_timer: threading.Timer | None = None
    self._poll_lock = threading.Lock()
    self._stopped = False

    super().__init__(
        metric_key=metric_key,
        poll_seconds=poll_seconds,
        smoothing_kind=smoothing_kind,
        smoothing_param=smoothing_param,
        show_slider=show_slider,
        x_axis=x_axis,
        width=width,
        height=height,
        **kwargs,
    )

    if self._run_specs and metric_key:
        self.refresh()

    if poll_seconds is not None and poll_seconds > 0:
        self._schedule_poll()

refresh

refresh() -> None

Manually refresh metric data from MLflow.

Source code in src/mlflow_widgets/chart.py

def refresh(self) -> None:
    """Manually refresh metric data from MLflow."""
    try:
        series = self._fetch_series()
        total_points = sum(len(s["points"]) for s in series)
        now = datetime.now().strftime("%H:%M:%S")
        self._series_data = series
        self._status = f"{total_points} points across {len(series)} runs (updated {now})"

        still_running = any(
            r.get("status") in ("RUNNING", None) for r in self._run_specs
        )
        if not still_running and self._poll_timer is not None:
            self._status += " — all runs finished"
            self.stop()
    except Exception as exc:
        self._status = f"Error: {exc}"

stop

stop() -> None

Stop auto-polling.

Source code in src/mlflow_widgets/chart.py

def stop(self) -> None:
    """Stop auto-polling."""
    self._stopped = True
    with self._poll_lock:
        if self._poll_timer is not None:
            self._poll_timer.cancel()
            self._poll_timer = None

close

close() -> None

Source code in src/mlflow_widgets/chart.py

def close(self) -> None:
    self.stop()
    super().close()