Let callers add MLflow tags to a fakequant serve's run (#2364)

### What does this PR do?

Type of change: new feature

The quantization run records what this library can see — the model, the
checkpoint, the vLLM and ModelOpt versions — but nothing about the
harness that launched it. A downstream tool that wants its own revision,
a sweep id, or a ticket number on the run has no way to put it there
today:

- `_run_tags()` returns a fixed dict
- `quant_config` (which becomes the run's params) is a hardcoded set of
`QUANT_*` variables
- MLflow itself has no environment variable for arbitrary tags

`MODELOPT_MLFLOW_EXTRA_TAGS` takes comma-separated `key=value` pairs and
merges them into the run's tags.

Two details worth a reviewer's attention:

**It joins `MLFLOW_ENV_VARS`.** A Ray-backed serve receives only the
variables named there, and the tracker runs in the rank-0 worker —
omitting it would make the feature silently do nothing under Ray.

**Caller tags are merged first**, so the library's own keys (`tool`,
`model`, `checkpoint_path`, `vllm_version`) are written over them and
keep describing the run truthfully whatever a caller sends.

`key=value` rather than JSON, learned from a live run: the variable
reaches the worker through a shell `export VAR="..."`, and JSON's own
double quotes terminate that quoting —

```
export MODELOPT_MLFLOW_EXTRA_TAGS_732b_DEPLOYMENT="{"internal_version": "4d8c"}"
```

arrived as `{`. A quote-free format survives verbatim and needs no
`json` import or exception handling. Splitting on the first `=` keeps
values that contain one, such as a URL with a query string.

### Usage

```bash
export MODELOPT_MLFLOW_EXTRA_TAGS="modelopt_internal_version=49fa29d5,sweep=kv-study"
python3 vllm_serve_fakequant.py "$MODEL" --mlflow https://your-mlflow-server/ ...
```

### Testing

Unit-level, over the helper: unset and empty variable, one and several
pairs, surrounding whitespace, an empty value, an entry with no `=`, a
trailing comma, and a value containing `=`. None raise; malformed
entries warn and are skipped.

End to end on a real fakequant serve (Nemotron-3-Nano-30B-A3B BF16,
`NVFP4_DEFAULT_CFG`, TP=8, Ray executor, vLLM 0.15, SLURM):

```
modelopt_internal_version  '49fa29d5'
modelopt_version           '0.47.0rc0.post32+gd38ed5ead'
git_sha                    'd38ed5ead'
quant_cfg                  'NVFP4_DEFAULT_CFG'
```

The tag was written by the `RayWorkerWrapper` process, which exercises
the whole path — env var → shell export → `--container-env` → raylet →
Ray actor → `_run_tags` — and confirms the `MLFLOW_ENV_VARS` entry is
doing its job. Also verified that the emitted payload survives a shell
export round-trip unchanged.

### Before your PR is "*Ready for review*"

- Is this change backward compatible?: ✅ <!--- Additive; with the
variable unset the tags are exactly as before. -->
- If you copied code from any other sources or added a new PIP
dependency, did you follow guidance in `CONTRIBUTING.md`: N/A
- Did you write any new necessary tests?: ❌ <!--- Verified manually as
above; there is no existing test module for vllm_mlflow_utils. Happy to
add one if you would like it. -->
- Did you update
[Changelog](https://github.com/NVIDIA/Model-Optimizer/blob/main/CHANGELOG.rst)?:
❌ <!--- Small additive feature in an example; tell me if it warrants an
entry. -->
- Did you get Claude approval on this PR?: ❌

### Additional Information

Consumed by Model-Optimizer-Internal MR !141/!147, which sets the
variable so a fakequant eval records the same harness commit on both its
quantization run and its evaluation-score run.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Signed-off-by: Chenjie Luo <chenjiel@nvidia.com>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

examples/vllm_serve/vllm_mlflow_utils.py

@@ -60,6 +60,7 @@ EXPERIMENT_ENV = "MLFLOW_EXPERIMENT_NAME"
 RUN_NAME_ENV = "MODELOPT_MLFLOW_RUN_NAME"
 REQUIRED_ENV = "MODELOPT_MLFLOW_REQUIRED"
 COMMAND_ENV = "MODELOPT_MLFLOW_COMMAND"
+EXTRA_TAGS_ENV = "MODELOPT_MLFLOW_EXTRA_TAGS"
 
 # Everything the rank-0 worker needs in its environment to reach the tracking server. The
 # credentials are never set here, only forwarded when the launching shell exported them --
@@ -71,6 +72,7 @@ MLFLOW_ENV_VARS = frozenset(
         RUN_NAME_ENV,
         REQUIRED_ENV,
         COMMAND_ENV,
+        EXTRA_TAGS_ENV,
         "MLFLOW_TRACKING_TOKEN",
         "MLFLOW_TRACKING_USERNAME",
         "MLFLOW_TRACKING_PASSWORD",
@@ -375,6 +377,30 @@ def _vllm_params(worker: Any) -> dict[str, Any]:
     return {k: v for k, v in params.items() if v is not None}
 
 
+def _extra_tags() -> dict[str, str]:
+    """Caller-supplied tags from ``$MODELOPT_MLFLOW_EXTRA_TAGS``: ``key=value`` pairs,
+    comma separated.
+
+    For what this library cannot know -- the revision of the harness that launched the
+    serve, a sweep id -- so a caller can join its own records to this run.
+
+    Deliberately not JSON. The variable reaches the worker through a shell
+    ``export VAR="..."``, and JSON's own double quotes terminate that quoting, so the
+    value arrives truncated at the first one. Malformed entries are reported and skipped:
+    a typo in a tag must not lose a serve that has already spent minutes getting here.
+    """
+    tags, dropped = {}, []
+    for item in (os.environ.get(EXTRA_TAGS_ENV) or "").split(","):
+        key, sep, value = item.partition("=")
+        if sep and key.strip() and value.strip():
+            tags[key.strip()] = value.strip()
+        elif item.strip():
+            dropped.append(item.strip())
+    if dropped:
+        warnings.warn(f"Ignoring malformed entries in ${EXTRA_TAGS_ENV}: {dropped}")
+    return tags
+
+
 def _run_tags(worker: Any) -> dict[str, str]:
     """Tags shared with ``hf_ptq``, so a checkpoint's PTQ run and the serves of it join up.
 
@@ -384,6 +410,7 @@ def _run_tags(worker: Any) -> dict[str, str]:
     """
     model = _stringify(_model_config_value(worker, "model")) or "unknown"
     tags = {
+        **_extra_tags(),  # first, so the keys below always win
         "tool": TOOL_NAME,
         "model": Path(model).name,
         "checkpoint_path": _resolved(model),