从零构建AI工程：Python/TypeScript/Rust协同实践

1. 为什么“从零构建AI工程”不是写个Hello World那么简单

“AI Engineering from Scratch”这个标题，乍看像极了那些教你怎么用几行Python调用Hugging Face模型的入门教程——但其实它背后藏着一个被严重低估的真相：真正的AI工程，90%的工作量不在模型训练本身，而在让模型能稳定、可维护、可扩展、可监控地跑在真实业务场景里。我带过三个AI产品团队，每次新项目启动，最常听到的抱怨不是“模型精度不够”，而是“训练脚本本地能跑，一上服务器就报错”“API响应忽快忽慢，日志里找不到线索”“数据管道凌晨三点崩了，没人知道是上游ETL还是下游特征服务的问题”。这些都不是算法问题，是工程问题。而“from scratch”的核心价值，恰恰在于亲手把每一块砖垒起来，才能看清承重墙在哪、裂缝会从哪开始裂、补丁该焊在哪个焊点上。

这和你搜到的“scratch desktop for windows”或“scratch jr desktop”完全不是一回事——那是面向儿童的图形化编程启蒙工具；而这里的“scratch”，指的是不依赖任何现成AI平台（如SageMaker、Vertex AI、Model Studio）封装的黑盒能力，从操作系统层、依赖管理、代码结构、CI/CD流水线、监控告警，全部自己定义、自己实现、自己验证。它更接近《Build a Large Language Model from Scratch》那本书的精神内核：不是为了造轮子而造轮子，而是为了理解轮子为什么这么设计、在什么路况下会打滑、换胎时该拧多大扭矩。

关键词里反复出现的Python、TypeScript、Rust，绝非随意堆砌。它们各自承担着不可替代的工程角色：Python是AI研发的事实标准语言，生态丰富但GIL限制并发；TypeScript是前端与后端服务（尤其是API网关、管理后台）的强类型保障，避免“undefined is not a function”这种线上事故；Rust则是高性能组件（如实时推理引擎、向量索引服务、日志采集Agent）的首选，内存安全+零成本抽象让它能在高吞吐场景下扛住压力。这三者不是并列选项，而是分层协作的工程栈：Python负责算法迭代与实验，TypeScript负责用户交互与服务编排，Rust负责性能敏感的底层模块。忽略其中任何一层，所谓的“from scratch”都只是空中楼阁。

提示：很多团队误以为“用Docker打包Python脚本就是工程化”，结果发现镜像体积动辄2GB，启动耗时47秒，CPU占用率波动超过300%，根本无法纳入K8s滚动更新流程。真正的“from scratch”，第一步就得从Dockerfile的每一行COPY指令开始较真。

2. 工程骨架：为什么必须亲手定义项目结构而非套用Cookiecutter模板

市面上有几十个号称“AI Project Template”的GitHub仓库，从cookiecutter-data-science到mlflow-project-template，看似省事，实则埋雷。我曾接手一个用cookiecutter-data-science搭建的推荐系统，上线三个月后，运维同事深夜打电话说：“特征计算任务卡在/tmp目录，磁盘爆了，但代码里根本没看到任何临时文件清理逻辑。”翻源码才发现，模板默认把所有中间数据写入/tmp，而生产环境/tmp挂载的是内存盘，容量仅2GB——这是模板作者在个人笔记本上测试时根本不会遇到的边界条件。

真正的“from scratch”项目结构，必须回答五个硬性问题：

• 数据生命周期如何闭环？训练数据、验证数据、线上推理数据、A/B测试数据，是否物理隔离？版本如何追踪？
• 代码职责是否严格分层？src/models/里能否混入数据预处理逻辑？src/utils/是否允许直接调用数据库连接？
• 环境配置如何零歧义？.env文件是否被Git忽略？Docker Compose里的environment:变量是否与Python代码里的os.getenv()一一对应？
• 依赖管理是否可重现？requirements.txt是否包含--find-links指向私有PyPI？poetry.lock是否锁定到patch版本？
• 测试覆盖是否分层有效？单元测试是否Mock了所有外部API？集成测试是否使用真实Redis实例而非fakeredis？

我们最终采用的结构如下（已脱敏，但保留关键设计逻辑）：

ai-engineering-from-scratch/ ├── .github/ # CI/CD工作流定义（非模板生成，手动编写） │ └── workflows/ │ ├── ci.yml # 运行单元测试+类型检查+安全扫描 │ └── cd.yml # 部署到Staging环境，触发金丝雀发布 ├── docker/ # 所有Docker相关文件集中管理 │ ├── base/ # 基础镜像（Python 3.11 + Rust 1.75 + Node.js 18） │ │ ├── Dockerfile # 多阶段构建：build-stage编译Rust，final-stage只含运行时 │ │ └── requirements.txt # 仅含生产必需依赖（无pytest、black等dev工具） │ ├── api/ # TypeScript编写的FastAPI兼容网关（用tRPC实现类型安全） │ │ ├── Dockerfile # 构建时执行tsc --noEmit + type-check，失败即中断 │ │ └── tsconfig.json # 启用`"strict": true`和`"skipLibCheck": false` │ └── inference/ # Rust编写的轻量级推理服务（基于Tch-rs） │ ├── Dockerfile # 使用musl静态链接，镜像体积<80MB │ └── Cargo.toml # 显式声明`[profile.release]`优化参数 ├── src/ # Python核心代码（遵循Clean Architecture） │ ├── core/ # 领域层（纯Python，无框架依赖） │ │ ├── models/ # 数据模型（Pydantic v2 BaseSettings） │ │ └── services/ # 业务逻辑（如FeatureService、ModelOrchestrator） │ ├── infrastructure/ # 基础设施层（适配器） │ │ ├── data/ # 数据源适配器（S3Client、PostgresRepository） │ │ └── ml/ # ML平台适配器（MLflowTrackingClient、ONNXRuntimeInference） │ └── application/ # 应用层（协调者） │ └── api/ # FastAPI路由（仅做输入校验+调用core层） ├── rust/ # 独立Rust工作区（非Python子模块） │ ├── inference-engine/ # 核心推理引擎（支持ONNX/TorchScript） │ ├── vector-index/ # 基于HNSW的向量索引（内存+磁盘混合存储） │ └── Cargo.toml # workspace定义，禁止跨crate循环依赖 ├── typescript/ # 独立TypeScript工作区（非嵌套在Python项目中） │ ├── gateway/ # tRPC网关（暴露Python API的类型安全接口） │ ├── dashboard/ # React管理后台（使用Vite+TanStack Query） │ └── tsconfig.base.json # 基础类型配置，所有子项目继承 ├── tests/ # 测试金字塔（非全放tests/下，按层分离） │ ├── unit/ # 核心层单元测试（pytest + pytest-mock） │ ├── integration/ # 基础设施层集成测试（启动真实Postgres+Redis容器） │ └── e2e/ # 端到端测试（Playwright模拟用户操作） ├── scripts/ # 可执行脚本（非Makefile，因Windows兼容性差） │ ├── setup-dev.sh # 开发环境一键初始化（含Rust toolchain + Node.js版本检查） │ └── deploy-staging.py # Staging部署脚本（调用AWS CLI + kubectl，非Ansible） └── pyproject.toml # Poetry配置（显式禁用`virtualenvs.create = false`，强制使用venv）

这个结构的关键设计点在于物理隔离与契约先行：Python、Rust、TypeScript各自拥有独立的Cargo.toml、package.json、pyproject.toml，通过明确定义的API契约（如OpenAPI 3.1规范、Protobuf Schema）交互，而非共享代码库。例如，Rust推理引擎只暴露gRPC接口，Python应用层通过tch-rs生成的客户端调用；TypeScript网关则通过HTTP代理转发请求，并注入JWT鉴权头。这样做的好处是：当需要将推理引擎迁移到WebAssembly时，只需替换Rust crate，Python和TS代码完全无需修改。

注意：docker/base/Dockerfile中禁用pip install -r requirements.txt，改用pip install --no-deps --find-links ./wheels/ -f ./wheels/ -r requirements.txt。原因？我们自建PyPI镜像时发现，某些包（如torch）的wheel文件名包含CUDA版本号（torch-2.1.0+cu118-cp311-cp311-linux_x86_64.whl），直接pip install会因版本匹配失败而回退到源码编译，耗时增加12分钟。提前下载并重命名wheel文件，是保证CI构建稳定性的关键细节。

3. 类型系统战争：Python、TypeScript、Rust如何协同而不互相污染

很多团队试图用单一语言统一AI工程栈，结果要么是Python强行写异步服务导致GIL锁死，要么是Rust硬啃机器学习导致生态缺失。真正的“from scratch”必须接受类型系统的分层治理：Python用typing做轻量契约，TypeScript用strict模式做前端防御，Rust用所有权系统做底层兜底。三者不是竞争关系，而是像三层滤网——越靠近数据源头（Rust），过滤越严格；越靠近用户界面（TS），反馈越及时；中间层（Python）则保持最大灵活性。

先看Python层的类型实践。我们禁用mypy的--follow-imports=skip，强制所有导入模块都经过类型检查。关键改造点有三处：

1. 配置管理：弃用os.getenv()，改用pydantic.BaseSettings，自动转换类型并校验必填项：

from pydantic import BaseSettings, validator class Settings(BaseSettings): DATABASE_URL: str MODEL_PATH: str INFERENCE_TIMEOUT_MS: int = 5000 @validator("INFERENCE_TIMEOUT_MS") def timeout_must_be_positive(cls, v): if v <= 0: raise ValueError("Timeout must be > 0") return v settings = Settings() # 启动时即校验，而非运行时报错

2. 数据模型：所有输入输出模型继承pydantic.BaseModel，启用extra = "forbid"防止字段污染：

class PredictionRequest(BaseModel): user_id: str features: List[float] model_version: str = "v1.2.0" # 默认值明确，非None class PredictionResponse(BaseModel): prediction: float confidence: float latency_ms: int # 若请求体含未知字段"debug_mode: true"，直接422错误，不进入业务逻辑

3. 服务接口：用Protocol定义契约，而非继承抽象基类：

from typing import Protocol, List class FeatureStore(Protocol): def get_features(self, user_id: str) -> List[float]: ... def update_features(self, user_id: str, features: List[float]) -> None: ... # 实现类可自由选择Postgres/Redis/S3，只要满足Protocol即可 class PostgresFeatureStore(FeatureStore): def __init__(self, conn: psycopg2.Connection): ...

TypeScript层则更激进。我们禁用any和@ts-ignore，所有API响应类型由OpenAPI 3.1 YAML自动生成（使用openapi-typescript）。关键约束包括：

• tsconfig.json中"noImplicitAny": true、"strictNullChecks": true、"exactOptionalPropertyTypes": true全开；
• 禁止interface中使用?可选属性，改用Partial<T>显式声明；
• 所有HTTP客户端（如Axios）封装为createApiClient()函数，返回类型安全的ApiMethods对象：

// 自动生成的types/api.ts export interface PredictionRequest { user_id: string; features: number[]; model_version?: string; } export interface PredictionResponse { prediction: number; confidence: number; latency_ms: number; } // 封装层 export const createApiClient = (baseUrl: string) => ({ predict: (req: PredictionRequest) => axios.post<PredictionResponse>(`${baseUrl}/predict`, req), });

Rust层是类型安全的终极防线。我们不用Box<dyn Trait>，而用enum显式枚举所有可能错误：

#[derive(Debug, Clone, PartialEq)] pub enum InferenceError { ModelNotFound(String), InvalidInput(String), Timeout(std::time::Duration), HardwareFailure(String), } impl std::fmt::Display for InferenceError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { InferenceError::ModelNotFound(path) => write!(f, "Model not found: {}", path), InferenceError::InvalidInput(msg) => write!(f, "Invalid input: {}", msg), InferenceError::Timeout(dur) => write!(f, "Inference timeout: {:?}", dur), InferenceError::HardwareFailure(msg) => write!(f, "Hardware failure: {}", msg), } } }

这样，Python调用Rust gRPC服务时，错误会被序列化为JSON字符串，TS网关解析后可直接映射为ApiError类型，前端能精准显示“模型未找到”而非笼统的“服务异常”。

实操心得：TypeScript的unknown类型是救命稻草。我们所有从localStorage读取的配置都标注为unknown，再用Zod Schema校验：

const configSchema = z.object({ theme: z.enum(["light", "dark"]) }); const rawConfig = localStorage.getItem("ui-config"); const config = configSchema.parse(rawConfig); // 校验失败抛ZodError，非运行时崩溃

这比JSON.parse()+as Config安全十倍——后者在配置格式变更时静默失败，前者立刻中断并提示具体字段错误。

4. 构建流水线：为什么CI/CD必须为AI工程定制而非复用通用模板

通用CI/CD模板（如GitHub Actions的setup-python）对AI工程是灾难。我见过最典型的事故：某次模型更新后，CI流水线通过，但线上服务OOM。排查发现，setup-python默认安装最新版numpy（1.26.0），而模型训练时用的numpy==1.24.3，新版在特定矩阵运算中内存分配策略改变，导致推理时内存峰值翻倍。这不是代码bug，是环境漂移。

我们的CI/CD流水线（.github/workflows/ci.yml）设计原则是：每个阶段只做一件事，且这件事必须可验证、可回滚、可审计。具体拆解如下：

4.1 环境一致性校验阶段

- name: Validate Python Environment run: | python -c " import sys, numpy, torch print(f'Python {sys.version}') print(f'NumPy {numpy.__version__}') print(f'PyTorch {torch.__version__}') assert numpy.__version__ == '1.24.3', 'NumPy version mismatch' assert torch.__version__ == '2.1.0+cu118', 'PyTorch version mismatch' "

此步骤在actions/setup-python@v4之后立即执行，强制校验实际安装版本。若失败，整个流水线终止，避免后续测试在错误环境中进行。

4.2 类型安全双校验阶段

- name: Check Python Types run: mypy --show-error-codes --disallow-untyped-defs --disallow-incomplete-defs src/ - name: Check TypeScript Types run: cd typescript && npm ci && npx tsc --noEmit --skipLibCheck

关键点：mypy启用--disallow-untyped-defs（禁止无类型函数）和--disallow-incomplete-defs（禁止部分类型注解），确保类型契约完整；TS校验禁用--emit，只做类型检查，避免生成JS文件污染工作区。

4.3 模型验证阶段（核心创新点）

这是AI工程特有的环节。我们不只跑单元测试，还验证模型行为一致性：

- name: Validate Model Behavior run: | # 加载训练时保存的参考模型（.pt格式） python -c " import torch ref_model = torch.load('models/ref_v1.2.0.pt') ref_model.eval() # 加载当前PR中的模型（.onnx格式） import onnxruntime as ort ort_session = ort.InferenceSession('models/current.onnx') # 用相同输入测试 test_input = torch.randn(1, 128) ref_output = ref_model(test_input).detach().numpy() ort_output = ort_session.run(None, {'input': test_input.numpy()})[0] # 计算最大绝对误差（MAE < 1e-5视为一致） import numpy as np mae = np.max(np.abs(ref_output - ort_output)) assert mae < 1e-5, f'Model drift detected: MAE={mae}' "

此步骤确保ONNX导出未引入数值误差，避免因量化或算子替换导致线上预测偏差。

4.4 安全扫描阶段（非摆设）

- name: Scan Dependencies uses: anchore/scan-action@v4 with: image: ${{ env.REGISTRY }}/ai-api:${{ github.sha }} policy-path: .anchore/policy.yaml fail-on-policy: true

policy.yaml中明确禁止：

• CVE-2023-XXXXX及以上严重漏洞（CVSS >= 7.0）；
• 包含eval()、exec()、pickle.loads()的第三方库（如requests旧版存在反序列化风险）；
• 未签名的PyPI包（allow_unsigned: false）。

4.5 部署阶段：金丝雀发布的原子性保障

CD流水线（.github/workflows/cd.yml）不直接更新Production，而是：

1. 部署到Staging集群（独立Namespace）；
2. 运行端到端测试（Playwright访问Staging URL，验证UI+API+模型输出）；
3. 若通过，触发金丝雀发布：将5%流量切至新版本，监控10分钟内错误率、P95延迟、GPU显存使用率；
4. 全部指标达标后，自动扩至100%；任一指标超标，自动回滚至前一版本（K8s Deployment的revisionHistoryLimit: 5确保可追溯）。

关键经验：K8s的readinessProbe不能只检查HTTP 200，必须包含模型加载状态。我们在FastAPI中添加/healthz端点：

@app.get("/healthz") async def healthz(): if not model_manager.is_ready(): # 检查模型是否完成warmup raise HTTPException(status_code=503, detail="Model not ready") return {"status": "ok"}

否则K8s可能在模型还在加载时就将流量导入，导致首批请求超时。

5. 监控告警体系：为什么AI服务的Metrics必须包含业务语义而不仅是CPU利用率

传统监控（如Prometheus+Grafana）对AI服务是盲人摸象。CPU使用率低于30%不代表服务健康——可能模型卡在IO等待，或特征计算陷入死循环。我们构建的监控体系分三层：基础设施层（CPU/Memory）、服务层（HTTP状态码/延迟）、业务层（模型性能指标）。后两者才是AI工程的核心。

5.1 业务指标埋点规范

所有Python服务必须注入以下指标（使用prometheus_client）：

• model_prediction_latency_seconds_bucket：按模型版本、输入长度分桶的P50/P90/P99延迟；
• model_prediction_error_total：按错误类型（model_not_found、invalid_input、timeout）计数；
• feature_computation_duration_seconds：特征计算耗时（区分实时特征vs离线特征）；
• data_drift_alert_total：数据分布偏移告警（使用KS检验，阈值动态调整）。

关键设计：指标标签（Labels）必须包含业务维度。例如：

# 错误计数指标 error_counter = Counter( 'model_prediction_error_total', 'Total number of prediction errors', ['model_version', 'error_type', 'user_segment'] # 用户分群是业务关键维度 ) # 在预测函数中 try: result = model.predict(input_data) except ModelNotFoundError as e: error_counter.labels( model_version=settings.MODEL_VERSION, error_type='model_not_found', user_segment=get_user_segment(user_id) # 从Redis缓存获取用户分群 ).inc()

5.2 实时数据漂移检测

我们不依赖离线批处理，而是在特征服务中嵌入实时漂移检测：

from scipy.stats import ks_2samp import numpy as np class DriftDetector: def __init__(self, reference_data: np.ndarray, threshold: float = 0.05): self.reference_data = reference_data self.threshold = threshold def check_drift(self, current_batch: np.ndarray) -> bool: # KS检验比较当前批次与参考分布 _, p_value = ks_2samp(self.reference_data, current_batch) return p_value < self.threshold # 在特征计算Pipeline中 drift_detector = DriftDetector(reference_features) if drift_detector.check_drift(new_features): alert_manager.send_alert( f"Data drift detected in feature {feature_name}", severity="warning" )

告警触发后，自动冻结该特征在模型中的权重，并通知数据工程师核查上游数据源。

5.3 推理服务的Rust级监控

Rust推理引擎暴露/metrics端点，但指标更底层：

• inference_engine_gpu_memory_bytes：GPU显存实际使用量（非nvidia-smi报告的已分配量）；
• inference_engine_queue_length：等待推理的请求队列长度；
• inference_engine_cache_hit_ratio：模型缓存命中率（避免重复加载大模型）。

这些指标通过prometheus-client的Gauge和Counter暴露，与Python服务指标在同一Prometheus实例中抓取，便于关联分析。例如，当queue_length突增而gpu_memory_bytes未变化，说明是CPU瓶颈；若gpu_memory_bytes飙升，则需检查模型加载逻辑。

5.4 告警策略：避免告警疲劳

我们禁用“CPU > 80%”这类基础设施告警，改为：

• P99延迟 > 2秒持续5分钟→ 触发Slack通知（On-Call工程师）；
• 数据漂移告警连续3次→ 创建Jira工单并邮件通知数据团队；
• 模型错误率 > 1%持续10分钟→ 自动触发回滚流程（调用K8s API切换Deployment镜像）。

所有告警均附带根因建议：

“Alert: P99 latency > 2s for model v1.2.0
Root Cause Suggestion: Checkfeature_computation_duration_secondsforuser_segment=premium— historical data shows correlation with Redis latency spikes.”

这比单纯说“服务变慢”有用十倍。

最后分享一个血泪教训：某次上线新模型后，监控显示错误率正常，但业务方反馈转化率下降。排查发现，模型预测的confidence分数整体偏低，导致下游排序策略失效。自此，我们新增指标prediction_confidence_distribution（直方图），并设置告警：当confidence的P10值下降超过20%，立即告警。技术指标必须对齐业务结果，否则就是无效监控。

6. 团队协作契约：为什么AI工程师必须懂Rust的borrow checker而数据科学家要写TypeScript测试

“from scratch”的最大挑战不是技术，是认知对齐。很多AI团队失败，源于Python数据科学家、TypeScript前端、Rust后端工程师各说各话。我们强制推行三项协作契约：

6.1 接口契约先行（Contract-First Development）

所有跨语言调用，必须先定义契约：

• Python与Rust间：用Protobuf定义gRPC服务（.proto文件存于/protos/）；
• Python与TypeScript间：用OpenAPI 3.1 YAML定义REST API（/openapi.yaml）；
• 数据库Schema：用SQLAlchemy Core定义，生成CREATE TABLE语句（非ORM模型）。

契约文件由架构师审核，三方代表签字确认。任何变更必须：

1. 更新契约文件；
2. 生成新客户端代码；
3. 运行契约兼容性测试（验证旧客户端能否调用新服务）；
4. 方可合并。

6.2 跨语言测试责任共担

• Python工程师：编写integration/测试，启动真实Rust服务容器，验证gRPC调用；
• Rust工程师：编写cargo test，模拟Python客户端发送请求，验证响应格式；
• TypeScript工程师：编写Playwright测试，访问API网关，断言JSON响应符合OpenAPI定义。

我们有个硬性规定：任何PR若未覆盖其修改的契约变更，CI自动拒绝。例如，若openapi.yaml中新增/v2/predict端点，TS测试必须包含对该端点的调用，Python测试必须包含对该端点的mock，否则流水线失败。

6.3 文档即代码（Documentation as Code）

所有文档存于/docs/，用Markdown编写，但关键部分嵌入可执行代码块：

## 模型加载流程 1. Python应用调用Rust gRPC服务： ```python # This code is executed in CI to verify correctness from inference_pb2 import LoadModelRequest from inference_pb2_grpc import InferenceServiceStub stub = InferenceServiceStub(channel) response = stub.LoadModel(LoadModelRequest(model_path="/models/v1.2.0.onnx")) assert response.status == "success"

2. Rust服务验证ONNX模型签名...

CI流水线会提取所有` ```python `代码块，用`pytest`运行，确保文档示例永远与代码同步。过期文档比没有文档更危险。 > 个人体会：推行这些契约初期阻力巨大，数据科学家抱怨“写TypeScript测试浪费时间”，Rust工程师说“Protobuf太重”。但三个月后，跨团队联调时间从平均3天缩短到2小时，线上事故中90%的根因能在5分钟内定位到具体契约违反点。工程复杂度无法消除，但可以将其转化为可管理的契约成本——这才是“from scratch”的真正价值。