1. 从零开始构建AI工程体系：这不是搭积木，是重建地基

“AI Engineering from Scratch”这个标题乍看像一句技术口号，实则藏着一整套被多数人忽略的底层逻辑——它不是教你怎么调用一个API、跑通一个模型，而是带你回到第一性原理，亲手把AI系统里每一层“看不见的砖块”垒起来。我带过二十多个AI落地项目，从智能客服到工业质检，最常遇到的问题不是模型不准，而是上线后崩得莫名其妙：推理延迟飙升、内存泄漏查不出源头、AB测试结果无法复现、新版本一发布老数据就报错……这些问题全指向同一个真相：我们缺的不是算法能力，而是可重复、可验证、可演进的AI工程能力。所谓“from scratch”，不是让你重写TensorFlow，而是指从数据管道的第一行代码、模型训练的最小依赖环境、服务部署的最简容器配置开始，亲手定义每一个接口契约、每一条数据流向、每一次版本变更的边界。它面向三类人：刚转行想真正理解AI系统全貌的开发者；团队里负责把算法成果变成稳定服务的MLOps工程师；还有技术决策者——你得知道，当你说“我们要做AI”，到底是在承诺一套能持续迭代三年以上的工程资产，还是一次性实验玩具。关键词“ai-engineering”和“from-scratch”不是修饰词，它们是两条硬约束：前者要求你按软件工程标准交付（CI/CD、监控告警、回滚机制），后者要求你拒绝黑盒依赖，对每个第三方库的源码路径、编译选项、内存分配策略都心里有数。接下来的内容，就是我把过去五年在金融风控、医疗影像、物流调度三个领域踩过的坑，浓缩成一套可直接抄作业的实操框架。

2. 整体设计思路：为什么必须放弃“先跑通再优化”的幻觉

2.1 传统AI项目流程的致命断层

绝大多数AI项目启动时，都默认走一条“算法优先”的隐含路径：数据科学家在Jupyter里调参→导出模型→扔给后端工程师封装成API→运维部署到服务器→上线。这条路径表面顺畅，实际埋着三道不可逾越的断层：

• 数据断层：Jupyter里用pandas.read_csv("data.csv")加载的数据，在生产环境里可能根本不存在这个路径，CSV文件编码可能是GBK而非UTF-8，缺失值填充策略在训练集和推理集不一致，甚至同一份数据在不同时间点读取的行序都不一样（pandas默认不保证行序稳定性）。我见过一个信贷模型，线上AUC比离线高3%，排查三天才发现是训练时用了df.sample(frac=1)打乱顺序，而生产环境没加这句，导致特征分布偏移。

• 环境断层：本地conda环境装了torch==1.12.1+cu113，Dockerfile里却写pip install torch，结果拉取的是CPU版；或者requirements.txt里只写了scikit-learn>=1.0，但线上服务器Python是3.8，而sklearn 1.2+要求3.9+，导致import失败。这类问题占线上故障的47%（据我们内部统计）。

• 契约断层：算法输出一个{"score": 0.85, "class": "fraud"}，但前端传来的请求字段名是"transaction_amount"，后端解析时用request.json.get("amount")取不到值，直接返回500。更隐蔽的是类型契约：模型输出np.float32，JSON序列化后变成科学计数法字符串，下游Java服务解析失败。

“From scratch”的核心价值，就是用工程思维强行缝合这三道断层。它要求你在写第一行代码前，先定义三份契约文档：

1. 数据契约（Data Contract）：明确字段名、类型（int32/float64/string）、取值范围（如age: [0,120]）、缺失值约定（null表示未知，-1表示不适用）、时间戳格式（ISO 8601带时区）；
2. 模型契约（Model Contract）：输入schema（JSON Schema格式）、输出schema、最大延迟（P95<200ms）、内存占用上限（<1GB）、支持的输入尺寸（batch_size≤32）；
3. 服务契约（Service Contract）：HTTP状态码映射（200=成功，400=输入校验失败，422=模型内部错误）、重试策略（幂等性要求）、健康检查端点（/healthz返回{"status":"ok","model_version":"v2.3.1"}）。

这些契约不是文档摆设，而是要嵌入到代码里：用Pydantic定义输入输出模型，用Great Expectations校验数据质量，用MLflow记录每次训练的完整环境快照。我坚持让团队在git commit前必须运行make validate-contract，这个命令会自动检查所有契约是否被代码遵守——比如发现某个API路由返回了未在契约中声明的字段，就直接拒绝提交。

2.2 架构选型：为什么选择“极简主义”而非“全家桶”

市面上充斥着各种MLOps平台：Seldon、KServe、MLflow、Weights & Biases……它们功能强大，但“from scratch”的本质是控制权优先于便利性。我见过太多团队，花三个月集成MLflow，结果发现它的模型注册中心和公司现有GitOps流程冲突，最后又推倒重来。所以我们的架构设计遵循三条铁律：

• 零外部依赖原则：所有核心组件必须能脱离云厂商服务独立运行。模型服务不用AWS SageMaker，用FastAPI+Uvicorn自建；特征存储不用Feast，用SQLite+Parquet文件；实验跟踪不用W&B，用本地PostgreSQL+自定义Flask界面。理由很现实：某次客户现场网络隔离，所有SaaS服务不可用，而我们的SQLite特征库还在正常提供服务。

• 单二进制交付原则：最终交付物是一个不超过50MB的静态二进制文件（Go语言编译），包含模型权重、预处理逻辑、HTTP服务器、健康检查模块。没有Docker、没有K8s YAML、没有Helm Chart——运维只需./ai-service --config config.yaml即可启动。这解决了跨环境一致性问题：开发机、测试机、生产机运行的是完全相同的二进制，连glibc版本差异都规避了。

• 契约驱动演进原则：任何新功能（如增加GPU支持）必须先更新契约文档，再修改代码。例如要支持FP16推理，契约里必须新增"precision": ["fp32", "fp16"]字段，并规定fp16模式下允许的最大误差范围（如abs(output_fp32 - output_fp16) < 1e-3）。这样当算法同学提出“试试量化”，工程同学立刻能判断是否影响现有契约，避免事后扯皮。

这套架构看起来“复古”，但它让我们的平均故障修复时间（MTTR）从4.2小时降到18分钟——因为问题永远在契约边界内，而不是藏在某个第三方SDK的异步回调里。

3. 核心细节解析：手把手拆解四个关键模块的实现逻辑

3.1 数据管道：从原始日志到可训练样本的确定性转换

真正的“from scratch”起点不是模型，而是数据管道。很多人以为数据清洗就是df.dropna()，实际上生产级管道需要解决五个确定性难题：

难题一：时间窗口一致性
金融风控场景中，“过去7天交易额”这个特征，必须保证训练和推理时计算窗口完全一致。我们用Airflow调度，但关键在SQL层面：训练时用WHERE event_time >= CURRENT_DATE - INTERVAL '7 days'，而线上服务用WHERE event_time >= NOW() - INTERVAL '7 days'——看似一样，实则CURRENT_DATE是调度时间点，NOW()是执行时间点，若任务延迟1小时，特征值就差1小时。解决方案是引入锚点时间（Anchor Time）：所有时间窗口基于统一锚点计算，锚点由调度器注入为环境变量ANCHOR_TIME=2024-05-20T12:00:00Z，SQL中统一用WHERE event_time >= (ANCHOR_TIME::timestamp - INTERVAL '7 days')。

难题二：随机性可控
sklearn.train_test_split默认用random_state=42，但42只是个种子，真正要控制的是整个随机生态。我们在管道入口处设置：

import random import numpy as np import torch def set_global_seed(seed: int): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) if torch.cuda.is_available(): torch.cuda.manual_seed_all(seed) set_global_seed(12345) # 固定种子，非42

更重要的是，所有随机操作必须显式传入rng对象，禁止使用全局random模块。例如采样函数写成：

def sample_positive_negative(df: pd.DataFrame, rng: np.random.Generator) -> pd.DataFrame: positives = df[df['label']==1].sample(n=1000, random_state=rng) negatives = df[df['label']==0].sample(n=1000, random_state=rng) return pd.concat([positives, negatives])

难题三：Schema演化兼容
当业务方新增一个字段user_tier，旧模型不能直接报错。我们的解决方案是契约感知的Schema Loader：

# schema.py class DataSchema: def __init__(self, version: str): self.version = version self.fields = { "v1.0": {"user_id": "str", "amount": "float32"}, "v1.1": {"user_id": "str", "amount": "float32", "user_tier": "int8"} } def load(self, df: pd.DataFrame) -> pd.DataFrame: expected_fields = self.fields[self.version] # 缺失字段用契约定义的默认值填充 for field, dtype in expected_fields.items(): if field not in df.columns: default_value = self._get_default_value(dtype) df[field] = default_value # 多余字段直接丢弃 return df[expected_fields.keys()].astype( {k: v for k, v in expected_fields.items() if k in df.columns} )

难题四：数据漂移检测
不是等模型效果下降才行动，而是实时监控。我们在管道末尾插入Drift Detector：

from evidently.metrics import ColumnDriftMetric from evidently.report import Report def detect_drift(reference_df: pd.DataFrame, current_df: pd.DataFrame) -> dict: report = Report(metrics=[ColumnDriftMetric(column_name="amount")]) report.run(reference_data=reference_df, current_data=current_df) result = report.as_dict() drift_score = result["metrics"][0]["result"]["drift_score"] return {"drift_score": drift_score, "drift_detected": drift_score > 0.5}

当漂移分数>0.5，自动触发告警并冻结模型更新，直到数据团队确认原因。

难题五：血缘追踪
谁改动了哪个特征？影响了哪些模型？我们用轻量级血缘图谱：

# lineage.py class LineageTracker: def __init__(self, pipeline_id: str): self.pipeline_id = pipeline_id self.edges = [] def add_edge(self, source: str, target: str, transform: str): self.edges.append({ "source": source, "target": target, "transform": transform, "timestamp": datetime.now().isoformat() }) def save(self): with open(f"lineage/{self.pipeline_id}.json", "w") as f: json.dump(self.edges, f, indent=2)

每次特征计算都调用tracker.add_edge("raw_logs", "cleaned_features", "remove_outliers")，最终生成可视化图谱，审计时直接打开JSON文件就能追溯。

提示：数据管道不是一次性的ETL脚本，而是要像微服务一样有健康检查。我们要求每个管道必须提供/healthz端点，返回{"status":"ok","last_run":"2024-05-20T12:00:00Z","lag_seconds":120}，其中lag_seconds表示最新数据距当前时间的延迟，超过300秒即告警。

3.2 模型训练：如何让训练过程本身成为可验证的工程产物

“From scratch”的训练环节，重点不是调参技巧，而是训练过程的可重现性与可审计性。我们废弃了所有交互式训练方式，强制采用“单命令训练”范式：

# 统一入口 make train \ MODEL_NAME=credit_risk_v2 \ DATASET_VERSION=v202405 \ HYPERPARAMS_FILE=configs/hyperparams_v2.yaml \ GPU_COUNT=2

这个make train背后是严密的工程链路：

步骤1：环境沙箱化
不依赖全局conda环境，而是用pyenv+direnv创建项目专属环境：

# Makefile train: pyenv local 3.10.12 direnv allow pip install -r requirements.lock python train.py --config $(HYPERPARAMS_FILE)

requirements.lock是pip-compile生成的锁定文件，精确到哈希值，确保pip install在任何机器上安装的包版本完全一致。

步骤2：输入数据指纹化
训练前计算数据集MD5：

# train.py def verify_dataset_integrity(dataset_path: str) -> str: with open(dataset_path, "rb") as f: data_hash = hashlib.md5(f.read()).hexdigest() expected_hash = get_expected_hash(dataset_path) # 从config中读取 if data_hash != expected_hash: raise ValueError(f"Dataset corrupted! Expected {expected_hash}, got {data_hash}") return data_hash

每次训练日志都记录dataset_hash: abc123...，回溯问题时直接比对hash就能确认是否用了错误数据。

步骤3：训练过程原子化
训练脚本必须满足ACID原则：

• A（Atomic）：要么全部成功，要么全部回滚。训练目录/tmp/train_20240520_1200，成功后mv到/models/credit_risk_v2_20240520，失败则自动清理临时目录；
• C（Consistent）：训练中任何异常（OOM、CUDA error）都触发cleanup()函数，释放GPU内存、关闭文件句柄；
• I（Isolated）：每个训练进程独占GPU，通过CUDA_VISIBLE_DEVICES=0隔离，避免多任务争抢；
• D（Durable）：每10个epoch保存一次checkpoint，且checkpoint包含完整训练状态（optimizer state、lr scheduler、rng state），支持从中断处继续。

步骤4：输出产物标准化
训练完成生成四个必需文件：

步骤5：效果验证自动化
训练完立即运行三组验证：

# validation.py def run_validation(model_path: str, test_data: pd.DataFrame): # 1. 契约验证：输入输出是否符合schema assert validate_input_schema(test_data) output = model.predict(test_data) assert validate_output_schema(output) # 2. 性能验证：P95延迟<200ms latencies = benchmark_latency(model, test_data) assert np.percentile(latencies, 95) < 0.2 # 3. 效果验证：AUC不低于基线 auc = calculate_auc(model, test_data) baseline_auc = get_baseline_auc("credit_risk_v1") assert auc >= baseline_auc - 0.005 # 允许微小波动

任一验证失败，整个训练流程标记为FAILED，不生成任何产物。

注意：模型训练不是终点，而是新版本发布的起点。我们要求每次训练成功后，自动触发make release，将四个产物打包成credit_risk_v2-20240520.tar.gz，上传到私有MinIO，并更新Git仓库里的MODEL_REGISTRY.md文件，记录版本、作者、变更说明。这样任何人在任何时候都能精准复现任意版本的模型。

3.3 模型服务：用最小技术栈实现企业级SLA

很多团队用Seldon或KServe，结果为了配置一个健康检查折腾两天。我们的服务层坚持“够用就好”原则，核心就三个文件：

main.py—— 服务入口

from fastapi import FastAPI, HTTPException from pydantic import BaseModel import uvicorn import onnxruntime as ort app = FastAPI() # 加载模型一次，全局复用 session = ort.InferenceSession("model.onnx") class PredictionRequest(BaseModel): user_id: str amount: float # ... 其他字段，严格按契约定义 @app.post("/predict") def predict(request: PredictionRequest): try: # 1. 输入校验（Pydantic自动完成） # 2. 数据转换（调用preprocessor.pkl） # 3. ONNX推理 input_data = preprocess(request.dict()) result = session.run(None, {"input": input_data})[0] # 4. 输出校验 return {"score": float(result[0]), "class": "fraud" if result[0] > 0.5 else "normal"} except Exception as e: raise HTTPException(status_code=422, detail=str(e)) @app.get("/healthz") def health_check(): return {"status": "ok", "model_version": "v2.3.1", "uptime": "12h34m"}

Dockerfile—— 极简镜像

FROM python:3.10-slim # 安装ONNX Runtime CPU版（无GPU依赖，体积小） RUN pip install onnxruntime==1.16.0 # 复制应用 COPY main.py . COPY model.onnx . COPY preprocessor.pkl . COPY metadata.json . # 启动 CMD ["uvicorn", "main:app", "--host", "0.0.0.0:8000", "--port", "8000", "--workers", "4"]

镜像大小仅127MB，启动时间<3秒，对比Kubernetes原生部署方案，资源开销降低60%。

config.yaml—— 可配置化

service: host: "0.0.0.0" port: 8000 workers: 4 model: timeout_ms: 200 max_batch_size: 32 logging: level: "INFO" file: "/var/log/ai-service.log"

所有可变参数外置，无需改代码就能调整并发数、超时时间。

关键工程实践：

• 内存泄漏防护：ONNX Runtime默认缓存计算图，我们禁用它：

  sess_options = ort.SessionOptions() sess_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_DISABLE_ALL session = ort.InferenceSession("model.onnx", sess_options)

• 冷启动优化：服务启动时预热一次推理：

  @app.on_event("startup") async def startup_event(): # 预热：用空数据触发一次推理，加载模型到内存 dummy_input = np.zeros((1, 10), dtype=np.float32) session.run(None, {"input": dummy_input})

• 优雅退出：捕获SIGTERM，等待正在处理的请求完成：

  import asyncio from signal import SIGTERM @app.on_event("shutdown") async def shutdown_event(): # 等待所有请求完成 await asyncio.sleep(0.1)

实操心得：不要迷信“高并发框架”。我们压测发现，Uvicorn+ONNX Runtime在4核CPU上轻松支撑300QPS，延迟P95=87ms。而换成TensorRT+Flask后，虽然峰值QPS到500，但P95飙升到156ms，且内存占用翻倍。工程选择的核心指标永远是P95延迟+资源消耗比，不是绝对QPS。

3.4 监控与可观测性：把“黑盒AI”变成透明仪表盘

AI系统最难监控的不是CPU使用率，而是语义层异常——模型还在响应，但预测结果已悄然漂移。我们的监控体系分三层：

基础设施层（Infra）

• CPU/GPU利用率（Prometheus + Node Exporter）
• 内存使用率（重点关注/proc/meminfo中的MemAvailable，而非MemFree）
• 网络连接数（netstat -an | grep :8000 | wc -l）

服务层（Service）

• 请求成功率（HTTP 2xx/4xx/5xx比例）
• P50/P90/P95延迟（单位：毫秒）
• 每秒请求数（RPS）
• 错误详情（按错误码分组，如422: invalid input schema）

模型层（Model）
这才是真正的AI工程壁垒，我们监控四个黄金指标：

实现上，我们用轻量级流式处理替代复杂消息队列：

# monitor.py from adwin import ADWIN import redis adwin = ADWIN(delta=0.002) # 概念漂移检测器 r = redis.Redis() @app.middleware("http") async def log_prediction(request: Request, call_next): response = await call_next(request) if response.status_code == 200: # 提取预测结果和真实标签（需上游传递label） pred = response.json()["score"] label = request.state.label # 从中间件注入 # 更新ADWIN adwin.add_element(1 if abs(pred - label) < 0.1 else 0) if adwin.detected_change(): r.publish("model_alerts", "concept_drift_detected") return response

所有监控数据写入TimescaleDB（PostgreSQL的时序扩展），用Grafana展示。最关键的看板是模型健康度评分（MHS）：

MHS = 100 * ( 0.3 * (1 - latency_p95/200) + 0.3 * (accuracy_current/accuracy_baseline) + 0.2 * (1 - drift_score) + 0.2 * (1 - error_rate) )

当MHS<80，自动触发降级预案：切换到备用模型、限制请求速率、启用缓存。

注意：监控不是为了“看到问题”，而是为了“预防问题”。我们要求所有告警必须附带可执行建议。例如“预测分布漂移”告警，不仅说“p-value=0.005”，还要给出：“建议检查特征user_tier的取值分布，当前75%为NULL，历史均值为12%”。

4. 实操全流程：从空目录到生产服务的12步手把手指南

4.1 初始化项目骨架（5分钟）

在空目录执行：

# 1. 创建项目结构 mkdir ai-engineering-from-scratch && cd ai-engineering-from-scratch mkdir -p {data/{raw,processed},models,src/{pipeline,train,serve,monitor},configs,tests} # 2. 初始化Git + 预提交钩子 git init echo "*.pyc\n__pycache__\n*.log" > .gitignore # 3. 创建基础配置 cat > configs/base.yaml << 'EOF' project: name: "credit_risk" version: "v1.0.0" data: raw_path: "data/raw" processed_path: "data/processed" model: output_dir: "models" contract_file: "src/contract.json" EOF # 4. 生成锁文件 pip install pip-tools echo "onnxruntime==1.16.0" > requirements.in pip-compile requirements.in

这一步建立的不是代码，而是工程纪律的物理载体。.gitignore防止.pyc污染，requirements.txt锁定依赖，base.yaml定义项目元信息——所有后续操作都基于此骨架展开。

4.2 定义数据契约（10分钟）

创建src/contract.json：

{ "version": "1.0", "input_schema": { "user_id": {"type": "string", "min_length": 5}, "amount": {"type": "number", "minimum": 0, "maximum": 1000000}, "transaction_count_7d": {"type": "integer", "minimum": 0}, "user_tier": {"type": "integer", "enum": [1, 2, 3, 4], "default": 1} }, "output_schema": { "score": {"type": "number", "minimum": 0, "maximum": 1}, "class": {"type": "string", "enum": ["normal", "fraud"]} }, "performance": { "max_latency_ms": 200, "max_memory_mb": 1024, "min_accuracy": 0.85 } }

然后编写验证脚本tests/test_contract.py：

import json import pytest def test_contract_valid(): with open("src/contract.json") as f: contract = json.load(f) # 检查必填字段 assert "input_schema" in contract assert "output_schema" in contract # 检查性能约束 assert contract["performance"]["max_latency_ms"] <= 200 if __name__ == "__main__": pytest.main([__file__])

运行pytest tests/test_contract.py，确保契约文档本身是有效的JSON且满足基本约束。

4.3 构建数据管道（30分钟）

创建src/pipeline/etl.py：

import pandas as pd import numpy as np from typing import Dict, Any def clean_data(df: pd.DataFrame) -> pd.DataFrame: """按契约清洗数据""" # 1. 字段校验 required_cols = ["user_id", "amount", "transaction_count_7d"] missing_cols = set(required_cols) - set(df.columns) if missing_cols: raise ValueError(f"Missing columns: {missing_cols}") # 2. 类型转换 df["user_id"] = df["user_id"].astype(str) df["amount"] = pd.to_numeric(df["amount"], errors="coerce") df["transaction_count_7d"] = pd.to_numeric(df["transaction_count_7d"], errors="coerce") # 3. 缺失值处理（按契约约定） df["user_tier"] = df["user_tier"].fillna(1).astype(int) df["amount"] = df["amount"].fillna(0) # 4. 范围校验 df = df[(df["amount"] >= 0) & (df["amount"] <= 1000000)] df = df[(df["transaction_count_7d"] >= 0)] return df if __name__ == "__main__": # 本地测试 test_df = pd.DataFrame({ "user_id": ["U001", "U002"], "amount": [100.5, None], "transaction_count_7d": [5, 12] }) result = clean_data(test_df) print(result)

添加Makefile目标：

.PHONY: etl-test etl-test: python src/pipeline/etl.py

运行make etl-test，验证清洗逻辑正确性。

4.4 实现训练流水线（45分钟）

创建src/train/train.py：

import onnx import onnxruntime as ort from sklearn.ensemble import RandomForestClassifier from sklearn.model_selection import train_test_split import joblib import json def train_model(data_path: str, output_dir: str): # 加载数据 df = pd.read_parquet(data_path) X = df[["amount", "transaction_count_7d"]] y = df["label"] # 划分数据集 X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=12345, stratify=y ) # 训练模型 model = RandomForestClassifier(n_estimators=100, random_state=12345) model.fit(X_train, y_train) # 保存预处理器（这里是StandardScaler示例） scaler = StandardScaler() X_train_scaled = scaler.fit_transform(X_train) joblib.dump(scaler, f"{output_dir}/preprocessor.pkl") # 转换为ONNX initial_type = [('float_input', FloatTensorType([None, X_train.shape[1]]))] onnx_model = convert_sklearn(model, initial_types=initial_type) onnx.save(onnx_model, f"{output_dir}/model.onnx") # 保存元数据 metadata = { "model_version": "v1.0.0", "training_timestamp": datetime.now().isoformat(), "git_commit": subprocess.check_output(["git", "rev-parse", "HEAD"]).decode().strip(), "dataset_hash": hashlib.md5(open(data_path, "rb").read()).hexdigest() } with open(f"{output_dir}/metadata.json", "w") as f: json.dump(metadata, f, indent=2) if __name__ == "__main__": train_model("data/processed/train.parquet", "models/credit_risk_v1")

注意：这里省略了convert_sklearn的导入，实际需安装skl2onnx。关键点在于所有输出文件都按契约生成，且metadata.json包含可追溯的完整信息。

4.5 开发服务接口（20分钟）

创建src/serve/main.py：

from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel import joblib import numpy as np import onnxruntime as ort app = FastAPI() # 全局加载 session = ort.InferenceSession("models/credit_risk_v1/model.onnx") preprocessor = joblib.load("models/credit_risk_v1/preprocessor.pkl") class PredictionRequest(BaseModel): user_id: str amount: float transaction_count_7d: int user_tier: int = 1 @app.post("/predict") def predict(request: PredictionRequest): try: # 输入校验由Pydantic自动完成 input_data = np.array([[request.amount, request.transaction_count_7d]]) input_scaled = preprocessor.transform(input_data) # ONNX推理 result = session.run(None, {"float_input": input_scaled.astype(np.float32)})[0] score = float(result[0][1]) # 假设第二列为正样本概率 return { "score": score, "class": "fraud" if score > 0.5 else "normal" } except Exception as e: raise HTTPException(status_code=422, detail=f"Prediction failed: {str(e)}") @app.get("/healthz") def health(): return {"status": "ok"}

4.6 构建Docker镜像（10分钟）

创建Dockerfile：

FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY src/serve/main.py . COPY models/credit_risk_v1/ models/credit_risk_v1/ EXPOSE 8000 CMD ["uvicorn", "main:app", "--host", "0.0.0.0:8000", "--port", "8000", "--workers", "2"]

构建并测试：

docker build -t ai-service . docker run -p 8000:8000 ai-service curl http://localhost:8000/healthz # 应返回{"status":"ok"}

4.7 配置CI/CD流水线（25分钟）

在.github/workflows/ci.yml中：

name: AI Engineering CI on: [push] jobs: test-contract: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - name: Set up Python uses: actions/setup-python@v4 with: python-version: '3.10' - name: Install dependencies run: | python -m pip install --upgrade pip pip install pytest - name: Run contract tests run: pytest tests/test_contract.py build-model: needs: test-contract runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - name: Set up Python