model-registry-governance

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Model Registry Governance

模型注册治理

Create a trustworthy system of record for model artifacts, prompts, adapters, and evaluation evidence.

为模型工件、提示词、适配器和评估证据创建可信的记录系统。

When to Use This Skill

何时使用此技能

Setting up a centralized model registry for your organization
Defining metadata standards for model artifacts
Building approval workflows for model promotion to production
Implementing lifecycle policies for model retirement
Preparing for compliance audits of AI systems

为组织搭建中心化模型注册库
定义模型工件的元数据标准
构建模型上线至生产环境的审批工作流
实施模型退役的生命周期策略
为AI系统合规审计做准备

Prerequisites

前提条件

MLflow Tracking Server or Weights & Biases instance deployed
Object storage for model artifacts (S3, GCS, or MinIO)
CI/CD pipeline with access to the registry API
OPA or similar policy engine for governance checks
Git repository for policy definitions and promotion scripts

已部署MLflow Tracking Server或Weights & Biases实例
用于存储模型工件的对象存储（S3、GCS或MinIO）
可访问注册库API的CI/CD流水线
用于治理检查的OPA或类似策略引擎
用于存储策略定义和上线脚本的Git仓库

Core Principles

核心原则

Traceability: every production model maps to source code, data snapshot, and evaluation results.
Reproducibility: builds are deterministic with pinned dependencies.
Policy-driven promotion: no manual bypass for critical safety checks.
Lifecycle hygiene: stale, vulnerable, or unowned models are retired automatically.

可追溯性：每个生产模型都对应源代码、数据快照和评估结果。
可复现性：构建过程是确定性的，依赖项已固定版本。
策略驱动的上线：关键安全检查不允许手动绕过。
生命周期整洁性：过时、存在漏洞或无人维护的模型会自动退役。

MLflow Registry Setup

MLflow注册库设置

bash

undefined

bash

undefined

Install MLflow with required backends

安装带有所需后端的MLflow

pip install mlflow[extras] psycopg2-binary boto3

Start MLflow tracking server with PostgreSQL backend and S3 artifact store

启动带有PostgreSQL后端和S3工件存储的MLflow跟踪服务器

mlflow server
--backend-store-uri postgresql://mlflow:password@db:5432/mlflow
--default-artifact-root s3://mlflow-artifacts/models
--host 0.0.0.0
--port 5000
--serve-artifacts


```yaml

mlflow server
--backend-store-uri postgresql://mlflow:password@db:5432/mlflow
--default-artifact-root s3://mlflow-artifacts/models
--host 0.0.0.0
--port 5000
--serve-artifacts


```yaml

docker-compose.yaml for MLflow

MLflow的docker-compose.yaml

services: mlflow: image: ghcr.io/mlflow/mlflow:2.12.0 command: > mlflow server --backend-store-uri postgresql://mlflow:${DB_PASSWORD}@db:5432/mlflow --default-artifact-root s3://mlflow-artifacts/models --host 0.0.0.0 --port 5000 --serve-artifacts ports: - "5000:5000" environment: AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID} AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY} depends_on: - db

db: image: postgres:16-alpine environment: POSTGRES_DB: mlflow POSTGRES_USER: mlflow POSTGRES_PASSWORD: ${DB_PASSWORD} volumes: - pgdata:/var/lib/postgresql/data

volumes: pgdata:

undefined

db: image: postgres:16-alpine environment: POSTGRES_DB: mlflow POSTGRES_USER: mlflow POSTGRES_PASSWORD: ${DB_PASSWORD} volumes: - pgdata:/var/lib/postgresql/data

volumes: pgdata:

undefined

Required Metadata Schema

必需的元数据架构

python

undefined

python

undefined

model_metadata_schema.py

from pydantic import BaseModel, Field from typing import List, Optional from datetime import datetime from enum import Enum

class LifecycleState(str, Enum): DRAFT = "draft" CANDIDATE = "candidate" APPROVED = "approved" DEPRECATED = "deprecated" RETIRED = "retired"

class RiskRating(str, Enum): LOW = "low" MEDIUM = "medium" HIGH = "high" CRITICAL = "critical"

class ModelMetadata(BaseModel): """Required metadata for every registered model.""" # Identity name: str = Field(description="Model name matching registry key") version: str = Field(description="Semantic version") checksum: str = Field(description="SHA-256 of model artifact") storage_uri: str = Field(description="Artifact store path")

# Lineage
base_model: str = Field(description="Parent model identifier")
fine_tune_method: Optional[str] = Field(default=None)
training_dataset: Optional[str] = Field(default=None)
training_date: Optional[datetime] = Field(default=None)
source_commit: str = Field(description="Git SHA of training code")

# Evaluation
eval_datasets: List[str] = Field(description="Evaluation dataset IDs")
eval_report_uri: str = Field(description="Path to evaluation results")
quality_score: float = Field(ge=0, le=1)
safety_score: float = Field(ge=0, le=1)

# Governance
license: str = Field(description="SPDX license identifier")
allowed_use_cases: List[str]
prohibited_use_cases: List[str]
risk_rating: RiskRating
security_controls: List[str]

# Ownership
owner: str = Field(description="Primary owner email")
backup_owner: str = Field(description="Backup owner email")
escalation_contact: str
team: str

# Lifecycle
state: LifecycleState = LifecycleState.DRAFT
created_at: datetime = Field(default_factory=datetime.utcnow)
approved_at: Optional[datetime] = None
approved_by: Optional[str] = None
expires_at: Optional[datetime] = None

undefined

from pydantic import BaseModel, Field from typing import List, Optional from datetime import datetime from enum import Enum

class LifecycleState(str, Enum): DRAFT = "draft" CANDIDATE = "candidate" APPROVED = "approved" DEPRECATED = "deprecated" RETIRED = "retired"

class RiskRating(str, Enum): LOW = "low" MEDIUM = "medium" HIGH = "high" CRITICAL = "critical"

class ModelMetadata(BaseModel): """每个注册模型的必需元数据。""" # 标识信息 name: str = Field(description="与注册库键匹配的模型名称") version: str = Field(description="语义化版本") checksum: str = Field(description="模型工件的SHA-256校验值") storage_uri: str = Field(description="工件存储路径")

#  lineage
base_model: str = Field(description="父模型标识符")
fine_tune_method: Optional[str] = Field(default=None)
training_dataset: Optional[str] = Field(default=None)
training_date: Optional[datetime] = Field(default=None)
source_commit: str = Field(description="训练代码的Git SHA值")

# 评估信息
eval_datasets: List[str] = Field(description="评估数据集ID列表")
eval_report_uri: str = Field(description="评估结果的路径")
quality_score: float = Field(ge=0, le=1)
safety_score: float = Field(ge=0, le=1)

# 治理信息
license: str = Field(description="SPDX许可证标识符")
allowed_use_cases: List[str]
prohibited_use_cases: List[str]
risk_rating: RiskRating
security_controls: List[str]

# 归属信息
owner: str = Field(description="主要负责人邮箱")
backup_owner: str = Field(description="备份负责人邮箱")
escalation_contact: str
team: str

# 生命周期
state: LifecycleState = LifecycleState.DRAFT
created_at: datetime = Field(default_factory=datetime.utcnow)
approved_at: Optional[datetime] = None
approved_by: Optional[str] = None
expires_at: Optional[datetime] = None

undefined

Model Registration Script

模型注册脚本

python

undefined

python

undefined

register_model.py

import mlflow from mlflow.tracking import MlflowClient import json import hashlib

def register_model( model_path: str, model_name: str, metadata: dict, mlflow_uri: str = "http://mlflow:5000" ): """Register a model with full metadata and governance tags.""" mlflow.set_tracking_uri(mlflow_uri) client = MlflowClient()

# Compute artifact checksum
with open(model_path, "rb") as f:
    checksum = hashlib.sha256(f.read()).hexdigest()
metadata["checksum"] = checksum

# Log model with metadata
with mlflow.start_run(run_name=f"register-{model_name}-{metadata['version']}") as run:
    # Log all metadata as params
    mlflow.log_params({
        "model_name": model_name,
        "version": metadata["version"],
        "base_model": metadata["base_model"],
        "risk_rating": metadata["risk_rating"],
        "owner": metadata["owner"],
        "license": metadata["license"],
    })

    # Log quality metrics
    mlflow.log_metrics({
        "quality_score": metadata["quality_score"],
        "safety_score": metadata["safety_score"],
    })

    # Log full metadata as artifact
    with open("metadata.json", "w") as f:
        json.dump(metadata, f, indent=2, default=str)
    mlflow.log_artifact("metadata.json")

    # Log model artifact
    mlflow.log_artifact(model_path)

    # Register in model registry
    model_uri = f"runs:/{run.info.run_id}/model"
    result = mlflow.register_model(model_uri, model_name)

    # Set lifecycle tags
    client.set_model_version_tag(
        model_name, result.version, "state", "draft"
    )
    client.set_model_version_tag(
        model_name, result.version, "risk_rating", metadata["risk_rating"]
    )
    client.set_model_version_tag(
        model_name, result.version, "checksum", checksum
    )

return result

undefined

import mlflow from mlflow.tracking import MlflowClient import json import hashlib

def register_model( model_path: str, model_name: str, metadata: dict, mlflow_uri: str = "http://mlflow:5000" ): """使用完整元数据和治理标签注册模型。""" mlflow.set_tracking_uri(mlflow_uri) client = MlflowClient()

# 计算工件校验值
with open(model_path, "rb") as f:
    checksum = hashlib.sha256(f.read()).hexdigest()
metadata["checksum"] = checksum

# 记录带元数据的模型
with mlflow.start_run(run_name=f"register-{model_name}-{metadata['version']}") as run:
    # 将所有元数据记录为参数
    mlflow.log_params({
        "model_name": model_name,
        "version": metadata["version"],
        "base_model": metadata["base_model"],
        "risk_rating": metadata["risk_rating"],
        "owner": metadata["owner"],
        "license": metadata["license"],
    })

    # 记录质量指标
    mlflow.log_metrics({
        "quality_score": metadata["quality_score"],
        "safety_score": metadata["safety_score"],
    })

    # 将完整元数据记录为工件
    with open("metadata.json", "w") as f:
        json.dump(metadata, f, indent=2, default=str)
    mlflow.log_artifact("metadata.json")

    # 记录模型工件
    mlflow.log_artifact(model_path)

    # 在模型注册库中注册
    model_uri = f"runs:/{run.info.run_id}/model"
    result = mlflow.register_model(model_uri, model_name)

    # 设置生命周期标签
    client.set_model_version_tag(
        model_name, result.version, "state", "draft"
    )
    client.set_model_version_tag(
        model_name, result.version, "risk_rating", metadata["risk_rating"]
    )
    client.set_model_version_tag(
        model_name, result.version, "checksum", checksum
    )

return result

undefined

Approval Workflow

审批工作流

Registration request created from CI.
Security checks (artifact scan, dependency scan, provenance).
Evaluation package uploaded (quality, toxicity, jailbreak, bias, latency, cost).
Required approvals: platform + product + security (as policy dictates).
Promotion to stage/prod based on signed decision record.

由CI发起注册请求。
安全检查（工件扫描、依赖扫描、溯源验证）。
上传评估包（质量、毒性、越狱测试、偏差、延迟、成本）。
必需审批：平台团队+产品团队+安全团队（按策略要求）。
根据签署的决策记录将模型上线至预发布/生产环境。

Promotion Script

上线脚本

python

undefined

python

undefined

promote_model.py

import mlflow from mlflow.tracking import MlflowClient from datetime import datetime import sys

def promote_model( model_name: str, version: str, target_stage: str, approver: str, mlflow_uri: str = "http://mlflow:5000" ): """Promote a model version after governance checks pass.""" mlflow.set_tracking_uri(mlflow_uri) client = MlflowClient()

# Verify current state allows promotion
mv = client.get_model_version(model_name, version)
current_state = mv.tags.get("state", "draft")

valid_transitions = {
    "draft": ["candidate"],
    "candidate": ["approved", "draft"],
    "approved": ["deprecated"],
    "deprecated": ["retired"],
}

if target_stage not in valid_transitions.get(current_state, []):
    raise ValueError(
        f"Invalid transition: {current_state} -> {target_stage}. "
        f"Allowed: {valid_transitions.get(current_state, [])}"
    )

# Verify required eval scores for production promotion
if target_stage == "approved":
    run = client.get_run(mv.run_id)
    quality = float(run.data.metrics.get("quality_score", 0))
    safety = float(run.data.metrics.get("safety_score", 0))

    if quality < 0.85:
        raise ValueError(f"Quality score {quality} below threshold 0.85")
    if safety < 0.95:
        raise ValueError(f"Safety score {safety} below threshold 0.95")

# Record promotion
now = datetime.utcnow().isoformat()
client.set_model_version_tag(model_name, version, "state", target_stage)
client.set_model_version_tag(model_name, version, f"promoted_to_{target_stage}_at", now)
client.set_model_version_tag(model_name, version, f"promoted_to_{target_stage}_by", approver)

# Transition MLflow stage alias
stage_map = {
    "candidate": "Staging",
    "approved": "Production",
    "deprecated": "Archived",
}
if target_stage in stage_map:
    client.transition_model_version_stage(
        model_name, version, stage_map[target_stage]
    )

print(f"Model {model_name} v{version}: {current_state} -> {target_stage}")
print(f"Approved by: {approver} at {now}")

if name == "main": promote_model( model_name=sys.argv[1], version=sys.argv[2], target_stage=sys.argv[3], approver=sys.argv[4], )

undefined

import mlflow from mlflow.tracking import MlflowClient from datetime import datetime import sys

def promote_model( model_name: str, version: str, target_stage: str, approver: str, mlflow_uri: str = "http://mlflow:5000" ): """治理检查通过后上线模型版本。""" mlflow.set_tracking_uri(mlflow_uri) client = MlflowClient()

# 验证当前状态是否允许上线
mv = client.get_model_version(model_name, version)
current_state = mv.tags.get("state", "draft")

valid_transitions = {
    "draft": ["candidate"],
    "candidate": ["approved", "draft"],
    "approved": ["deprecated"],
    "deprecated": ["retired"],
}

if target_stage not in valid_transitions.get(current_state, []):
    raise ValueError(
        f"无效状态转换: {current_state} -> {target_stage}. "
        f"允许的转换: {valid_transitions.get(current_state, [])}"
    )

# 验证生产环境上线所需的最低评估分数
if target_stage == "approved":
    run = client.get_run(mv.run_id)
    quality = float(run.data.metrics.get("quality_score", 0))
    safety = float(run.data.metrics.get("safety_score", 0))

    if quality < 0.85:
        raise ValueError(f"质量分数 {quality} 低于阈值0.85")
    if safety < 0.95:
        raise ValueError(f"安全分数 {safety} 低于阈值0.95")

# 记录上线操作
now = datetime.utcnow().isoformat()
client.set_model_version_tag(model_name, version, "state", target_stage)
client.set_model_version_tag(model_name, version, f"promoted_to_{target_stage}_at", now)
client.set_model_version_tag(model_name, version, f"promoted_to_{target_stage}_by", approver)

# 转换MLflow阶段别名
stage_map = {
    "candidate": "Staging",
    "approved": "Production",
    "deprecated": "Archived",
}
if target_stage in stage_map:
    client.transition_model_version_stage(
        model_name, version, stage_map[target_stage]
    )

print(f"模型 {model_name} v{version}: {current_state} -> {target_stage}")
print(f"审批人: {approver} 时间: {now}")

if name == "main": promote_model( model_name=sys.argv[1], version=sys.argv[2], target_stage=sys.argv[3], approver=sys.argv[4], )

undefined

Lifecycle States

生命周期状态

State	Description	Serving Allowed	New Usage
`draft`	Internal experimentation	Dev only	Dev only
`candidate`	Passed baseline tests	Staging	Staging
`approved`	Authorized for production	All environments	Yes
`deprecated`	Replacement announced	Existing only	Blocked
`retired`	Archived for audit	None	None

状态	描述	是否允许提供服务	是否允许新用途
`draft`	内部实验阶段	仅开发环境	仅开发环境
`candidate`	通过基线测试	预发布环境	预发布环境
`approved`	已授权用于生产	所有环境	是
`deprecated`	已宣布替代方案	仅现有用途	禁止
`retired`	已归档用于审计	无	无

Lifecycle Automation

生命周期自动化

python

undefined

python

undefined

lifecycle_policy.py

from mlflow.tracking import MlflowClient from datetime import datetime, timedelta

def enforce_lifecycle_policies(mlflow_uri: str = "http://mlflow:5000"): """Run periodic lifecycle enforcement.""" client = MlflowClient()

for rm in client.search_registered_models():
    for mv in client.search_model_versions(f"name='{rm.name}'"):
        tags = mv.tags
        state = tags.get("state", "draft")

        # Auto-deprecate models with expired approvals (90 days)
        if state == "approved":
            approved_at = tags.get("promoted_to_approved_at")
            if approved_at:
                approved_date = datetime.fromisoformat(approved_at)
                if datetime.utcnow() - approved_date > timedelta(days=90):
                    print(f"Auto-deprecating {rm.name} v{mv.version}: approval expired")
                    client.set_model_version_tag(rm.name, mv.version, "state", "deprecated")
                    client.set_model_version_tag(
                        rm.name, mv.version, "auto_deprecated_reason", "approval_expired"
                    )

        # Auto-retire deprecated models after 30 days
        if state == "deprecated":
            deprecated_at = tags.get("promoted_to_deprecated_at")
            if deprecated_at:
                deprecated_date = datetime.fromisoformat(deprecated_at)
                if datetime.utcnow() - deprecated_date > timedelta(days=30):
                    print(f"Auto-retiring {rm.name} v{mv.version}")
                    client.set_model_version_tag(rm.name, mv.version, "state", "retired")
                    client.transition_model_version_stage(
                        rm.name, mv.version, "Archived"
                    )

        # Flag drafts with no activity for 14 days
        if state == "draft":
            created = datetime.fromisoformat(mv.creation_timestamp / 1000)
            if datetime.utcnow() - created > timedelta(days=14):
                print(f"Stale draft: {rm.name} v{mv.version}")

undefined

from mlflow.tracking import MlflowClient from datetime import datetime, timedelta

def enforce_lifecycle_policies(mlflow_uri: str = "http://mlflow:5000"): """定期执行生命周期策略。""" client = MlflowClient()

for rm in client.search_registered_models():
    for mv in client.search_model_versions(f"name='{rm.name}'"):
        tags = mv.tags
        state = tags.get("state", "draft")

        # 自动弃用审批过期的模型（90天）
        if state == "approved":
            approved_at = tags.get("promoted_to_approved_at")
            if approved_at:
                approved_date = datetime.fromisoformat(approved_at)
                if datetime.utcnow() - approved_date > timedelta(days=90):
                    print(f"自动弃用 {rm.name} v{mv.version}: 审批已过期")
                    client.set_model_version_tag(rm.name, mv.version, "state", "deprecated")
                    client.set_model_version_tag(
                        rm.name, mv.version, "auto_deprecated_reason", "approval_expired"
                    )

        # 自动退役已弃用30天的模型
        if state == "deprecated":
            deprecated_at = tags.get("promoted_to_deprecated_at")
            if deprecated_at:
                deprecated_date = datetime.fromisoformat(deprecated_at)
                if datetime.utcnow() - deprecated_date > timedelta(days=30):
                    print(f"自动退役 {rm.name} v{mv.version}")
                    client.set_model_version_tag(rm.name, mv.version, "state", "retired")
                    client.transition_model_version_stage(
                        rm.name, mv.version, "Archived"
                    )

        # 标记14天无活动的草稿模型
        if state == "draft":
            created = datetime.fromisoformat(mv.creation_timestamp / 1000)
            if datetime.utcnow() - created > timedelta(days=14):
                print(f"过期草稿: {rm.name} v{mv.version}")

undefined

Governance Policies (OPA/Rego)

治理策略（OPA/Rego）

rego

undefined

rego

undefined

policy/model_governance.rego

package model.governance

Reject artifacts without SBOM

拒绝不包含SBOM的工件

deny[msg] { not input.metadata.sbom_uri msg := "Model must include SBOM artifact URI" }

Block promotion if critical CVEs remain

若存在未解决的严重CVE则阻止上线

deny[msg] { input.target_state == "approved" input.security_scan.critical_cves > 0 msg := sprintf("Cannot promote: %d critical CVEs unresolved", [input.security_scan.critical_cves]) }

Require refreshed evals after prompt changes

提示词变更后需重新评估

deny[msg] { input.target_state == "approved" input.prompt_changed not input.eval_refreshed_after_prompt_change msg := "Evaluation must be re-run after prompt template changes" }

Require minimum eval scores for production

生产环境上线需满足最低评估分数

deny[msg] { input.target_state == "approved" input.metadata.quality_score < 0.85 msg := sprintf("Quality score %.2f below threshold 0.85", [input.metadata.quality_score]) }

Require dual approval for high-risk models

高风险模型需要双重审批

deny[msg] { input.target_state == "approved" input.metadata.risk_rating == "high" count(input.approvals) < 2 msg := "High-risk models require at least 2 approvals" }

undefined

deny[msg] { input.target_state == "approved" input.metadata.risk_rating == "high" count(input.approvals) < 2 msg := "High-risk models require at least 2 approvals" }

undefined

Audit Readiness

审计就绪

Maintain immutable records of:

Who approved and when
Which policy checks executed
Which exceptions were granted
What model/version served each customer request window

维护不可变的记录：

审批人及审批时间
执行的策略检查
获批的例外情况
每个客户请求时段使用的模型/版本

Troubleshooting

故障排查

Issue	Diagnosis	Resolution
Model registration fails	Check MLflow server connectivity and artifact store permissions	Verify S3/GCS credentials and bucket policy
Promotion blocked by policy	Review OPA deny messages in CI output	Fix metadata gaps or request policy exception
Stale models not auto-retiring	Lifecycle cron job not running	Check CronJob status in Kubernetes
Duplicate model versions	Race condition in CI pipeline	Add locking via registry API or database
Missing eval evidence	Eval pipeline skipped or failed	Re-run eval suite and re-register

问题	诊断	解决方案
模型注册失败	检查MLflow服务器连通性和工件存储权限	验证S3/GCS凭证和存储桶策略
策略阻止上线	查看CI输出中的OPA拒绝信息	修复元数据缺失或申请策略例外
过期模型未自动退役	生命周期定时任务未运行	检查Kubernetes中的CronJob状态
重复模型版本	CI流水线存在竞争条件	通过注册库API或数据库添加锁机制
缺失评估证据	评估流水线被跳过或执行失败	重新运行评估套件并重新注册模型

model-registry-governance

Original

Translation

Model Registry Governance

模型注册治理

When to Use This Skill

何时使用此技能

Prerequisites

前提条件

Core Principles

核心原则

MLflow Registry Setup

MLflow注册库设置

Install MLflow with required backends

安装带有所需后端的MLflow

Start MLflow tracking server with PostgreSQL backend and S3 artifact store

启动带有PostgreSQL后端和S3工件存储的MLflow跟踪服务器

docker-compose.yaml for MLflow

MLflow的docker-compose.yaml

Required Metadata Schema

必需的元数据架构

model_metadata_schema.py

model_metadata_schema.py

Model Registration Script

模型注册脚本

register_model.py

register_model.py

Approval Workflow

审批工作流

Promotion Script

上线脚本

promote_model.py

promote_model.py

Lifecycle States

生命周期状态

Lifecycle Automation

生命周期自动化

lifecycle_policy.py

lifecycle_policy.py

Governance Policies (OPA/Rego)

治理策略（OPA/Rego）

policy/model_governance.rego

policy/model_governance.rego

Reject artifacts without SBOM

拒绝不包含SBOM的工件

Block promotion if critical CVEs remain

若存在未解决的严重CVE则阻止上线

Require refreshed evals after prompt changes

提示词变更后需重新评估

Require minimum eval scores for production

生产环境上线需满足最低评估分数

Require dual approval for high-risk models

高风险模型需要双重审批

Audit Readiness

审计就绪

Troubleshooting

故障排查

Related Skills

相关技能