Experiment Tracking with MLflow and W&B

In the synthetic incident from LLM Observability & Monitoring, a supplied outcome label marks req_205 as an unsupported deploy approval. Its trace locates the event. Check the underlying evidence for claim truth and evaluate a repair independently. All cases and candidate behaviors here are hand-authored teaching fixtures.

deploy-answerer-v1.2-fix stays BLOCKED_PENDING_EVALUATION. The handoff carries the incident policy, failing template, exemplar request, evidence version, candidate template, and required safety metric. It carries no raw production trace into the eval set. Did deploy-answerer-v1.2-fix earn a controlled next step?

For a large language model (LLM) application, an experiment can evaluate a changed prompt template, evidence policy, retriever, tool schema, model, or fine-tuned checkpoint, even when no weights change. The repair above is one such candidate. A passing development score can qualify a candidate for canary evaluation. First run fresh factual and operational checks.

Record the candidate with its score

Training curves and checkpoints remain useful. Traditional ML tracking also records datasets, preprocessing, code, evaluation, and deployment evidence; stochastic training isn't unique to LLMs. The same evidence discipline applies when a candidate changes prompts or retrieval instead of weights.^{[1]Reference 1ML Experiment Tracking.https://mlflow.org/docs/latest/ml/tracking/}

This fixture explores a routing repair. A real incident could instead involve model weights, stale data, infrastructure, or several interacting faults. Don't infer the mechanism merely because it involves a prompt.

An edit could fix req_205 and still break supported rollback guidance. Treat the incident as a hypothesis to test, then compare the proposed repair under fixed conditions.

Turn an incident into a question

Keep three questions distinct. Observability helps us find a reviewed failure. Tracking records whether a controlled candidate addresses it. Deployment control decides whether an artifact may receive traffic:

Next, name the objects that carry the question through the workflow. Each points back to req_205, but each gives a reviewer a different handle:

The experiment produces a review record. Keep passing and failing runs with their reasons, then nominate a passing candidate for fresh evaluation and release review before a canary.

Once the question is explicit, tracker choice becomes an implementation detail. MLflow Tracking records runs with parameters, metrics, tags, artifacts, source-code versions, and dataset inputs.^{[1]Reference 1ML Experiment Tracking.https://mlflow.org/docs/latest/ml/tracking/} Weights & Biases (W&B) uses runs with configuration, metrics, and artifacts for the same evidence workflow.^{[2]Reference 2Experiments overview.https://docs.wandb.ai/models/track} Start with a local version of that workflow so the logic is visible without credentials or a hosted service. Freeze the inputs before opening either service, so every later run answers the same question.

Freeze what the run is allowed to answer

The candidate can change, and so can the test set. If they both move, a better score has no clear cause. First, turn monitoring's handoff into a small regression suite and keep its expected behavior fixed.

req_205 contributes a simplified category and expected route, not production text. no_approval_support means the available material does not support an approval. It doesn't mean the previous lesson's versioned deploy policy disappeared. Categories keep this simulation small; they cannot reproduce a real model response without the actual question and admitted evidence.

Before running a candidate, predict the two labels: the incident row must abstain; the known-good row must serve. The contract makes those expectations explicit:

from dataclasses import dataclass, field
import hashlib
import json

def require_text(value: object, name: str) -> None:
    if type(value) is not str or not value.strip():
        raise ValueError(f"{name} must be a nonempty string")

ROUTES = frozenset({"SERVE", "ABSTAIN"})
EVIDENCE_STATES = frozenset({"no_approval_support", "release_record", "rollback_runbook"})

@dataclass(frozen=True)
class IncidentHandoff:
    incident_policy_id: str
    failing_template_id: str
    exemplar_request_id: str
    evidence_version: str
    candidate_template_id: str
    required_metric: str

    def __post_init__(self) -> None:
        for name in self.__dataclass_fields__:
            require_text(getattr(self, name), name)

@dataclass(frozen=True)
class EvalCase:
    case_id: str
    slice_name: str
    evidence_state: str
    expected_route: str
    source: str

    def __post_init__(self) -> None:
        for name in self.__dataclass_fields__:
            require_text(getattr(self, name), name)
        if self.evidence_state not in EVIDENCE_STATES or self.expected_route not in ROUTES:
            raise ValueError("unknown evidence category or expected route")

def validate_suite(cases: tuple[EvalCase, ...]) -> None:
    if type(cases) is not tuple or not cases or any(type(case) is not EvalCase for case in cases):
        raise ValueError("need a nonempty tuple of validated cases")
    if len({case.case_id for case in cases}) != len(cases):
        raise ValueError("case IDs must be unique")

handoff = IncidentHandoff(
    incident_policy_id="deploy-grounding-slo-v1",
    failing_template_id="deploy-answerer-v1.1-regression",
    exemplar_request_id="req_205",
    evidence_version="deploy-policy@v17",
    candidate_template_id="deploy-answerer-v1.2-fix",
    required_metric="zero unsafe claims; complete review; retained clean-case usefulness",
)

regression_cases = (
    EvalCase("req_205", "unsupported_approval", "no_approval_support", "ABSTAIN", "incident"),
    EvalCase("reg_002", "confirmed_freeze_status", "release_record", "SERVE", "known_good"),
)

print(f"failing_template={handoff.failing_template_id}")
print(f"candidate={handoff.candidate_template_id}")
print(f"regression_cases={len(regression_cases)}")
print(f"incident_expected_route={regression_cases[0].expected_route}")
print(f"required_metric={handoff.required_metric}")

failing_template=deploy-answerer-v1.1-regression
candidate=deploy-answerer-v1.2-fix
regression_cases=2
incident_expected_route=ABSTAIN
required_metric=zero unsafe claims; complete review; retained clean-case usefulness

The incident row should withhold an unsupported approval. The known-good row should serve supported freeze-status information. Those expectations are supplied labels. Now give the exact fixture representation an identity that travels with every run.

Hash the evaluation contract

A run ID tells you which execution happened, but not which cases it saw. Change a label, add a case, or update the evidence snapshot, and you've changed the experiment conditions.

A fingerprint is a deterministic checksum of those evaluation inputs. A matching cryptographic digest confirms the hashed representation when the run used those inputs. Review label quality, retain missing files, and record the process's actual inputs separately.

def suite_fingerprint(cases: tuple[EvalCase, ...]) -> str:
    validate_suite(cases)
    normalized = [
        {
            "case_id": case.case_id,
            "evidence_state": case.evidence_state,
            "expected_route": case.expected_route,
            "slice_name": case.slice_name,
            "source": case.source,
        }
        for case in sorted(cases, key=lambda item: item.case_id)
    ]
    payload = json.dumps(normalized, sort_keys=True, separators=(",", ":")).encode()
    return hashlib.sha256(payload).hexdigest()

regression_fingerprint = suite_fingerprint(regression_cases)
corrupted_cases = (
    regression_cases[0],
    EvalCase("reg_002", "confirmed_freeze_status", "release_record", "ABSTAIN", "known_good"),
)

print(f"regression_sha={regression_fingerprint[:16]}")
print(f"same_rows_same_sha={suite_fingerprint(tuple(reversed(regression_cases))) == regression_fingerprint}")
print(f"corrupted_label_detected={suite_fingerprint(corrupted_cases) != regression_fingerprint}")

regression_sha=c4039bef2635c843
same_rows_same_sha=True
corrupted_label_detected=True

Sorting by case_id makes row order irrelevant. The example then changes the known-good freeze-status case from SERVE to ABSTAIN, so the checksum changes too.

The lab stores the full SHA-256 digest and prints a 16-hex prefix to keep output readable. It rejects duplicate case IDs so sorting can't hide ambiguous row identities. Also record the dataset version, evidence snapshot, labeling policy, and code commit that built the cases.

This digest covers the categorical rows above. It does not hash real question text, retrieved documents, or model inputs, which the simulation does not contain. A real evaluation should hash or immutably identify those actual inputs and its labeling policy. MLflow dataset digest formats depend on the dataset implementation; logging dataset metadata is not the same as archiving its rows.^{[3]Reference 3mlflow.datahttps://mlflow.org/docs/latest/api_reference/python_api/mlflow.data.html}

Evaluation hygiene: preventing benchmark leakage

Suppose an engineer uses an incident to shape a repair and then sees that same case pass. That result is useful regression evidence. Test generalization on fresh cases because the incident may point to a symptom without revealing its cause.

If test feedback guides the candidate, the test has participated in development even without a gradient update. Adaptive selection of aggregate test scores can leak information even when engineers never inspect individual questions.

One practical separation is:

1. Regression fixtures: Reviewed historical cases. Here req_205 is synthetic. Preserve required behavior under the current policy; if the policy or a label is corrected, review and version that change rather than quietly dropping a failing case.
2. Development Exploration Slices: A broader set of queries where engineers experiment with prompt wording, few-shot exemplars, and tool parameters. Inspecting failures here is expected during iterative development.
3. Final holdout: A representative test kept separate from tuning and candidate selection. Define who can inspect it, when it is used, and how repeated release evaluations are governed.

When we use a holdout failure to motivate v1.2-fix-b, those cases become development feedback. Freeze the revised candidate before a fresh independent final test. Representative cases can come from authorized sampled traffic, reviewed task examples, or another justified sampling design; they do not have to be copied from live traffic.

Let the first pass expose a trade-off

Keep the incident case, but don't let it become the whole test. Tuning only to that failure risks overfitting: a candidate that abstains everywhere could drive unsafe_route_decisions to zero while breaking supported answers. Before selecting it, try a different safe evidence shape, such as a rollback runbook.

Add cases outside the incident suite before inspecting candidate results. Keep their origin visible: they test a different evidence shape, not another copy of the incident. Once you use their failures to revise a candidate, they become development data rather than an untouched final test.

Predict the useful behavior before scoring: serve rollback_runbook, abstain when the approval record is missing. Put those expectations in the holdout:

holdout_cases = (
    EvalCase("hold_001", "rollback_guidance", "rollback_runbook", "SERVE", "holdout"),
    EvalCase("hold_002", "missing_approval_record", "no_approval_support", "ABSTAIN", "holdout"),
)
all_cases = regression_cases + holdout_cases
holdout_fingerprint = suite_fingerprint(holdout_cases)
evaluation_fingerprint = suite_fingerprint(all_cases)

print(f"holdout_cases={len(holdout_cases)}")
print(f"holdout_sha={holdout_fingerprint[:16]}")
print(f"evaluation_sha={evaluation_fingerprint[:16]}")
print("guardrails=unsafe_route_decisions==0, unnecessary_abstention==0")

holdout_cases=2
holdout_sha=f83679bd6d6b8807
evaluation_sha=a20b1372381e9ca8
guardrails=unsafe_route_decisions==0, unnecessary_abstention==0

Two guardrails expose the routing trade-off: unsafe_route_decisions counts SERVE where the label requires ABSTAIN; unnecessary_abstention counts the reverse. Neither measures generated claim support. A model could choose the expected SERVE route and still invent an approval in its text. Actual payload review remains a separate requirement from the previous lesson.

Compare behavior on frozen cases

Three template versions make the trade-off visible. The code simulates their routing policies; it doesn't call an LLM or measure whether generated text is grounded. These deterministic offline evaluation results isolate the tracking problem:

• v1.1-regression returns SERVE for every category, including unsupported approvals.
• v1.2-fix repairs the incident but recognizes only explicit release records.
• v1.2-fix-b handles both release records and rollback-runbook evidence.

The implementation below reads only evidence_state, not expected_route. That avoids directly returning the label. It doesn't prevent a developer from tuning to known categories or cases, so a perfect development score still needs an independent final test.

@dataclass(frozen=True)
class Evaluation:
    template_id: str
    case_count: int
    unsafe_route_decisions: int
    unnecessary_abstention: int
    exact_route_rate: float

def predict_route(template_id: str, case: EvalCase) -> str:
    if template_id == "deploy-answerer-v1.1-regression":
        return "SERVE"
    if template_id == "deploy-answerer-v1.2-fix":
        return "SERVE" if case.evidence_state == "release_record" else "ABSTAIN"
    if template_id == "deploy-answerer-v1.2-fix-b":
        supported = {"release_record", "rollback_runbook"}
        return "SERVE" if case.evidence_state in supported else "ABSTAIN"
    raise ValueError(f"unknown template: {template_id}")

def evaluate(template_id: str, cases: tuple[EvalCase, ...]) -> Evaluation:
    validate_suite(cases)
    predicted = [predict_route(template_id, case) for case in cases]
    unsafe = sum(
        route == "SERVE" and case.expected_route == "ABSTAIN"
        for route, case in zip(predicted, cases, strict=True)
    )
    unnecessary = sum(
        route == "ABSTAIN" and case.expected_route == "SERVE"
        for route, case in zip(predicted, cases, strict=True)
    )
    exact = sum(route == case.expected_route for route, case in zip(predicted, cases, strict=True))
    return Evaluation(template_id, len(cases), unsafe, unnecessary, exact / len(cases))

known_fix = evaluate("deploy-answerer-v1.2-fix", regression_cases)
print(f"candidate={known_fix.template_id}")
print(f"regression_unsafe_routes={known_fix.unsafe_route_decisions}")
print(f"regression_unnecessary_abstention={known_fix.unnecessary_abstention}")
print(f"regression_pass={known_fix.exact_route_rate == 1.0}")
print("decision=CONTINUE_TO_HOLDOUT")

candidate=deploy-answerer-v1.2-fix
regression_unsafe_routes=0
regression_unnecessary_abstention=0
regression_pass=True
decision=CONTINUE_TO_HOLDOUT

The clean regression result earns a closer look, not approval. Now run the candidate against the holdout contract you declared earlier.

templates = (
    "deploy-answerer-v1.1-regression",
    "deploy-answerer-v1.2-fix",
    "deploy-answerer-v1.2-fix-b",
)
evaluations = [evaluate(template_id, all_cases) for template_id in templates]

print("template                          unsafe  unnecessary  exact")
for result in evaluations:
    print(
        f"{result.template_id:<33} "
        f"{result.unsafe_route_decisions:>6} "
        f"{result.unnecessary_abstention:>12} "
        f"{result.exact_route_rate:>6.0%}"
    )

first_fix = evaluations[1]
revised_fix = evaluations[2]
print(f"first_fix_blocked={first_fix.unnecessary_abstention > 0}")
print(f"revised_fix_passes_guardrails={revised_fix.unsafe_route_decisions == 0 and revised_fix.unnecessary_abstention == 0}")

template                          unsafe  unnecessary  exact
deploy-answerer-v1.1-regression        2            0    50%
deploy-answerer-v1.2-fix               0            1    75%
deploy-answerer-v1.2-fix-b             0            0   100%
first_fix_blocked=True
revised_fix_passes_guardrails=True

The first fix gets the incident route right but abstains on supported rollback guidance. Keep that rejected run so the revision has a clear reason to exist. Evaluate generated answers for factual safety separately from this routing simulation.

We used that holdout failure to motivate v1.2-fix-b, so the same four cases are now development evidence for the revision. The variable name holdout_cases records the original split; using its results for a revision makes it development data. Before real traffic, freeze the revised candidate and evaluate it on a fresh, independently reviewed test set. These four synthetic routes demonstrate tracker behavior; assess deployment safety on representative cases.

View larger

Keep evidence attached to each run

Attach enough context to a score that another engineer can inspect its cause:

1. What changed?
2. What exactly was evaluated?
3. Which metrics and guardrails moved?
4. Which evidence artifacts explain the decision?
5. Which incident or hypothesis motivated the run?

The local tracker below keeps metadata in memory and writes JSON artifacts under a temporary directory. Its report contains case IDs, categories, expected routes, and predicted routes. It demonstrates the record shape; the SDK examples also use temporary local storage. A shared project needs approved persistent storage and retention controls.

from dataclasses import asdict
from pathlib import Path
from tempfile import TemporaryDirectory
import math

@dataclass
class RunRecord:
    run_id: str
    params: dict[str, str] = field(default_factory=dict)
    metrics: dict[str, float] = field(default_factory=dict)
    artifacts: list[str] = field(default_factory=list)
    tags: dict[str, str] = field(default_factory=dict)

class LocalTracker:
    def __init__(self) -> None:
        self.runs: list[RunRecord] = []
        self.storage = TemporaryDirectory(prefix="tracking-lab-")
        self.root = Path(self.storage.name)

    def start_run(self, run_id: str) -> RunRecord:
        require_text(run_id, "run_id")
        if any(character not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-" for character in run_id):
            raise ValueError("run ID must be a safe directory name")
        if any(run.run_id == run_id for run in self.runs):
            raise ValueError("run IDs must be unique")
        run = RunRecord(run_id)
        self.runs.append(run)
        return run

    def require_owned(self, run: RunRecord) -> None:
        if not any(item is run for item in self.runs):
            raise ValueError("run isn't owned by this tracker")

    def log_params(self, run: RunRecord, params: dict[str, str]) -> None:
        self.require_owned(run)
        for key, value in params.items():
            require_text(key, "parameter name")
            require_text(value, key)
        if any(key in run.params and run.params[key] != value for key, value in params.items()):
            raise ValueError("a run parameter can't change after logging")
        run.params.update(params)

    def log_metrics(self, run: RunRecord, metrics: dict[str, float]) -> None:
        self.require_owned(run)
        if any(type(value) not in (int, float) or not math.isfinite(value) for value in metrics.values()):
            raise ValueError("metrics must be finite numbers, excluding booleans")
        run.metrics.update(metrics)

    def artifact_path(self, run: RunRecord, path: str) -> Path:
        self.require_owned(run)
        require_text(path, "artifact path")
        relative = Path(path)
        if relative.is_absolute() or ".." in relative.parts:
            raise ValueError("artifact path must stay inside its run")
        run_root = (self.root / run.run_id).resolve()
        if not run_root.is_relative_to(self.root.resolve()):
            raise ValueError("run directory must stay inside the tracker")
        target = (run_root / relative).resolve()
        if target == run_root or not target.is_relative_to(run_root):
            raise ValueError("artifact path must name a file inside its run")
        return target

    def log_artifact(self, run: RunRecord, path: str, payload: dict | list) -> None:
        target = self.artifact_path(run, path)
        target.parent.mkdir(parents=True, exist_ok=True)
        with target.open("x", encoding="utf-8") as artifact:
            artifact.write(json.dumps(payload, sort_keys=True, indent=2, allow_nan=False))
        run.artifacts.append(path)

    def set_tags(self, run: RunRecord, tags: dict[str, str]) -> None:
        self.require_owned(run)
        run.tags.update(tags)

evaluator_version = "route-eval-v2"
tracker = LocalTracker()
for index, result in enumerate(evaluations, start=1):
    run = tracker.start_run(f"run_{index:03d}")
    tracker.log_params(run, {
        "template_id": result.template_id,
        "policy_id": handoff.incident_policy_id,
        "evaluator_version": evaluator_version,
        "regression_sha": regression_fingerprint,
        "holdout_sha": holdout_fingerprint,
        "evaluation_sha": evaluation_fingerprint,
        "evidence_version": handoff.evidence_version,
        "code_commit": "UNCOMMITTED_TEACHING_FIXTURE",
    })
    tracker.log_metrics(run, {
        "case_count": float(result.case_count),
        "unsafe_route_decisions": float(result.unsafe_route_decisions),
        "unnecessary_abstention": float(result.unnecessary_abstention),
        "exact_route_rate": result.exact_route_rate,
    })
    report = [
        {**asdict(case), "predicted_route": predict_route(result.template_id, case)}
        for case in all_cases
    ]
    tracker.log_artifact(run, f"reports/{result.template_id}-redacted-eval.json", report)
    tracker.log_artifact(run, f"templates/{result.template_id}.json", {
        "template_id": result.template_id,
        "implementation": "predict_route in evaluate-templates.py",
        "simulation_only": True,
    })
    tracker.set_tags(run, {
        "incident_request_id": handoff.exemplar_request_id,
        "hypothesis": "restore evidence-gated deploy approvals",
        "raw_log_text_stored": "false",
    })

print(f"tracked_runs={len(tracker.runs)}")
print(f"shared_evaluation_sha={tracker.runs[0].params['evaluation_sha'][:16]}")
print(f"run_002_artifact={tracker.runs[1].artifacts[0]}")
print(f"run_002_template_artifact={tracker.runs[1].artifacts[1]}")
print(f"raw_log_text_stored={tracker.runs[2].tags['raw_log_text_stored']}")
report_path = tracker.root / "run_002" / tracker.runs[1].artifacts[0]
print(f"report_rows_read_back={len(json.loads(report_path.read_text()))}")

tracked_runs=3
shared_evaluation_sha=a20b1372381e9ca8
run_002_artifact=reports/deploy-answerer-v1.2-fix-redacted-eval.json
run_002_template_artifact=templates/deploy-answerer-v1.2-fix.json
raw_log_text_stored=false
report_rows_read_back=4

The three records share regression, holdout, and combined checksums, policy, evidence snapshot, and evaluator version. Each report now exists and can be read back. The template file is explicitly a simulation manifest, not a real prompt bundle; shipping a prompt requires its actual text and configuration. raw_log_text_stored=false is a descriptive tag, not a redaction control.

For real LLM runs, also capture the resolved model revision, generation settings, dependencies, and repeated trial IDs. The same inputs can produce different sampled outputs. Log per-case outcomes, latency, and token usage so aggregate quality doesn't conceal a slower or more expensive candidate.

That comparison contract changes when inputs or scoring rules change. Update the corresponding fingerprint or evaluator version and label the new conditions. The next cell checks deliberately altered metadata; it doesn't simulate the new dataset or execute a different evaluator.

def comparable(runs: list[RunRecord]) -> bool:
    comparison_fields = (
        "regression_sha",
        "holdout_sha",
        "evaluation_sha",
        "policy_id",
        "evidence_version",
        "evaluator_version",
    )
    return bool(runs) and all(
        all(type(run.params.get(field)) is str and run.params[field].strip() for run in runs)
        and len({run.params[field] for run in runs}) == 1
        for field in comparison_fields
    )

changed_suite_run = RunRecord(
    run_id="run_004",
    params={**tracker.runs[2].params, "holdout_sha": "different-holdout-sha"},
)
changed_evaluator_run = RunRecord(
    run_id="run_005",
    params={**tracker.runs[2].params, "evaluator_version": "route-eval-v3"},
)

print(f"tracked_runs_comparable={comparable(tracker.runs)}")
print(f"changed_suite_comparable={comparable(tracker.runs + [changed_suite_run])}")
print(f"changed_evaluator_comparable={comparable(tracker.runs + [changed_evaluator_run])}")
print("review_rule=match_inputs_and_evaluator_for_this_controlled_comparison")

tracked_runs_comparable=True
changed_suite_comparable=False
changed_evaluator_comparable=False
review_rule=match_inputs_and_evaluator_for_this_controlled_comparison

A dashboard can plot drifted runs alongside the original trio. For this controlled comparison, require matching inputs and evaluator semantics. Other valid study designs might compare a shared subset or account for changed sample composition; merely placing aggregate scores on the same chart does neither. These fields record claimed conditions, so also inspect the actual report and evaluated artifact.

View larger

When the suite stops separating candidates

A frozen suite can stop separating candidates. If each one scores exact_route_rate == 1.0, that metric can no longer rank them. Maybe the candidates solved these cases; maybe the suite is too narrow or the scoring too coarse. Check the evaluator before adding reviewed cases that expose useful differences.

Version any suite expansion and preserve valid regressions. Adding cases changes the fingerprint and comparison conditions. Record what coverage changed and why.

The second row matters when an LLM judges an answer using a scoring prompt instead of deterministic code. Its instruction text is part of the score. If that prompt drifts without a version bump, two runs can differ because the judge was asked a different question. Pin the judge prompt, model revision, and inference settings alongside evaluator_version, for example claim-route-eval-v2@judge-prompt-3. The separate LLM-as-a-Judge lesson examines how to validate those judges; this lab uses exact route comparisons.

Put the record into MLflow or W&B

The local tracker makes the record design visible, but it isn't a shared service. Now map each kind of evidence to a platform with shared storage, access, and review:

The mapping follows the official tracking, artifact, and run-tag APIs.^{[1]Reference 1ML Experiment Tracking.https://mlflow.org/docs/latest/ml/tracking/[2]Reference 2Experiments overview.https://docs.wandb.ai/models/track[4]Reference 4Artifacts overview.https://docs.wandb.ai/models/artifacts[5]Reference 5Organize runs with tags.https://docs.wandb.ai/models/runs/tags} In MLflow, mlflow.log_input(dataset, context="evaluation") logs input metadata; separately archive the approved suite file or pin an immutable source. In an online W&B run, run.use_artifact(...) records consumption of a versioned input artifact. Prefer an exact version or digest over a movable latest alias for reproducible evaluation.

Parameters describe a run's chosen inputs; changing one midway makes the run ambiguous. MLflow rejects changing an already logged parameter value. Metrics can have a history indexed by step, which preserves a training curve rather than only its last value. Tags such as review_only help organize runs. Record approval through a separate workflow and retain tamper-evident audit events.

Prompt Registry versus Model Registry

Start by asking what you need to version. MLflow Model Registry manages registered MLflow model packages and their source lineage. A package can contain trained weights, a custom Python function, or a wrapper around inference code. It isn't restricted to large checkpoints or weight-training runs.^{[6]Reference 6Model Registry Workflows | MLflow AI Platformhttps://mlflow.org/docs/latest/ml/model-registry/workflow/[7]Reference 7mlflow.pyfunc Python APIhttps://mlflow.org/docs/latest/api_reference/python_api/mlflow.pyfunc.html}

Prompt Registry versions instruction templates. MLflow accepts a string template or a list of chat messages, with placeholders such as {{context}}, and optional model configuration. It doesn't require a JSON/YAML bundle containing every serving setting.^{[8]Reference 8Prompt Registryhttps://mlflow.org/docs/latest/genai/prompt-registry/index.html}

For this prompt-only repair, the prompt template is the natural registered object. A team could also package a larger application as an MLflow model. Whichever object it chooses, the release record must identify the evaluated artifact and its effective serving configuration.

What a prompt version actually pins

A prompt version may look like it freezes the whole request. It keeps the prompt text immutable within that version, so edited text gets a new version. The model_config can still be updated or deleted within that same version. Pinning prompts:/deploy-answerer/3 doesn't freeze every inference setting. Versions can also be deleted, so a pin needs a retention policy.^{[8]Reference 8Prompt Registryhttps://mlflow.org/docs/latest/genai/prompt-registry/index.html[9]Reference 9mlflow.genai Python APIhttps://mlflow.org/docs/latest/api_reference/python_api/mlflow.genai.html}

Consider a version evaluated at temperature 0.0. Someone updates its model configuration to 0.7. The URI still ends in /3, but the effective configuration has changed. Archive the approved configuration and its digest in the release record; verify what serving actually uses. A version pin alone cannot provide that check.

Aliases are mutable names that point to versions. A team might define @canary_candidate and @champion; the names don't themselves grant approval, route requests, or stop a canary. Its deployment controller must enforce those actions and restrict who can move the pointers.

What goes wrong when serving follows @latest

MLflow reserves @latest for the latest available version. A deployment that automatically consumes it may load a newly registered draft:

# Dangerous operational anti-pattern in production:
prompt = mlflow.genai.load_prompt("prompts:/deploy-answerer@latest")

Saving a draft doesn't send traffic by itself. The risk arises when a serving process resolves @latest without a release gate. Workers can load different versions as they restart or refresh, and any mutable deployment alias can have that cache/fleet problem too.

Rollback remains possible: load an earlier retained version explicitly, or move an owned deployment alias back to it. Preserve the known-good version and configuration in advance, rather than reconstructing them during an incident. Use explicit pins or governed aliases for reviewed releases, then verify loaded versions across workers.

As of the reviewed MLflow 3.16.1 SDK, alias loads default to a 60-second cache lifetime; version loads default to indefinite caching. Choose cache behavior deliberately. The local check below demonstrates both alias rollback and mutable configuration; it doesn't validate a distributed rollout.^{[8]Reference 8Prompt Registryhttps://mlflow.org/docs/latest/genai/prompt-registry/index.html[10]Reference 10MLflow 3.16.1https://github.com/mlflow/mlflow/releases/tag/v3.16.1}

A tracking SDK records results; it doesn't automatically evaluate your candidate. MLflow's mlflow.genai.evaluate(...) runs a prediction function and scorers against curated data. It can use local code-based scorers or model-backed judges; model-backed calls may incur costs. Automatic evaluation of live traces is a different workflow from a frozen offline repair suite.^{[11]Reference 11Evaluation Quickstarthttps://mlflow.org/docs/latest/genai/eval-monitor/quickstart/[12]Reference 12Automatic Evaluationhttps://mlflow.org/docs/latest/genai/eval-monitor/automatic-evaluations/}

The next cell uses the real MLflow SDK against an explicitly local SQLite database and local artifact directory. It was checked with MLflow 3.16.1. Install that version and run it after the earlier lab cells in a fresh Python process. It reads back the run and report, then demonstrates the registry boundaries using placeholder templates. No tracking server, account, model call, or remote upload is involved.^{[13]Reference 13Backend Storeshttps://mlflow.org/docs/latest/self-hosting/architecture/backend-store/}

import os

os.environ["MLFLOW_DISABLE_TELEMETRY"] = "true"
import mlflow

with TemporaryDirectory(prefix="mlflow-local-") as directory:
    local_root = Path(directory).resolve()
    uri = f"sqlite:///{local_root / 'tracking.db'}"
    mlflow.set_tracking_uri(uri)
    mlflow.set_registry_uri(uri)
    client = mlflow.MlflowClient(tracking_uri=uri, registry_uri=uri)
    experiment_id = client.create_experiment(
        "deploy-claim-gate-repair",
        artifact_location=(local_root / "artifacts").as_uri(),
    )
    selected_record = tracker.runs[2]
    with mlflow.start_run(experiment_id=experiment_id, log_system_metrics=False) as active:
        mlflow.log_params(selected_record.params)
        mlflow.log_metrics(selected_record.metrics)
        mlflow.set_tags({**selected_record.tags, "decision": "review_only"})
        mlflow.log_dict([asdict(case) for case in all_cases], "inputs/eval-suite.json")
        for relative_path in selected_record.artifacts:
            source = tracker.root / selected_record.run_id / relative_path
            mlflow.log_artifact(str(source), artifact_path=str(Path(relative_path).parent))
        sdk_run_id = active.info.run_id

    stored = client.get_run(sdk_run_id)
    downloaded = client.download_artifacts(sdk_run_id, selected_record.artifacts[0])
    assert stored.data.params["evaluation_sha"] == evaluation_fingerprint
    assert len(json.loads(Path(downloaded).read_text())) == 4
    print(f"mlflow_status={stored.info.status}")
    print(f"stored_case_count={stored.data.metrics['case_count']:.0f}")
    print("report_read_back=True")

    first = mlflow.genai.register_prompt(
        "boundary-demo", "Summarize {{text}}.",
        model_config={"model_name": "local-fixture-model", "temperature": 0.0},
    )
    second = mlflow.genai.register_prompt("boundary-demo", "Summarize briefly: {{text}}.")
    latest = mlflow.genai.load_prompt("prompts:/boundary-demo@latest", cache_ttl_seconds=0)
    assert latest.version == second.version
    mlflow.genai.set_prompt_alias("boundary-demo", "champion", second.version)
    mlflow.genai.set_prompt_alias("boundary-demo", "champion", first.version)
    reverted = mlflow.genai.load_prompt("prompts:/boundary-demo@champion", cache_ttl_seconds=0)
    print(f"alias_rollback_to_older_version={reverted.version == first.version}")

    mlflow.genai.set_prompt_model_config(
        name=first.name, version=first.version,
        model_config={"model_name": "local-fixture-model", "temperature": 0.7},
    )
    pinned = mlflow.genai.load_prompt(
        f"prompts:/boundary-demo/{first.version}", cache_ttl_seconds=0,
    )
    assert pinned.version == first.version and pinned.template == first.template
    print(f"same_version_updated_temperature={pinned.model_config['temperature']}")

mlflow_status=FINISHED
stored_case_count=4
report_read_back=True
alias_rollback_to_older_version=True
same_version_updated_temperature=0.7

The temporary database is removed after this smoke check. For a shared project, configure an approved database and artifact store with access and retention controls. A FINISHED run means the logging context completed; its metrics may still fail your release policy.

When registering a real prompt, use the version returned by register_prompt. Hardcoding version=1 can point to an older template in an existing registry. This integration sketch requires your approved registry and an actually evaluated prompt string; it isn't executed by the lab and doesn't set an alias until you explicitly call the function after review:

import mlflow

def register_reviewed_prompt(evaluated_template: str):
    prompt = mlflow.genai.register_prompt(
        name="deploy-answerer",
        template=evaluated_template,
        commit_message="reviewed candidate; preserve its source run in release records",
    )
    mlflow.genai.set_prompt_alias(
        name=prompt.name,
        alias="canary_candidate",
        version=prompt.version,
    )
    return prompt

W&B can also log without contacting its service. With mode="offline", the SDK saves a real local run that could be synchronized later; mode="disabled" instead makes logging calls no-ops. This cell uses offline mode, doesn't call wandb sync, and removes the temporary files afterward. It was checked with wandb==0.30.0.^{[14]Reference 14init()https://docs.wandb.ai/models/ref/python/functions/init}

import wandb

with TemporaryDirectory(prefix="wandb-local-") as directory:
    os.environ["WANDB_CACHE_DIR"] = str(Path(directory) / "cache")
    os.environ["WANDB_DATA_DIR"] = str(Path(directory) / "staging")
    with wandb.init(
        project="deploy-claim-gate-repair",
        name="deploy-answerer-v1.2-fix-b",
        mode="offline",
        dir=directory,
        config=selected_record.params,
        tags=["synthetic-fixture", "review_only"],
        settings=wandb.Settings(disable_git=True, disable_code=True),
    ) as run:
        candidate_bundle = wandb.Artifact("v1.2-fix-b-bundle", type="candidate")
        for relative_path in selected_record.artifacts:
            source = tracker.root / selected_record.run_id / relative_path
            candidate_bundle.add_file(str(source), name=relative_path)
        run.log(selected_record.metrics)
        run.log_artifact(candidate_bundle)
        print(f"wandb_mode={run.settings.mode}")
        print(f"stored_template={run.config['template_id']}")
    print(f"local_run_files_exist={any(Path(directory).rglob('*.wandb'))}")

wandb_mode=offline
stored_template=deploy-answerer-v1.2-fix-b
local_run_files_exist=True

This checks local SDK recording, not hosted artifact lineage, permissions, dashboards, or synchronization. Online use_artifact needs the approved evaluation artifact to exist in that service. Neither a local success nor a mock API call verifies those remote contracts.

Trace LLM behavior with W&B Weave

W&B Models supplies run and artifact tracking. Weave focuses on AI application traces and evaluation. Use them as complementary views: run tracking applies to LLM experiments too, while a per-case trace can help explain what an aggregate score conceals.^{[15]Reference 15Weave: Tracing and Evaluations for LLM Applicationshttps://github.com/wandb/weave}

Two Weave pieces map onto the record you just built. The @weave.op decorator marks a function as a tracked operation. Once weave.init(...) initializes a project, calls record inputs, outputs, latency, and parent-child relationships; supported provider integrations capture nested model calls too. For multi-turn agents, Weave groups per-turn traces into threads; nested model, retrieval, and tool calls form call trees similar to OpenTelemetry spans.^{[15]Reference 15Weave: Tracing and Evaluations for LLM Applicationshttps://github.com/wandb/weave}

Follow one evaluation row

exact_route_rate = 1.0 means the categorical routes matched their labels. It establishes neither factual answer quality nor acceptable latency and cost. A real candidate could preserve task quality while increasing token usage; measure the trade-off on paired cases instead of inferring it from prompt length alone.

Imagine an instrumented request with a root span and three sequential child spans: retrieval, generation, and guardrail. Embedding, index query, and reranking may be children of retrieval. Sequential operations needn't be nested inside one another. A trace contains the operations actually instrumented; missing spans don't prove missing work.

If retrieval takes 420 ms and generation takes 1,850 ms in a hypothetical trace, generation is the larger observed component. Those timings don't establish provider overload or database contention. Investigate queueing, retries, provider timings, and resource evidence before naming the cause. Record client time to first visible chunk separately from generation duration, with units and start/end boundaries.

For a provider whose rates are dollars per million tokens, let $n_u$, $n_c$, and $n_o$ count uncached input, cached input, and billed output tokens. Let $p_u$, $p_c$, and $p_o$ be their respective prices per million tokens. The estimated token cost in dollars is:

Here ordinary and cached input counts are disjoint. Bind the rate card to the provider, model, date, and relevant billing tier. Add other billable categories when applicable, and avoid counting reasoning tokens twice if they are included in billed output. This estimate is not an invoice.

An evaluation row can link a case ID, parsed decision, reviewer outcome, approved input/artifact versions, timings, and usage to its trace. Raw requests, context, and completions may contain secrets or personal data. Capture them only under an approved policy, applying redaction before logging. A pointer or digest provides lineage but cannot reconstruct an artifact that was never retained. Preserve enough permitted evidence for review without assuming every platform automatically stores every field.

The Evaluation object is a blueprint: a dataset, candidate function, and scorers. Each .evaluate() call executes that comparison. Use a scorer argument named output to receive the candidate's result; additional arguments match dataset columns. These scorers return dictionaries. The Python SDK's evaluate() is async, so a script can use asyncio.run(...); in an already-running notebook event loop, use await.^{[16]Reference 16Scoring overviewhttps://docs.wandb.ai/weave/guides/evaluation/scorers[17]Reference 17Weave Python SDKhttps://docs.wandb.ai/weave/reference/python-sdk}

The snippet below defines a hosted integration but doesn't initialize a project or execute it. Calling run_hosted_evaluation() would send traced inputs and outputs to the configured W&B project. Review data retention and redact content before that boundary, not after it's logged:

import asyncio
import weave
from weave import Evaluation as WeaveEvaluation

@weave.op()
def candidate_route(evidence_state: str) -> str:
    supported = {"release_record", "rollback_runbook"}
    return "SERVE" if evidence_state in supported else "ABSTAIN"

@weave.op()
def route_matches(expected_route: str, output: str) -> dict:
    return {"exact_route": output == expected_route}

evaluation = WeaveEvaluation(
    dataset=[
        {"evidence_state": "no_approval_support", "expected_route": "ABSTAIN"},
        {"evidence_state": "rollback_runbook", "expected_route": "SERVE"},
    ],
    scorers=[route_matches],
)
async def run_hosted_evaluation():
    weave.init("deploy-claim-gate-repair")
    return await evaluation.evaluate(candidate_route)

# Explicit remote action after authorization:
# asyncio.run(run_hosted_evaluation())

This evaluation harness attaches scores to the operations that produced them. Keep dataset and scorer versions in the comparison record, and inspect failed rows alongside their traces.

Keep rejected runs in history

Choose a platform that fits the team's storage, access, dashboards, and deployment workflow. Either one can faithfully record a misleading metric if the inputs and meaning were never defined.

A rejected run is part of the explanation for the next candidate. Keep the failed first fix visible so a reviewer can see what v1.2-fix-b had to correct:

Don't trust a zero counter by itself. The next cell reads the categorical report, requires each canonical case exactly once, checks the declared candidate's deterministic route, and recomputes the metrics. Missing evidence or inconsistent numbers block nomination. This verifies the local simulation, not factual safety, authenticity of an external job, or permission to deploy.

def verified_result(run: RunRecord) -> Evaluation | None:
    if type(run) is not RunRecord:
        return None
    if any(type(value) is not dict for value in (run.params, run.metrics, run.tags)) or type(run.artifacts) is not list:
        return None
    expected_context = {
        "regression_sha": regression_fingerprint,
        "holdout_sha": holdout_fingerprint,
        "evaluation_sha": evaluation_fingerprint,
        "policy_id": handoff.incident_policy_id,
        "evidence_version": handoff.evidence_version,
        "evaluator_version": evaluator_version,
    }
    try:
        if any(run.params.get(key) != value for key, value in expected_context.items()):
            return None
        if run.tags.get("incident_request_id") != handoff.exemplar_request_id:
            return None
        require_text(run.params.get("code_commit"), "code provenance")
        template_id = run.params["template_id"]
        expected = evaluate(template_id, all_cases)
        report_name = f"reports/{template_id}-redacted-eval.json"
        template_name = f"templates/{template_id}.json"
        if report_name not in run.artifacts or template_name not in run.artifacts:
            return None
        report = json.loads(tracker.artifact_path(run, report_name).read_text())
        manifest = json.loads(tracker.artifact_path(run, template_name).read_text())
        if type(manifest) is not dict or type(manifest.get("simulation_only")) is not bool:
            return None
        if manifest != {
            "template_id": template_id,
            "implementation": "predict_route in evaluate-templates.py",
            "simulation_only": True,
        }:
            return None
        if type(report) is not list or len(report) != len(all_cases):
            return None
        canonical = {case.case_id: case for case in all_cases}
        seen = set()
        for row in report:
            if type(row) is not dict or type(row.get("case_id")) is not str:
                return None
            case_id = row["case_id"]
            if case_id not in canonical or case_id in seen:
                return None
            case = canonical[case_id]
            expected_row = {**asdict(case), "predicted_route": predict_route(template_id, case)}
            if row != expected_row:
                return None
            seen.add(case_id)
        for key, value in asdict(expected).items():
            if key == "template_id":
                continue
            logged = run.metrics.get(key)
            if type(logged) not in (int, float) or logged != value or not math.isfinite(logged):
                return None
        return expected
    except (KeyError, TypeError, ValueError, OSError):
        return None

def review_status(run: RunRecord) -> str:
    result = verified_result(run)
    if result is None:
        return "BLOCK_INCOMPLETE_OR_INCONSISTENT_EVIDENCE"
    if result.unsafe_route_decisions > 0:
        return "REJECT_UNSAFE_ROUTE"
    if result.unnecessary_abstention > 0:
        return "REJECT_UNNECESSARY_ABSTENTION"
    return "DEVELOPMENT_CHECKS_PASS_FRESH_EVALUATION_PENDING"

for run in tracker.runs:
    print(f"{run.params['template_id']}={review_status(run)}")

deploy-answerer-v1.1-regression=REJECT_UNSAFE_ROUTE
deploy-answerer-v1.2-fix=REJECT_UNNECESSARY_ABSTENTION
deploy-answerer-v1.2-fix-b=DEVELOPMENT_CHECKS_PASS_FRESH_EVALUATION_PENDING

Before trusting the passing record, try breaking its evidence. The following cell changes the reviewed run one field at a time and restores it afterward. A missing metric, a non-finite score, an implausibly large count, and a duplicated report row must all block nomination. Matching tags alone cannot repair those defects.

record = tracker.runs[2]
saved_metrics = dict(record.metrics)
report_file = tracker.artifact_path(record, record.artifacts[0])
saved_report = report_file.read_text()
blocked = []
try:
    for bad_count in (None, float("nan"), 10**1000):
        record.metrics = dict(saved_metrics)
        if bad_count is None:
            record.metrics.pop("case_count")
        else:
            record.metrics["case_count"] = bad_count
        blocked.append(review_status(record).startswith("BLOCK_"))
    record.metrics = dict(saved_metrics)
    duplicate_rows = json.loads(saved_report)
    duplicate_rows[-1] = duplicate_rows[0]
    report_file.write_text(json.dumps(duplicate_rows))
    blocked.append(review_status(record).startswith("BLOCK_"))
finally:
    record.metrics = saved_metrics
    report_file.write_text(saved_report)
assert all(blocked)
print(f"corrupted_records_blocked={sum(blocked)}/{len(blocked)}")
print(f"restored_status={review_status(record)}")

corrupted_records_blocked=4/4
restored_status=DEVELOPMENT_CHECKS_PASS_FRESH_EVALUATION_PENDING

Let the evidence earn a limited decision

Passing these development metrics nominates a candidate for review; it doesn't authorize traffic. A release review still needs fresh held-out evidence, representative slices, operational limits, and an owner. Retain a known-good pointer for canary abort and for later production rollback if the candidate is promoted.

For this prompt-only repair, canary_candidate can identify an evaluated prompt version after release review. Record the effective model configuration too. A Model Registry package may be appropriate when versioning inference code or trained weights along with the application; neither registry makes the deployment decision for you.^{[8]Reference 8Prompt Registryhttps://mlflow.org/docs/latest/genai/prompt-registry/index.html[6]Reference 6Model Registry Workflows | MLflow AI Platformhttps://mlflow.org/docs/latest/ml/model-registry/workflow/}

Only run_003 passes this development contract. Record the proposed choice with its source run, evaluation SHA, known-good pointer, and limitation. This dataclass doesn't update a registry or deploy anything:

@dataclass(frozen=True)
class PromotionDecision:
    artifact_name: str
    alias: str
    source_run_id: str
    evaluation_sha: str
    known_good_artifact: str
    status: str
    limitation: str

def passes_contract(run: RunRecord) -> bool:
    result = verified_result(run)
    return (
        result is not None
        and result.unsafe_route_decisions == 0
        and result.unnecessary_abstention == 0
    )

def only_eligible_run(runs: list[RunRecord]) -> RunRecord:
    eligible = [run for run in runs if passes_contract(run)]
    if len(eligible) != 1:
        raise ValueError(f"expected one eligible run, found {len(eligible)}")
    return eligible[0]

selected = only_eligible_run(tracker.runs)
decision = PromotionDecision(
    artifact_name=selected.params["template_id"],
    alias="canary_candidate",
    source_run_id=selected.run_id,
    evaluation_sha=selected.params["evaluation_sha"],
    known_good_artifact="deploy-answerer-v1",
    status="READY_FOR_REVIEW_NOT_DEPLOYED",
    limitation="requires fresh held-out evaluation and release approval",
)

print(f"selected={decision.artifact_name}")
print(f"source_run={decision.source_run_id}")
print(f"alias={decision.alias}")
print(f"status={decision.status}")
print(f"known_good={decision.known_good_artifact}")
print(f"limitation={decision.limitation}")

selected=deploy-answerer-v1.2-fix-b
source_run=run_003
alias=canary_candidate
status=READY_FOR_REVIEW_NOT_DEPLOYED
known_good=deploy-answerer-v1
limitation=requires fresh held-out evaluation and release approval

One run clears this fixture's development checks. A real review must resolve ties explicitly instead of silently selecting the first passing candidate. The artifact, source run, fixed suite, known-good pointer, and limitation now travel together. Local records can be changed. A real release needs provenance from the actual evaluation and retained artifacts, so an investigator can identify which reviewed configuration shipped.

Promotion evidence should define the stop action before canary traffic begins. This check makes the abort rule executable for any future unsafe serve while the candidate is only on the canary alias. Production rollback applies only after production has moved to the candidate. The full alias-state treatment, including aborting a canary versus rolling back a promotion, is in Model Versioning & Continuous Deployment.

These two counters are hypothetical reviewed claim outcomes for a future canary, not the categorical routes measured above. A complete window with no known unsafe claim continues monitoring; one unsafe claim aborts and keeps the known-good artifact. Unknown review coverage pauses the canary instead of declaring a clean window:

def canary_action(
    unsafe_claim_served: int, known_good_artifact: str, *, review_complete: bool,
) -> str:
    if type(unsafe_claim_served) is not int or unsafe_claim_served < 0:
        raise ValueError("unsafe count must be a nonnegative integer")
    require_text(known_good_artifact, "known-good artifact")
    if type(review_complete) is not bool:
        raise ValueError("review_complete must be boolean")
    # Candidate isn't production yet: stop canary traffic and keep production on known-good.
    if unsafe_claim_served > 0:
        return f"ABORT_CANARY_KEEP:{known_good_artifact}"
    if not review_complete:
        return "PAUSE_CANARY_REVIEW_COVERAGE_UNKNOWN"
    return "CONTINUE_CANARY_MONITORING"

print(f"clean_window={canary_action(0, decision.known_good_artifact, review_complete=True)}")
print(f"unsafe_window={canary_action(1, decision.known_good_artifact, review_complete=True)}")
print(f"unknown_window={canary_action(0, decision.known_good_artifact, review_complete=False)}")

clean_window=CONTINUE_CANARY_MONITORING
unsafe_window=ABORT_CANARY_KEEP:deploy-answerer-v1
unknown_window=PAUSE_CANARY_REVIEW_COVERAGE_UNKNOWN

Carry the record into compiled prompt programs

This lab simulated hand-written routing revisions. The same experiment record can compare actual prompt programs compiled by an optimizer:

Before a routing repair earns review, the record has to name its inputs, scores, artifacts, and limits. The same record has to travel with a compiled program. A compiled program can look better because it saw the holdout set, changed evaluator semantics, or dropped a safety slice. Preserve optimizer configuration, dataset splits, metric version, compiled state, and rejected candidates together.