企业官网建设流程全解析

1. 项目概述：这不是在搭平台，是在给AI系统“打地基”

“Steps toward MLOps research — Software Engineering your AI”这个标题乍看像学术论文的副标题，但实际它戳中了当前AI落地最痛的软肋——我们能调出98%准确率的模型，却经常卡在把它变成每天稳定跑、可回滚、能追责、被业务方信任的线上服务这一步。我带过7个从0到1的AI产品化项目，其中5个在模型上线后3个月内因数据漂移未告警、特征计算逻辑不一致、回滚失败导致线上资损，最后不是推倒重来，就是降级为离线报表。所谓MLOps research，根本不是去造新轮子，而是把软件工程里锤炼了半个世纪的那套东西——版本控制、接口契约、自动化测试、可观测性、环境隔离、变更审计——原样移植到AI系统生命周期里。它不解决“怎么让模型更准”，而是解决“怎么让准的模型持续准下去”。关键词里的Software Engineering是题眼：你写的不是一段Python脚本，而是一个需要被运维、被监控、被多人协作、被合规审查的生产级软件系统。适合三类人深度参考：一是刚从算法岗转MLOps岗的工程师，需要补全工程思维断层；二是技术负责人，正为AI项目交付周期长、故障定位难、跨团队协作成本高而头疼；三是高校研究者，想把实验室成果真正推向工业场景，而非止步于arXiv上的SOTA数字。这篇文章不讲Kubeflow或MLflow的安装命令，而是拆解我在金融风控、智能客服、工业质检三个领域踩坑后，亲手打磨出的一套最小可行MLOps研究框架——它用不到200行代码定义核心契约，用Git做唯一真相源，用单元测试覆盖特征逻辑，用轻量级日志实现全链路追踪。你可以直接抄作业，也可以按需裁剪。

2. 核心设计思路：为什么放弃“大平台”，选择“契约驱动”的轻量架构

2.1 拒绝“先建平台再填内容”的陷阱

很多团队一上来就立项搭建MLOps平台，采购Kubeflow、自研调度引擎、对接Prometheus监控，6个月投入200人日，结果发现80%的模型连Docker镜像都没打出来，还在用Jupyter Notebook手动改参数。我见过最典型的失败案例：某电商公司花300万建了“AI中台”，但算法同学依然把训练脚本发邮件给运维，运维手工拷贝到GPU服务器上运行——因为平台要求所有模型必须用TensorFlow 2.8+Keras API，而他们主力的时序预测模型基于PyTorch Lightning，改造成本远超预期。问题根源在于混淆了“工具链”和“工程实践”：平台是载体，契约才是灵魂。就像建筑行业不会先盖楼再设计承重标准，AI系统必须先定义清楚“什么是可部署的模型”、“什么是可信的特征”、“什么是可复现的实验”。因此，我的方案彻底绕开平台选型，从**代码即契约（Code as Contract）**出发，用极简的Python类强制约束关键行为。

2.2 “三契约一中心”架构的底层逻辑

整个框架只包含4个核心抽象，全部通过Python协议（Protocol）和装饰器实现，不依赖任何外部框架：

• ModelContract：规定模型必须提供predict()方法，输入为pd.DataFrame，输出为np.ndarray或pd.Series，且必须有get_metadata()返回字典（含模型版本、训练时间、特征列表）。拒绝接受model.predict(x)这种黑盒调用，强制暴露输入输出schema。

• FeatureContract：每个特征计算模块必须继承该基类，实现compute()方法，并声明depends_on字段（如["user_id", "order_time"]）。系统据此自动构建特征血缘图，当上游原始表字段变更时，可精准定位影响范围。

• ExperimentContract：实验必须封装为独立函数，用@experiment装饰器标记，自动注入run_id、config、logger。所有实验日志、指标、中间产物（如特征重要性图）均按run_id归档，杜绝“上次跑的结果在哪？”的混乱。

• PipelineCenter：唯一中心化协调器，不负责执行，只做三件事：校验所有组件是否满足契约、生成DAG执行计划、记录全局trace_id。它像交通指挥中心，不造车也不修路，只确保每辆车按规则上路。

提示：这个设计刻意规避了“统一调度引擎”。实践中，训练任务走Airflow，实时推理走K8s Deployment，批处理走Spark，它们通过PipelineCenter发布的标准化API（如/v1/predict?model_id=credit_v2.1）交互。解耦带来的是韧性——当Airflow集群宕机，不影响线上推理服务。

2.3 为什么坚持“Git作为唯一真相源”

有人质疑：“模型权重文件动辄GB，Git怎么存？”我的答案是：Git不存权重，只存权重指针。具体操作分三层：

1. 代码层：所有模型定义、特征逻辑、实验配置均以.py文件形式提交Git，每次commit生成唯一SHA；
2. 元数据层：PipelineCenter在执行时，将本次运行关联的Git SHA、模型权重MD5、数据集版本号、硬件环境哈希值写入SQLite本地数据库（mlruns.db），该库随代码库一起Git管理；
3. 二进制层：权重文件存OSS/S3，路径格式为{project}/{git_sha}/{model_name}/weights.pth，由PipelineCenter在运行时动态拼接下载。

这样做的好处是：回滚只需git checkout abc123 && python train.py，所有依赖（代码、配置、数据版本、环境）自动锁定。我曾用此法在支付风控模型线上异常时，5分钟内完成从最新版回退到7天前稳定版本，而传统方式需手动比对12个配置文件、3个数据ETL脚本、2个模型加载逻辑。

3. 核心细节解析：从契约定义到可执行代码的完整实现

3.1 ModelContract：让模型“开口说话”的强制接口

真正的工程化起点，是让模型不再是个黑盒。以下是我实际项目中使用的ModelContract基类，它强制模型暴露足够信息供运维使用：

from typing import Dict, Any, Union, Optional, Protocol import pandas as pd import numpy as np from dataclasses import dataclass import json @dataclass class ModelMetadata: model_name: str version: str training_time: str # ISO format input_features: list[str] output_schema: Dict[str, str] # {"score": "float32", "risk_level": "category"} training_data_version: str git_commit: str class ModelContract(Protocol): def predict(self, X: pd.DataFrame) -> Union[np.ndarray, pd.Series]: ... def get_metadata(self) -> ModelMetadata: ... # 实际模型必须实现此接口，例如风控模型 class CreditRiskModel: def __init__(self, model_path: str): self.model = load_model(model_path) # 假设的加载逻辑 self._metadata = ModelMetadata( model_name="credit_risk", version="2.1.0", training_time="2024-03-15T14:22:01Z", input_features=["age", "income", "loan_amount", "employment_years"], output_schema={"default_prob": "float32", "risk_score": "int32"}, training_data_version="v2024q1", git_commit="a1b2c3d4e5f6" ) def predict(self, X: pd.DataFrame) -> np.ndarray: # 强制校验输入列 missing_cols = set(self._metadata.input_features) - set(X.columns) if missing_cols: raise ValueError(f"Missing input features: {missing_cols}") return self.model.predict(X[self._metadata.input_features].values) def get_metadata(self) -> ModelMetadata: return self._metadata

关键细节在于predict()方法中的输入校验。很多线上事故源于特征工程脚本更新后，训练数据列名从user_age改为age，但线上服务仍按旧名取列，导致传入全NaN数组，模型输出随机值。此处强制校验列名，错误在预测第一秒就抛出，而非等到业务资损发生。get_metadata()返回结构化字典，使监控系统能自动提取model_name、version打标到Prometheus指标，无需人工维护配置。

3.2 FeatureContract：用声明式依赖终结“特征幽灵”

特征计算是AI系统最易腐烂的部分。一个叫user_lifetime_value的特征，可能在A模型中用近30天订单金额加权，B模型中用历史总消费额，C模型中又引入RFM分群。当数据团队修改基础用户表时，没人知道哪些模型会崩。FeatureContract通过声明式依赖解决此问题：

from abc import ABC, abstractmethod from typing import List, Dict, Any class FeatureContract(ABC): @property @abstractmethod def depends_on(self) -> List[str]: """声明本特征直接依赖的原始字段，如['user_id', 'order_time', 'amount']""" pass @property @abstractmethod def name(self) -> str: """特征唯一标识符，用于缓存键和血缘追踪""" pass @abstractmethod def compute(self, raw_data: pd.DataFrame) -> pd.Series: """核心计算逻辑，输入为包含depends_on字段的DataFrame""" pass # 示例：实时用户活跃度特征 class UserActivityScore(FeatureContract): def __init__(self, window_hours: int = 24): self.window_hours = window_hours @property def depends_on(self) -> List[str]: return ["user_id", "event_time", "event_type"] @property def name(self) -> str: return f"user_activity_{self.window_hours}h" def compute(self, raw_data: pd.DataFrame) -> pd.Series: # 确保输入包含必要字段 assert all(col in raw_data.columns for col in self.depends_on), \ f"Missing required columns for {self.name}" # 时间窗口计算逻辑（简化版） cutoff = raw_data['event_time'].max() - pd.Timedelta(hours=self.window_hours) recent_events = raw_data[raw_data['event_time'] >= cutoff] score = recent_events.groupby('user_id').size() return score.reindex(raw_data['user_id'].unique(), fill_value=0) # 血缘分析：自动构建依赖图 def build_feature_lineage(features: List[FeatureContract]) -> Dict[str, List[str]]: """输入特征列表，输出{name: [upstream_names]}字典""" lineage = {} for feat in features: lineage[feat.name] = feat.depends_on.copy() # 这里可扩展：若depends_on中包含其他FeatureContract名称，则递归解析 return lineage

实操中，我们要求所有特征模块必须放在features/目录下，PipelineCenter启动时自动扫描并实例化。当数据团队通知“event_time字段类型从string改为timestamp”，我们运行build_feature_lineage()，立即得到受影响特征列表：user_activity_24h,session_duration_1h等，精准推送告警给对应模型Owner。

3.3 ExperimentContract：把“试错”变成可审计的工程行为

算法工程师的日常是大量试错，但试错过程常被当作临时行为，缺乏记录。ExperimentContract将实验升格为一等公民：

import functools import time import logging from datetime import datetime from typing import Callable, Any, Dict def experiment( name: str, tags: Optional[Dict[str, str]] = None, timeout: int = 3600 ) -> Callable: def decorator(func: Callable) -> Callable: @functools.wraps(func) def wrapper(*args, **kwargs) -> Dict[str, Any]: run_id = f"{name}_{int(time.time())}_{hash(str(kwargs)) % 10000}" start_time = datetime.now() # 初始化实验上下文 logger = logging.getLogger(f"experiment.{run_id}") logger.info(f"Starting experiment {run_id} with config: {json.dumps(kwargs, indent=2)}") try: # 执行实验 result = func(*args, **kwargs, run_id=run_id, logger=logger) # 记录元数据 duration = (datetime.now() - start_time).total_seconds() metadata = { "run_id": run_id, "name": name, "start_time": start_time.isoformat(), "duration_sec": duration, "status": "success", "tags": tags or {}, "git_commit": get_git_commit(), # 辅助函数，获取当前HEAD } # 写入mlruns.db（SQLite） save_experiment_metadata(metadata) logger.info(f"Experiment {run_id} completed in {duration:.2f}s") return result except Exception as e: duration = (datetime.now() - start_time).total_seconds() error_metadata = { "run_id": run_id, "name": name, "start_time": start_time.isoformat(), "duration_sec": duration, "status": "failed", "error": str(e), "tags": tags or {}, } save_experiment_metadata(error_metadata) logger.error(f"Experiment {run_id} failed: {e}", exc_info=True) raise return wrapper return decorator # 使用示例 @experiment(name="train_credit_model", tags={"team": "risk", "priority": "high"}) def train_model( run_id: str, logger: logging.Logger, data_version: str = "v2024q1", learning_rate: float = 0.01 ) -> Dict[str, Any]: logger.info(f"Loading data version {data_version}") X, y = load_training_data(data_version) logger.info("Training model...") model = train(X, y, lr=learning_rate) # 保存模型权重（到OSS） model_path = f"s3://models/credit/{run_id}/model.pth" save_model(model, model_path) # 记录关键指标 metrics = evaluate(model, X, y) logger.info(f"Metrics: {metrics}") return { "model_path": model_path, "metrics": metrics, "run_id": run_id }

这个装饰器带来的改变是质的：每次实验都有唯一run_id，所有日志、指标、产物路径都绑定此ID。运维同学查问题时，只需拿到run_id，就能在ELK中搜到完整日志流，在S3中找到对应模型，在SQLite中看到当时用的数据版本和Git Commit。没有“哪个版本的代码跑的？”，只有“run_id= train_credit_model_1710523421_8765对应Git commita1b2c3d”。

4. 实操全流程：从本地开发到生产部署的7个关键环节

4.1 环境准备：用Poetry锁定“可重现”的最小依赖

MLOps最大的敌人是环境不一致。我坚持不用requirements.txt，而用Poetry管理依赖，因为它能精确锁定每个包的版本及传递依赖：

# pyproject.toml [tool.poetry] name = "mlops-research" version = "0.1.0" description = "" authors = ["Your Name <you@example.com>"] [tool.poetry.dependencies] python = "^3.9" pandas = "^1.5.3" numpy = "^1.23.5" scikit-learn = "^1.2.0" pydantic = "^1.10.12" sqlalchemy = "^1.4.46" [tool.poetry.group.dev.dependencies] pytest = "^7.2.0" black = "^23.1.0" jupyter = "^1.0.0" [build-system] requires = ["poetry-core"] build-backend = "poetry.core.masonry.api"

关键点在于poetry.lock文件：它记录了每个包的exact version、checksum、source URL。团队成员执行poetry install，安装的依赖与你在本地完全一致。我曾遇到一个诡异bug：本地训练AUC 0.85，CI环境0.72。排查发现scikit-learn从1.2.0升级到1.2.1，其RandomForestClassifier默认max_features参数从"sqrt"变为1.0，导致特征重要性计算逻辑变化。Poetry锁死版本后，此类问题归零。

4.2 本地开发：用Docker Compose模拟生产数据流

本地开发不能只跑单个脚本，必须模拟真实数据管道。我们用Docker Compose启动三个服务：

# docker-compose.yml version: '3.8' services: # 模拟上游数据源（MySQL） mysql: image: mysql:8.0 environment: MYSQL_ROOT_PASSWORD: root MYSQL_DATABASE: user_db volumes: - ./data/mysql:/var/lib/mysql ports: - "3306:3306" # 模拟特征计算服务（Flask API） feature-service: build: . environment: DB_URL: mysql+pymysql://root:root@mysql:3306/user_db depends_on: - mysql ports: - "5000:5000" # 模拟模型服务（FastAPI） model-service: build: . environment: FEATURE_API_URL: http://feature-service:5000 depends_on: - feature-service ports: - "8000:8000"

开发时，执行docker-compose up --build，即可获得一个包含MySQL、特征服务、模型服务的完整环境。特征服务暴露/features/compute端点，接收{"user_id": 123}，返回{"user_activity_24h": 5, "avg_order_value": 298.5}；模型服务调用此API，拼接特征后返回预测。所有交互走HTTP，与生产K8s环境完全一致，避免“本地能跑，线上挂掉”。

4.3 特征验证：用Pytest编写“特征单元测试”

特征逻辑必须像普通代码一样被测试。我们为每个FeatureContract编写对应测试：

# tests/test_user_activity.py import pytest import pandas as pd from features.user_activity import UserActivityScore @pytest.fixture def sample_raw_data(): return pd.DataFrame({ "user_id": [1, 1, 1, 2, 2], "event_time": pd.to_datetime([ "2024-03-15 10:00:00", "2024-03-15 11:00:00", "2024-03-14 09:00:00", # 超出24h窗口 "2024-03-15 12:00:00", "2024-03-15 13:00:00" ]), "event_type": ["click", "purchase", "view", "click", "purchase"] }) def test_user_activity_24h_computation(sample_raw_data): """测试24小时活跃度计算""" feat = UserActivityScore(window_hours=24) # 计算结果 result = feat.compute(sample_raw_data) # 验证：user_id=1有2个事件在24h内，user_id=2有2个 expected = pd.Series([2, 2], index=[1, 2]) pd.testing.assert_series_equal(result, expected) def test_feature_depends_on_declaration(): """验证依赖声明正确""" feat = UserActivityScore(window_hours=12) assert feat.depends_on == ["user_id", "event_time", "event_type"]

执行poetry run pytest tests/ -v，所有特征测试通过才允许合并代码。这比“人工检查SQL”可靠百倍。曾有个特征因groupby().size()未处理空组，导致新用户返回NaN，测试用例sample_raw_data中加入user_id=3无事件，立即捕获此缺陷。

4.4 模型测试：不只是AUC，更要“行为一致性”测试

模型测试不能只看离线指标，必须验证线上行为。我们设计三类测试：

1. 输入边界测试：传入全NaN、全0、超长字符串，验证是否优雅降级；
2. 版本一致性测试：新模型与旧模型对同一输入的输出差异在阈值内（如abs(new-old) < 0.01）；
3. 特征敏感性测试：固定其他特征，仅改变income，验证default_prob单调递增。

# tests/test_credit_model.py def test_model_input_validation(): """测试输入校验""" model = CreditRiskModel("models/credit_v2.0.pth") # 缺少必要字段 with pytest.raises(ValueError, match="Missing input features"): model.predict(pd.DataFrame({"age": [30]})) def test_model_backward_compatibility(): """测试新旧模型输出一致性""" old_model = CreditRiskModel("models/credit_v1.9.pth") new_model = CreditRiskModel("models/credit_v2.0.pth") X_test = pd.read_csv("data/test_sample.csv") # 固定测试集 old_pred = old_model.predict(X_test) new_pred = new_model.predict(X_test) # 允许微小差异（如浮点精度） assert np.allclose(old_pred, new_pred, atol=0.001) def test_income_sensitivity(): """测试收入增加，违约概率应下降""" model = CreditRiskModel("models/credit_v2.0.pth") base_X = pd.DataFrame({ "age": [35], "income": [50000], "loan_amount": [10000], "employment_years": [5] }) base_pred = model.predict(base_X)[0] # 收入翻倍 high_income_X = base_X.copy() high_income_X["income"] = 100000 high_pred = model.predict(high_income_X)[0] assert high_pred < base_pred # 违约概率应更低

这些测试集成到CI流水线，任一失败则阻断发布。它让“模型迭代”从艺术变成科学。

4.5 CI/CD流水线：用GitHub Actions实现全自动验证

我们用GitHub Actions构建端到端流水线，关键步骤如下：

# .github/workflows/mlops-ci.yml name: MLOps CI Pipeline on: push: branches: [main] paths: - "src/**" - "features/**" - "models/**" - "tests/**" - "pyproject.toml" - "poetry.lock" jobs: test-features: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - name: Set up Python uses: actions/setup-python@v4 with: python-version: '3.9' - name: Install Poetry uses: snok/install-poetry@v1 - name: Install dependencies run: poetry install - name: Run feature tests run: poetry run pytest tests/test_features.py -v test-models: runs-on: ubuntu-latest needs: test-features steps: - uses: actions/checkout@v3 - name: Set up Python uses: actions/setup-python@v4 with: python-version: '3.9' - name: Install Poetry uses: snok/install-poetry@v1 - name: Install dependencies run: poetry install - name: Run model tests run: poetry run pytest tests/test_models.py -v build-and-push: runs-on: ubuntu-latest needs: test-models if: github.event_name == 'push' && github.ref == 'refs/heads/main' steps: - uses: actions/checkout@v3 - name: Set up Docker Buildx uses: docker/setup-buildx-action@v2 - name: Login to Container Registry uses: docker/login-action@v2 with: registry: ghcr.io username: ${{ secrets.GITHUB_ACTOR }} password: ${{ secrets.GITHUB_TOKEN }} - name: Build and push uses: docker/build-push-action@v4 with: context: . push: true tags: ghcr.io/${{ github.repository_owner }}/mlops-research:${{ github.sha }}

流水线严格遵循“测试先行”：只有特征和模型测试全通过，才构建Docker镜像。镜像Tag使用Git SHA，确保ghcr.io/your-org/mlops-research:a1b2c3d与代码库a1b2c3d完全对应。生产K8s集群通过imagePullPolicy: Always拉取，杜绝“镜像缓存导致旧代码运行”。

4.6 生产部署：K8s Deployment + Service Mesh可观测性

生产环境采用K8s原生部署，不引入Istio等复杂Service Mesh，而是用轻量级方案：

# k8s/model-deployment.yaml apiVersion: apps/v1 kind: Deployment metadata: name: credit-model spec: replicas: 3 selector: matchLabels: app: credit-model template: metadata: labels: app: credit-model annotations: # 注入Git Commit和模型版本，供Prometheus抓取 prometheus.io/scrape: "true" prometheus.io/path: "/metrics" prometheus.io/port: "8000" model.version: "2.1.0" git.commit: "a1b2c3d4e5f6" spec: containers: - name: model image: ghcr.io/your-org/mlops-research:a1b2c3d4e5f6 env: - name: MODEL_PATH value: "s3://models/credit/a1b2c3d4e5f6/model.pth" - name: FEATURE_API_URL value: "http://feature-service.default.svc.cluster.local:5000" ports: - containerPort: 8000 livenessProbe: httpGet: path: /healthz port: 8000 initialDelaySeconds: 30 periodSeconds: 10 readinessProbe: httpGet: path: /readyz port: 8000 initialDelaySeconds: 5 periodSeconds: 5 --- apiVersion: v1 kind: Service metadata: name: credit-model spec: selector: app: credit-model ports: - port: 80 targetPort: 8000

关键创新点在于健康探针设计：

• /healthz：检查模型加载状态、OSS连接、特征服务连通性，失败则重启Pod；
• /readyz：额外检查特征缓存命中率、最近1分钟P95延迟，低于阈值则从Service摘除流量。

Prometheus配置抓取/metrics端点，自动提取model_version、git_commit标签，绘制各版本模型的QPS、延迟、错误率曲线。当v2.1.0版本错误率突增，可立即对比v2.0.0基线，确认是否为模型问题。

4.7 故障排查：用Trace ID串联全链路日志

线上问题定位最耗时的是“请求在哪一步丢了”。我们强制所有服务在收到请求时生成唯一trace_id，并在日志、HTTP Header、数据库记录中透传：

# 在FastAPI中间件中注入trace_id from fastapi import Request, Response import uuid @app.middleware("http") async def add_trace_id(request: Request, call_next): trace_id = request.headers.get("X-Trace-ID", str(uuid.uuid4())) request.state.trace_id = trace_id # 记录请求开始日志 logger.info(f"Request start: {request.method} {request.url.path} | trace_id={trace_id}") response = await call_next(request) # 记录响应日志 logger.info(f"Request end: {response.status_code} | trace_id={trace_id}") return response # 在调用特征服务时透传 async def fetch_features(user_id: int): async with httpx.AsyncClient() as client: response = await client.post( "http://feature-service:5000/features/compute", json={"user_id": user_id}, headers={"X-Trace-ID": request.state.trace_id} # 透传 ) return response.json()

当业务方报告“用户123预测失败”，运维在ELK中搜索trace_id: xxxxx，即可看到完整链路：model-service收到请求 → 调用feature-service→feature-service查询MySQL → 返回特征 →model-service计算预测 → 返回结果。耗时分布一目了然，无需跨多个系统查日志。

5. 常见问题与实战排障指南：那些文档里不会写的坑

5.1 问题速查表：高频故障与根因定位

5.2 实操心得：来自7个项目的血泪教训

教训1：永远不要相信“数据已清洗”
在工业质检项目中，数据团队承诺“所有图像已去噪、尺寸归一化”，但我们在线上发现大量模糊图像导致模型误判。解决方案：在FeatureContract中加入数据质量检查钩子：

class ImageFeature(FeatureContract): def compute(self, raw_data: pd.DataFrame) -> pd.Series: # 质量检查 for img_path in raw_data["image_path"]: img = cv2.imread(img_path) if img is None: raise ValueError(f"Invalid image: {img_path}") if cv2.Laplacian(img, cv2.CV_64F).var() < 50: # 模糊度阈值 logger.warning(f"Blurry image detected: {img_path}") # 正常计算...

教训2：模型版本号必须语义化，禁止时间戳
曾用v20240315作版本号，结果因时区问题，不同地区构建的镜像版本号不一致。改用SemVer：v2.1.0表示向后兼容的功能更新，v2.1.1表示bug修复。PipelineCenter在get_metadata()中强制校验版本格式，非法版本拒绝注册。

教训3：特征缓存失效策略比缓存本身更重要
在金融项目中，user_balance特征缓存1小时，但用户还款后余额立即变化，导致模型用旧余额做决策。最终方案：对强时效性特征（如余额、实时点击），禁用缓存，直接查DB；对弱时效性特征（如用户年龄、职业），用LRU缓存+TTL。FeatureContract新增cache_policy属性，明确声明策略。

教训4：监控指标必须包含“非技术维度”
除了CPU、内存、QPS，我们增加了model_staleness_days（模型距最新训练数据的天数）、feature_drift_score（KS检验值）、prediction_distribution_skew（预测分数分布偏移）。当model_staleness_days > 30，自动触发告警，提醒算法同学重新训练。

5.3 避坑清单：新手最容易栽的5个深坑

1. 不要在模型代码中硬编码数据路径
   错误：df = pd.read_csv("/data/train.csv")
   正确：通过config参数传入，或从环境变量读取，确保本地、CI、生产路径可配置。

2. 不要用pickle序列化模型
   pickle不跨Python版本，且存在安全风险。改用joblib（sklearn）或torch.save（PyTorch），并验证反序列化兼容性。

3. 不要忽略特征计算的幂等性
   FeatureContract.compute()必须是纯函数：相同输入必得相同输出。避免在计算中调用time.time()、random.random()等非确定性操作。

4. 不要把Git当作文件存储
   权重文件、大型数据集严禁git add。用Git L