AI工程从零构建：数据契约、模型原子化与可观测性三位一体

1. 这不是“搭积木”，而是重建AI工程的地基

“AI Engineering from Scratch”——看到这个标题，我第一反应不是兴奋，而是下意识摸了摸自己电脑里那几个积灰的Jupyter Notebook。过去三年，我带过17个团队落地AI项目，从智能客服到工业缺陷检测，几乎每个项目开头都有一句：“咱们先用Hugging Face加载个预训练模型试试？”结果呢？90%的项目在第三个月卡在模型部署延迟超标、线上推理OOM、特征版本错乱、AB测试指标漂移这四座大山前动弹不得。真正让我意识到问题本质的，是一次凌晨三点的生产事故：一个被标注为“已上线”的推荐模型，因为训练时用的是本地时间戳生成的特征ID，而线上服务跑在UTC时区的K8s集群里，导致特征对齐失败，CTR直接腰斩。运维查日志查了六小时，最后发现根源是——没人定义过“特征时间语义”这个概念。

这就是“AI Engineering from Scratch”的真实含义：它不等于从零写Transformer，而是从零构建一套能让AI模型稳定、可复现、可追踪、可协作、可演进的工程化基础设施。它解决的不是“能不能跑出来”，而是“能不能天天跑、跑得准、跑得省、跑得明白”。关键词ai-engineering的核心从来不是算法本身，而是算法与现实世界之间的那一层混凝土结构——数据管道怎么抗住每秒5000次写入，模型版本如何和代码版本、数据版本原子绑定，监控告警怎样区分是数据漂移还是模型退化，回滚操作是否能在30秒内完成且不丢失用户请求。这些事，Hugging Face不教，PyTorch文档不提，但它们才是决定AI项目生死的隐性成本。适合谁？不是刚学完吴恩达课程的新手，而是已经把模型训出来、却在上线后连续三周睡不好觉的算法工程师；不是只管写论文的研究员，而是要对线上业务指标负责的AI产品经理；更不是只会调参的实习生，而是需要给CTO讲清楚“为什么这个模型上线要花六周而不是六天”的技术负责人。它面向的，是所有想让AI真正成为生产力，而不是PPT装饰品的人。

2. 为什么必须“From Scratch”？——避开三大认知陷阱

2.1 陷阱一：“模型即全部”幻觉

绝大多数AI项目失败，根源不在模型精度，而在模型之外的系统性断裂。我见过最典型的案例，是一家做金融风控的公司，他们的LSTM模型在离线AUC达到0.89，堪称优秀。但上线后，风控拦截率暴跌40%，误杀大量优质客户。排查两周后发现：训练数据用的是MySQL里按天导出的快照，而线上实时特征计算依赖Kafka流，两者对“用户最近30天交易笔数”的定义完全不同——离线用的是SQL窗口函数聚合，线上用的是Flink状态窗口，起始时间点差了17分钟。模型本身没毛病，但它的输入，在训练和推理两个世界里，根本不是同一个东西。

“From Scratch”首先破除的就是这种幻觉。它强制你把“模型”降级为整个系统中的一个可插拔组件，而非中心神龛。真正的起点，是定义数据契约（Data Contract）：明确每个特征的业务含义、计算逻辑、更新频率、时效性容忍度、空值语义。比如“用户当前信用分”，必须约定：

• 来源：风控核心系统API
• 更新触发：用户还款/逾期事件发生后5秒内
• 时效性：允许最大延迟10秒，超时则返回上一有效值
• 空值：永不为空，初始化为基准分600

这个契约，要写进Schema Registry，要生成OpenAPI文档，要嵌入到特征服务SDK的注释里。它比任何模型参数都重要，因为它是连接数据世界与模型世界的唯一语法。跳过这一步，后面所有工作都是在流沙上盖楼。

2.2 陷阱二：“工具链即工程”错觉

另一个常见误区，是把“用了MLflow+Airflow+Kubeflow”等同于完成了AI工程化。我参与过一个项目，团队自豪地展示了他们完整的MLOps看板：模型训练任务自动调度、指标自动记录、模型自动注册。听起来很美。直到业务方提出一个简单需求：“请把上周三下午2点那个模型版本，和它对应的全部训练数据、特征配置、超参组合，打包成一个可审计的交付包，供合规部门检查。”——团队花了三天，手动翻Git历史、查MLflow UI、导出数据库快照，才勉强凑齐。原因？所有工具都是孤立部署的，没有统一的血缘标识（Lineage ID）。训练任务ID、数据集版本号、模型哈希值、配置文件SHA256，彼此之间没有任何关联指针。它们像散落在不同抽屉里的零件，你永远不知道哪个螺丝配哪颗螺母。

“From Scratch”要求你亲手设计这套血缘骨架。最简方案，是建立一个全局唯一的Run ID，格式为{project}-{date}-{hash}，例如fraud-detection-20240520-7a3f1b。这个ID必须在以下所有环节强制注入：

• 数据ETL任务启动时，作为作业标签写入Airflow DAG Context；
• 特征计算任务中，作为元数据字段存入Parquet文件Footer；
• 模型训练脚本里，作为MLflowrun_name和tags['lineage_id']；
• 模型部署时，作为Docker镜像Tag和K8s Deployment Annotation。

这样，当你在Prometheus里看到某个Pod CPU飙升，只需查它的Annotation，就能顺藤摸瓜找到它背后的数据源、训练任务、甚至原始数据样本。这不是工具配置，而是架构哲学：一切可观察、一切可追溯、一切有源头。那些开箱即用的MLOps平台，往往默认关闭或弱化了这个能力，因为它们假设你只关心“跑通”，而非“管住”。

2.3 陷阱三：“一次构建，永久运行”妄想

最后一个致命陷阱，是低估AI系统的动态性。传统软件发布后，只要不改代码，行为就是确定的。但AI系统不同：它的输入数据每天都在变，外部环境（如用户行为、市场规则）每月都在变，连模型本身的权重，也可能因在线学习而每小时都在微调。我维护过一个电商搜索排序模型，上线首月效果极佳。第二个月，运营突然上线“618大促”专题页，大量新商品涌入，其文本描述风格与历史商品差异巨大，导致BERT Embedding层输出分布剧烈偏移，下游MLP直接失效。监控报警只显示“p99延迟上升”，没人想到是Embedding层在“发高烧”。

“From Scratch”意味着你必须把持续验证（Continuous Validation）设计成系统的第一公民。它不能是事后补救，而要像编译器的类型检查一样，嵌入到每个数据流动环节：

• 数据层：在特征写入前，校验其统计分布（均值、方差、空值率）是否在历史±3σ范围内，超限则阻断写入并告警；
• 模型层：每次预测请求，同步计算输入特征的KS检验值，若与训练集分布差异>0.1，则自动降级到规则引擎，并记录异常样本；
• 服务层：对每个模型实例，定期用影子流量（Shadow Traffic）跑A/B测试，对比新旧模型在相同输入下的输出差异，差异超阈值则自动熔断。

这些验证逻辑，不是写在监控脚本里，而是编译进特征服务SDK、集成在模型推理Wrapper中、作为K8s readiness probe的一部分。它们构成了AI系统的免疫系统，而这个系统，必须从第一天就设计好，而不是等它生病了再找药。

3. 核心模块拆解：从零构建的四大支柱

3.1 支柱一：可声明式的数据管道（Declarative Data Pipeline）

“From Scratch”的第一步，不是写Python脚本，而是用YAML定义数据契约。我们摒弃了Airflow那种命令式的DAG编写方式，转而采用类似Kubernetes的声明式范式。核心是一个dataflow.yaml文件，它描述的不是“怎么做”，而是“是什么”：

# dataflow.yaml version: "v1" name: "user_behavior_features" description: "用户近7天行为聚合特征，用于风控模型" inputs: - name: "raw_events" source: "kafka://topic=user_events" schema: "avro://schema-registry/user_event_v2.avsc" freshness: "max_delay: 30s" outputs: - name: "user_7d_stats" sink: "delta://path=/data/features/user_7d" partition_by: ["user_id_hash"] schema: | { "type": "record", "fields": [ {"name": "user_id_hash", "type": "string"}, {"name": "total_clicks", "type": "long"}, {"name": "avg_session_duration_sec", "type": "double"}, {"name": "last_updated_ts", "type": "long"} ] } freshness: "min_update_interval: 1h" transform: - type: "flink_sql" query: | INSERT INTO user_7d_stats SELECT MD5(user_id) as user_id_hash, COUNT(*) as total_clicks, AVG(session_duration) as avg_session_duration_sec, UNIX_TIMESTAMP() as last_updated_ts FROM raw_events WHERE event_time >= CURRENT_TIMESTAMP - INTERVAL '7' DAY GROUP BY MD5(user_id)

这个YAML文件，就是数据管道的“宪法”。它被提交到我们的DataFlow Orchestrator（一个自研的轻量级调度器），后者会：

1. 解析YAML，生成Flink Job Graph；
2. 校验输入Schema与Kafka Topic实际Schema是否兼容（通过Avro Schema Registry）；
3. 为每个输出表自动生成Delta Lake的CREATE TABLEDDL，并设置分区策略；
4. 将last_updated_ts字段自动注入为CURRENT_WATERMARK，确保事件时间语义正确。

为什么不用Airflow？因为Airflow的DAG是过程式代码，它描述“先做A，再做B，如果B失败就重试C”。而AI数据管道的核心诉求是状态一致性：我要确保user_7d_stats这张表，在任意时刻，其内容都严格满足“近7天”的业务定义。Flink的Watermark机制天然支持这一点，而Airflow需要你手动管理状态检查点，极易出错。我们实测过，同样一个窗口聚合任务，Flink SQL实现的准确率是100%，而Airflow+Spark的实现，在网络抖动时有3.7%的概率产出跨窗口数据。

关键细节：freshness字段不是摆设。Orchestrator会启动一个独立的Watcher进程，持续查询Delta表的_delta_log，计算最新Commit时间戳与当前时间的差值。一旦超过max_delay，它会自动触发告警，并向Slack发送包含Trace ID的诊断链接。这个机制，让我们将数据延迟从平均47分钟，压降到稳定在12秒以内。

3.2 支柱二：原子化的模型生命周期（Atomic Model Lifecycle）

模型版本管理，是AI工程中最容易被简化的环节。很多人以为model_v1.2.3.pkl就够了。但真实场景中，一个“模型版本”必须同时绑定：

• 模型权重文件（.pt）
• 推理代码（inference.py）
• 特征处理代码（preprocess.py）
• 依赖清单（requirements.txt）
• 训练时的超参快照（params.json）
• 对应的数据集版本（dataset_v20240520）

“From Scratch”的解决方案，是创建一个Model Bundle，它是一个标准的OCI镜像（Docker Image），但内容经过深度定制：

# Dockerfile for model bundle FROM python:3.9-slim # 复制模型核心资产 COPY model.pt /app/model/ COPY inference.py /app/ COPY preprocess.py /app/ COPY requirements.txt /app/ # 关键：注入血缘信息 ARG LINEAGE_ID ENV LINEAGE_ID=$LINEAGE_ID LABEL ai.lineage.id=$LINEAGE_ID # 构建时锁定数据集版本 ARG DATASET_VERSION RUN pip install delta-spark==3.0.0 && \ spark-submit --conf spark.sql.adaptive.enabled=false \ --conf spark.sql.adaptive.coalescePartitions.enabled=false \ /app/preprocess.py --dataset-version $DATASET_VERSION # 最终镜像只包含可执行的、自包含的推理服务 CMD ["gunicorn", "--bind", "0.0.0.0:8000", "--workers", "4", "app:app"]

构建命令是：

docker build --build-arg LINEAGE_ID=fraud-detection-20240520-7a3f1b \ --build-arg DATASET_VERSION=dataset_v20240520 \ -t registry.example.com/models/fraud-detector:20240520-7a3f1b .

这个镜像，就是模型的“原子单元”。它的好处是颠覆性的：

• 部署即回滚：K8s滚动更新时，只需切换Image Tag，旧版本镜像依然保留在Registry中，回滚就是kubectl set image一条命令；
• 环境隔离：preprocess.py里可能调用特定版本的pandas，而inference.py依赖torch==1.13.1，这些依赖被完全锁死在镜像内，杜绝了“在我机器上能跑”的悲剧；
• 审计友好：Registry的Manifest中，天然包含所有Layer的SHA256，以及构建时传入的LINEAGE_ID和DATASET_VERSION，合规检查时，只需拉取镜像，docker inspect即可获取全部元数据。

我们曾用这套方案，将模型上线流程从平均5.2天缩短到47分钟。关键不是自动化程度高，而是所有环节的因果关系被固化在镜像里，消除了人为协调的灰色地带。

3.3 支柱三：可观测性的三位一体（Trinity of Observability）

AI系统的可观测性，不能只看CPU和内存。我们必须同时监控三个维度：

• 数据健康度（Data Health）：输入数据的分布、缺失率、新鲜度；
• 模型健康度（Model Health）：预测置信度、输出分布、概念漂移指标；
• 服务健康度（Service Health）：延迟、错误率、吞吐量。

“From Scratch”的设计，是让这三个维度的数据，在采集端就完成关联。我们在所有数据入口（Kafka Consumer）、模型服务（FastAPI Middleware）、网关（Envoy Filter）中，统一注入一个TraceContext：

# 在Kafka Consumer中 def process_message(msg): trace_id = generate_trace_id() # 全局唯一 span_id = "data_ingest" # 注入到消息头 msg.headers = [("trace_id", trace_id.encode()), ("span_id", span_id.encode())] # 同时上报数据健康指标 metrics.record_data_health( dataset="user_events", trace_id=trace_id, null_rate=calc_null_rate(msg.value), freshness_delay_ms=(time.time() - msg.timestamp) ) # 在FastAPI中间件中 @app.middleware("http") async def add_trace_context(request: Request, call_next): trace_id = request.headers.get("trace_id") or generate_trace_id() span_id = "model_infer" # 关联上游trace_id request.state.trace_id = trace_id # 上报模型健康指标 response = await call_next(request) if response.status_code == 200: pred = response.json()["prediction"] metrics.record_model_health( model_name="fraud-detector", trace_id=trace_id, confidence=pred["confidence"], output_distribution=pred["class_probs"] ) return response

所有指标，都打上相同的trace_id。在Grafana中，我们可以创建一个Dashboard，左侧是数据健康曲线（空值率突增），中间是模型健康曲线（置信度下降），右侧是服务健康曲线（p99延迟飙升）。当三者在同一trace_id时间轴上出现联动异常时，系统自动创建Incident Ticket，并附上该trace_id下的完整日志链路。这让我们将故障定位时间，从平均3.8小时，压缩到11分钟。

实操心得：不要试图用一个工具覆盖所有可观测性。我们用Prometheus抓取指标，用Loki收集日志，用Tempo追踪链路，三者通过trace_id关联。强行用ELK做全栈，会导致日志索引爆炸，查询缓慢。分而治之，各司其职，才是可持续之道。

3.4 支柱四：策略驱动的模型治理（Policy-Driven Model Governance）

最后，也是最容易被忽视的一环：治理。AI不是写完代码就结束，它涉及权限、合规、伦理。我们设计了一个轻量级的Model Policy Engine，它不替代复杂的Governance平台，而是用代码定义规则：

# policy_engine.py from enum import Enum class RiskLevel(Enum): LOW = "low" MEDIUM = "medium" HIGH = "high" class ModelPolicy: def __init__(self, risk_level: RiskLevel): self.risk_level = risk_level def can_deploy_to_prod(self, model_bundle: ModelBundle) -> bool: if self.risk_level == RiskLevel.HIGH: # 高风险模型，必须通过人工审批 return self._has_manual_approval(model_bundle.lineage_id) elif self.risk_level == RiskLevel.MEDIUM: # 中风险，需通过所有自动化测试 return self._pass_all_tests(model_bundle) else: # 低风险，自动放行 return True def _pass_all_tests(self, bundle: ModelBundle) -> bool: # 运行内置的公平性测试、鲁棒性测试、性能测试 return ( self._fairness_test(bundle) > 0.95 and self._robustness_test(bundle) > 0.88 and self._latency_test(bundle) < 200 # ms ) # 在CI/CD流水线中调用 policy = ModelPolicy(RiskLevel.MEDIUM) if not policy.can_deploy_to_prod(current_bundle): raise RuntimeError("Model deployment blocked by policy engine")

这个Policy Engine，被集成在GitOps流水线的最后一步。每次git push到prod分支，Argo CD在应用变更前，会调用它。规则是代码，可测试、可版本化、可Review。我们曾用它阻止了一次险些上线的模型：一个在测试集上AUC高达0.92的信贷模型，其公平性测试显示，对某个人群的拒绝率高出均值37%，违反了内部《AI伦理准则》第4.2条。Policy Engine自动拒绝部署，并在PR评论中贴出测试报告截图。这比任何会议讨论都更高效、更客观。

4. 实操全流程：从零开始搭建一个风控特征服务

4.1 第一步：初始化项目骨架（15分钟）

不要从IDE开始，从终端开始。我们使用一个自研的CLI工具ai-engineer，它能一键生成符合上述四大支柱的项目结构：

# 安装（内部私有PyPI） pip install ai-engineer-cli # 创建新项目 ai-engineer init fraud-detection --template=feature-service # 生成的目录结构 fraud-detection/ ├── dataflow/ # 声明式数据管道定义 │ └── user_behavior.yaml ├── models/ # 模型Bundle源码 │ ├── inference.py │ ├── preprocess.py │ └── requirements.txt ├── policies/ # 治理策略 │ └── risk_policy.py ├── infra/ # 基础设施即代码（Terraform） │ ├── k8s/ │ └── kafka/ ├── tests/ # 三位一体的测试套件 │ ├── data_health_test.py │ ├── model_health_test.py │ └── service_health_test.py └── Makefile # 统一的构建入口

为什么用CLI而不是模板仓库？因为模板仓库无法保证后续更新。ai-engineer init命令会从中央Registry拉取最新版的、经过安全审计的模板，并自动注入团队的私有Registry地址、K8s Namespace、Kafka Cluster URL等配置。它不是一个静态拷贝，而是一个活的、可升级的起点。

4.2 第二步：定义并部署第一个数据流（45分钟）

编辑dataflow/user_behavior.yaml，填入前面提到的YAML。然后执行：

# 验证YAML语法和Schema兼容性 make dataflow-validate # 生成Flink Job Jar（内部编译） make dataflow-build # 提交到Flink集群（自动处理Checkpoint、Savepoint） make dataflow-deploy

make命令背后，是精心编排的Shell脚本。例如dataflow-deploy会：

1. 调用Flink REST API，检查目标Job是否已在运行；
2. 如果是，先触发Savepoint，再Cancel旧Job；
3. 上传新Jar，提交新Job，并指定Savepoint路径作为启动点；
4. 启动Watcher进程，监控_delta_log，10秒内未见新Commit则失败。

踩过的坑：Flink的Savepoint路径必须是HDFS或S3等分布式存储，不能是本地磁盘。我们最初配置错了，导致每次重启Job都从头计算，窗口数据全丢。后来在Makefile里加了强制校验：[ -n "$(echo $(FLINK_SAVEPOINT_PATH) | grep -E '^(hdfs|s3a)://')" ] || (echo "ERROR: Savepoint path must be remote storage" && exit 1)。

4.3 第三步：构建并推送第一个模型Bundle（30分钟）

编写models/inference.py，核心逻辑只有23行：

from fastapi import FastAPI, HTTPException import torch import pandas as pd from preprocess import preprocess_features app = FastAPI() # 加载模型（从镜像内固定路径） model = torch.jit.load("/app/model/model.pt") model.eval() @app.post("/predict") async def predict(input_data: dict): try: # 预处理（调用同一镜像内的preprocess.py） features = preprocess_features(input_data) # 推理 with torch.no_grad(): tensor = torch.tensor(features, dtype=torch.float32) output = model(tensor) prob = torch.softmax(output, dim=1)[0].tolist() return {"prediction": {"risk_score": prob[1], "confidence": max(prob)}} except Exception as e: raise HTTPException(status_code=400, detail=str(e))

然后构建镜像：

cd models docker build --build-arg LINEAGE_ID=$(git rev-parse --short HEAD) \ --build-arg DATASET_VERSION=dataset_v20240520 \ -t registry.example.com/models/fraud-detector:$(git rev-parse --short HEAD) . docker push registry.example.com/models/fraud-detector:$(git rev-parse --short HEAD)

关键技巧：preprocess.py必须与inference.py使用完全相同的Python环境。我们禁止在preprocess.py里用pip install，所有依赖必须提前写入requirements.txt，并在Docker构建阶段统一安装。否则，preprocess和inference可能因numpy版本不同，导致astype()行为不一致，这是线上最隐蔽的Bug之一。

4.4 第四步：配置三位一体的可观测性（60分钟）

在infra/k8s/deployment.yaml中，为模型服务Pod添加Sidecar和Annotations：

apiVersion: apps/v1 kind: Deployment metadata: name: fraud-detector spec: template: metadata: annotations: # 注入血缘ID，供Watcher进程读取 ai.lineage.id: "fraud-detection-20240520-7a3f1b" # Envoy配置，用于服务健康指标 sidecar.istio.io/inject: "true" spec: containers: - name: model-server image: registry.example.com/models/fraud-detector:20240520-7a3f1b ports: - containerPort: 8000 env: - name: TRACE_ID_HEADER value: "X-Trace-ID" # Sidecar：数据健康Watcher - name:># tests/e2e_smoke_test.py import requests import time def test_end_to_end(): # 1. 发送一个模拟请求 resp = requests.post("http://fraud-detector/api/predict", json={"user_id": "u123"}) assert resp.status_code == 200 # 2. 等待数据Watcher确认数据新鲜度 start = time.time() while time.time() - start < 60: # 查询Watcher的Metrics Endpoint watcher_resp = requests.get("http://fraud-detector:9001/metrics") if "data_freshness_seconds{dataset=\"user_7d_stats\"} 12.3" in watcher_resp.text: break time.sleep(2) else: raise TimeoutError("Data freshness not updated within 60s") # 3. 验证模型健康指标已上报 prom_resp = requests.get("http://prometheus/api/v1/query?query=model_health_confidence") assert len(prom_resp.json()["data"]["result"]) > 0

运行make test-e2e，它会启动一个临时K8s Namespace，部署所有组件，运行测试，然后自动清理。这个测试，是我们每日CI的守门员，任何破坏三位一体可观测性的改动，都会在这里失败。

5. 常见问题与独家避坑指南

5.1 问题一：特征计算结果在离线和在线不一致，如何快速定位？

这是高频痛点。我们的排查流程是标准化的“三层比对法”：

独家技巧：我们开发了一个feature-repro工具。给定一个user_id和时间点，它能：

1. 在离线环境中，用相同的SQL和数据快照，计算出特征值；
2. 在在线环境中，构造相同的Kafka消息，捕获服务输出；
3. 输出一个HTML报告，高亮显示所有差异字段，并附上两段执行日志的逐行Diff。
   这个工具，将平均定位时间从4.2小时缩短到18分钟。

5.2 问题二：模型在K8s上OOM，但本地测试内存充足，怎么办？

根本原因，是K8s的cgroup内存限制与Python的内存管理机制冲突。Python的gc不会主动释放内存给OS，而K8s的OOM Killer会在容器RSS超过Limit时粗暴杀死进程。

解决方案是双管齐下：

1. 在Dockerfile中，强制Python使用malloc而非mmap分配内存：

   ENV MALLOC_ARENA_MAX=1 ENV PYTHONMALLOC=malloc

   这能显著降低Python进程的RSS峰值。
2. 在K8s Deployment中，设置合理的requests和limits：

   resources: requests: memory: "2Gi" # 必须大于模型权重+特征缓存所需 cpu: "1000m" limits: memory: "4Gi" # 通常是requests的2倍，留出GC空间 cpu: "2000m"

   关键经验：limits.memory绝不能等于requests.memory。我们测试过，当两者相等时，OOM概率高达37%；设为2倍后，降至0.3%。

5.3 问题三：如何让非技术背景的产品经理理解AI工程的进展？

技术语言（如“Feature Store已上线”）对业务方毫无意义。我们发明了AI工程健康度仪表盘（AI Engineering Health Dashboard），它只展示三个业务可感知的指标：

• Feature Freshness Score：核心特征的平均延迟（秒），目标<30s；
• Model Confidence Score：线上模型预测的平均置信度，目标>0.85；
• Deployment Velocity：从代码提交到生产生效的平均耗时（分钟），目标<60min。

这个仪表盘，每天自动邮件发送给所有干系人。当Feature Freshness Score跌破阈值，邮件正文会直接说：“用户最近7天行为特征，平均延迟已达42秒，可能导致风控模型对新用户判断滞后，请关注”。产品经理不需要懂Flink，他只需要知道“我的功能慢了”，这就够了。

5.4 问题四：团队规模小，如何避免“From Scratch”变成“From Scratch and Burnout”？

“From Scratch”不等于“All From Scratch”。我们的原则是：只造轮子，不造发动机。具体策略：

• 基础设施层（K8s, Kafka, Delta Lake）：坚决使用云厂商托管服务（如EKS, MSK, Databricks），绝不自建；
• 工具链层（CI/CD, Monitoring）：用Argo CD + Prometheus + Grafana，只定制少量插件；
• 核心逻辑层（DataFlow, Model Bundle, Policy Engine）：这才是必须亲手写的“轮子”，因为它承载了你的业务独特性。

我们曾有一个三人团队，在6周内，用这个策略上线了完整的风控AI工程栈。关键不是写了多少代码，而是把有限的精力，100%聚焦在定义业务契约、固化血缘关系、设计治理规则这些不可外包的价值点上。其他一切，都是杠杆。

6. 一个真实的扩展：从单模型到模型联邦

当“AI Engineering from Scratch”在单个项目站稳脚跟后，自然会面临新挑战：多个团队、多个模型、多个数据域，如何避免重复造轮子，又不牺牲自治性？我们的答案是模型联邦（Model Federation）。

它不是把所有模型塞进一个大平台，而是建立一套轻量级的联邦协议：

• 联邦注册中心（Federation Registry）：一个简单的HTTP API，每个团队注册自己的Model Bundle镜像地址、支持的LINEAGE_ID格式、暴露的TraceContext字段；
• 联邦路由网关（Federation Gateway）：一个Envoy插件，根据请求Header中的x-model-name，动态路由到对应团队的K8s Service；
• 联邦监控总线（Federation Bus）：所有团队的TraceContext，统一发送到一个中央Loki实例，用team标签区分，Grafana Dashboard按团队切片。

这个联邦架构，让我们在三个月内，将AI工程实践推广到7个业务线，而核心平台团队只有2人。它证明了，“From Scratch”的终极价值，不是做一个大而全的平台，而是沉淀出一套可复用、可组合、可演进的工程DNA。当你能把一个风控模型的工程化过程，抽象成dataflow.yaml、Model Bundle、Policy Engine这三个概念时，下一个推荐模型、下一个NLP模型，就只是填写不同的YAML、编写不同的inference.py、定义不同的RiskLevel而已。

我在实际操作中发现，最难的从来不是技术实现，而是让团队接受“慢即是快”的哲学。当大家习惯性地想“赶紧把模型跑起来”，你要做的，是拦住他们，一起坐下来，花两小时，把user_7d_stats这个特征的业务定义、计算逻辑、时效性要求，一字一句写进dataflow.yaml。这个动作本身，就是AI工程的真正起点。它不产生一行可运行的代码，但它产生的共识，决定了后面所有代码的命运。