从零构建AI工程能力：两年踩坑总结与全链路落地指南

从零构建AI工程能力这件事，我前前后后折腾了差不多两年。最开始的时候，我和大多数人一样，觉得"AI工程"嘛，无非就是调调API、跑跑模型、搭个Demo。结果真到了要把一个模型从Jupyter Notebook搬到生产环境的时候，才发现自己连最基本的工程化思维都没建立起来。模型在本地跑得好好的，一上线就各种问题：推理延迟不稳定、内存泄漏、版本管理混乱、监控缺失。那段时间我几乎每天都在填坑，也正是这些坑逼着我把AI工程的全链路从头到尾捋了一遍。

这篇内容我想聊的是"从零开始搭建AI工程能力"这件事。不是那种泛泛而谈的"AI工程师需要会什么"，而是我自己踩过坑之后总结出来的一套可落地的路径。适合谁看？如果你是一个会写Python、懂一些机器学习基础，但不知道如何把模型变成真正可用系统的开发者，那这篇内容就是写给你的。如果你已经在大厂做AI工程，里面的一些实操细节和避坑经验应该也能给你一些参考。我会从环境搭建、数据处理、模型训练、部署上线、监控运维这几个核心环节逐一拆解，每个环节都会说清楚"为什么这么做"以及"我踩过什么坑"。

1. 为什么"从零构建"比想象中难——AI工程的核心挑战拆解

1.1 学术代码和工程代码之间的鸿沟

我先说一个很多人不愿意承认的事实：大部分AI相关的教程和课程，教的是"实验代码"，不是"工程代码"。这两者之间的差距，比很多人想象的要大得多。

实验代码的目标是"跑通就行"。你在Notebook里写一个模型，数据直接pd.read_csv读进来，模型直接model.fit()训练，评估指标打印出来看一眼，完事。这种代码的生命周期可能就几个小时，跑完就扔。但工程代码不一样，它需要被别人阅读、被别人修改、被反复运行、在不同的环境下表现一致。这就意味着你需要考虑模块化、配置管理、错误处理、日志记录、测试覆盖等等一系列问题。

我举个具体的例子。在实验阶段，你可能会这样写数据预处理：

import pandas as pd from sklearn.preprocessing import StandardScaler df = pd.read_csv("data.csv") scaler = StandardScaler() df["feature_scaled"] = scaler.fit_transform(df[["feature"]])

这段代码在Notebook里跑没问题。但如果你把它放到生产环境，问题就来了：data.csv的路径是硬编码的，换一个环境就找不到；scaler没有保存，推理的时候没法用同样的参数做变换；没有处理缺失值，线上数据有NaN就直接崩了；没有日志，出了问题根本不知道是哪一步出的错。

工程化的写法应该是这样的：

import logging import joblib import pandas as pd from pathlib import Path from sklearn.preprocessing import StandardScaler logger = logging.getLogger(__name__) class FeaturePreprocessor: def __init__(self, config: dict): self.config = config self.scaler = StandardScaler() self.is_fitted = False def fit(self, df: pd.DataFrame) -> None: feature_col = self.config["feature_column"] if df[feature_col].isna().any(): logger.warning("Feature column contains NaN, filling with median") df[feature_col] = df[feature_col].fillna(df[feature_col].median()) self.scaler.fit(df[[feature_col]]) self.is_fitted = True logger.info("Preprocessor fitted successfully") def transform(self, df: pd.DataFrame) -> pd.DataFrame: if not self.is_fitted: raise RuntimeError("Preprocessor must be fitted before transform") feature_col = self.config["feature_column"] df["feature_scaled"] = self.scaler.transform(df[[feature_col]]) return df def save(self, path: str) -> None: joblib.dump(self.scaler, Path(path) / "scaler.pkl") logger.info(f"Preprocessor saved to {path}")

你看，功能是一样的，但工程代码多了配置管理、错误处理、日志、持久化。这些东西在实验阶段看起来是"多余的"，但在生产环境里，每一样都是必需的。

1.2 AI工程和传统后端工程的本质差异

有人可能会说，不就是把代码写规范一点吗，有什么难的？但AI工程和传统后端工程有一个本质差异：不确定性。

传统后端工程的行为是确定的。你写一个排序算法，输入一组数据，输出一定是可预期的。但AI工程不一样，模型的行为受到数据分布、训练随机性、硬件精度等多种因素影响。同一个模型，在不同的机器上跑，结果可能有微小差异；同一批数据，在不同的时间点推理，结果可能因为数据漂移而变化。

这种不确定性带来了一系列传统工程不需要考虑的问题：

• 可复现性：你需要确保同样的代码、同样的数据、同样的配置，能产出同样的模型。这要求你管理好随机种子、依赖版本、硬件环境。
• 数据版本管理：代码可以用Git管理，但数据呢？模型呢？你需要一套类似Git的机制来管理数据和模型的版本。
• 持续监控：传统后端服务挂了会报警，但AI模型不会"挂"，它只会慢慢变差。你需要监控模型的输入分布、输出分布、预测准确率等指标。
• 回滚机制：模型更新后效果变差了怎么办？你需要能快速回滚到上一个版本。

这些问题在实验阶段都不会暴露，只有当你真正把模型部署到生产环境、面对真实用户的时候，才会一个一个冒出来。

1.3 一个典型的失败案例复盘

我说一个我自己经历过的案例。之前做过一个文本分类的服务，实验阶段准确率92%，F1值0.91，看起来很不错。部署上线之后，第一周表现正常，第二周开始有用户反馈"分类结果不对"。我去查日志，发现模型的输入文本长度分布发生了变化——实验阶段的数据平均长度是50个字符，但线上用户输入的文本平均长度是200个字符。

问题出在哪里？我的模型是基于固定长度（64个token）截断的，实验阶段的数据短，截断影响不大；但线上数据长，截断之后丢失了大量信息，导致分类准确率大幅下降。这个问题在实验阶段完全没暴露，因为实验数据和生产数据的分布不一致。

这个案例告诉我几个道理：第一，实验数据的分布必须尽可能接近生产数据；第二，模型上线后必须持续监控输入分布；第三，任何假设（比如"文本不会太长"）都需要被验证。

2. 环境与工具链的搭建——别在第一步就埋雷

2.1 Python环境管理：为什么我最终选择了uv

Python环境管理这件事，我试过很多方案：virtualenv、conda、pipenv、poetry，最后稳定在uv上。不是说其他方案不好，而是uv在速度和依赖解析上确实有优势，尤其是当你需要频繁创建和销毁环境的时候。

先说为什么不用conda。conda的问题在于它太重了，安装一个包可能要下载几百MB的依赖，而且conda的依赖解析有时候会陷入"死循环"，半天解不出来。对于AI工程来说，你经常需要切换不同的项目、不同的模型、不同的框架版本，conda的环境切换成本太高。

poetry是一个不错的选择，依赖管理清晰，pyproject.toml也很规范。但poetry的安装速度偏慢，尤其是在国内网络环境下，经常卡在下载依赖上。

uv是Rust写的，安装速度快得离谱。我实测下来，创建一个包含PyTorch、transformers、fastapi的环境，uv大概30秒搞定，pip可能要3-5分钟。而且uv兼容pyproject.toml和requirements.txt，迁移成本很低。

具体操作：

# 安装uv curl -LsSf https://astral.sh/uv/install.sh | sh # 创建项目 uv init my-ai-project cd my-ai-project # 添加依赖 uv add torch transformers fastapi uvicorn # 创建虚拟环境并激活 uv venv source .venv/bin/activate

提示：uv默认会把虚拟环境放在项目目录下的.venv文件夹里，这和poetry的行为一致。如果你习惯conda那种全局环境，可以用uv venv --path指定路径。

2.2 依赖版本锁定：一次血泪教训

我踩过最大的坑之一，就是没有锁定依赖版本。有一次部署一个模型服务，本地测试一切正常，部署到服务器上就报错。查了半天发现是transformers库的版本不一致——本地是4.36，服务器上自动装了最新的4.40，而4.40里某个API的签名变了。

从那以后，我所有的项目都会做严格的版本锁定。uv的uv.lock文件会自动记录所有依赖的精确版本，包括间接依赖。部署的时候用uv sync --frozen，确保服务器上的环境和本地完全一致。

# 本地开发时 uv add transformers==4.36.0 # 部署时 uv sync --frozen

如果你用的是pip，那就用pip freeze > requirements.txt，但要注意pip freeze会把所有间接依赖都列出来，有时候会有平台相关的包（比如pywin32只在Windows上有），跨平台部署的时候需要手动清理。

2.3 GPU环境配置的常见陷阱

GPU环境配置是另一个容易踩坑的地方。CUDA版本、cuDNN版本、PyTorch版本，这三者之间的兼容性关系非常微妙。我见过太多人因为版本不匹配，折腾一整天都跑不起来。

我的建议是：不要自己手动装CUDA和cuDNN。直接用PyTorch官方提供的预编译版本，它会自动带上匹配的CUDA运行时。你只需要确保显卡驱动版本足够新就行。

# 查看显卡驱动支持的CUDA版本 nvidia-smi # 根据驱动版本选择PyTorch # 例如驱动支持CUDA 12.1 uv add torch --index-url https://download.pytorch.org/whl/cu121

注意：nvidia-smi显示的CUDA版本是驱动支持的最高版本，不是你必须使用的版本。PyTorch自带的CUDA运行时可以低于这个版本，但不能高于。

还有一个坑是显存管理。AI工程里经常需要同时加载多个模型，或者处理大批量数据，显存很容易爆。我的经验是：第一，用torch.cuda.empty_cache()及时释放不用的显存；第二，用torch.cuda.memory_summary()查看显存分配情况；第三，如果显存实在不够，考虑用梯度累积、混合精度训练、模型量化等手段降低显存占用。

3. 数据管道的工程化——从脏数据到可用特征

3.1 数据验证：在模型训练之前拦住问题

数据验证是我认为AI工程里最被低估的环节。很多人拿到数据直接就往模型里灌，结果训练出来的模型效果差，回头查半天才发现是数据里有异常值、缺失值、重复值。

我的做法是在数据管道的最前面加一层验证。验证的内容包括：schema验证（字段类型、字段名是否正确）、范围验证（数值是否在合理范围内）、分布验证（和训练数据的分布是否一致）、完整性验证（是否有缺失值、重复值）。

import pandera as pa from pandera import Column, DataFrameSchema, Check schema = DataFrameSchema({ "text": Column(str, Check(lambda s: s.str.len() > 0)), "label": Column(int, Check.isin([0, 1, 2])), "timestamp": Column("datetime64[ns]"), "user_id": Column(str, nullable=True), }) def validate_data(df): try: schema.validate(df, lazy=True) return True except pa.errors.SchemaErrors as e: logger.error(f"Data validation failed: {e.failure_cases}") return False

pandera是一个很好用的数据验证库，它支持链式API，可以定义复杂的验证规则。lazy=True参数会让它一次性收集所有错误，而不是遇到第一个错误就抛出异常，这样你能一次性看到所有问题。

3.2 特征工程的模块化设计

特征工程最容易犯的错误是把所有逻辑写在一个大函数里，改一个特征要动整个函数，测试也没法单独测。我的做法是把每个特征拆成独立的类，每个类负责一个特征的提取逻辑。

from abc import ABC, abstractmethod class BaseFeature(ABC): @abstractmethod def extract(self, df: pd.DataFrame) -> pd.Series: pass @property @abstractmethod def name(self) -> str: pass class TextLengthFeature(BaseFeature): @property def name(self): return "text_length" def extract(self, df): return df["text"].str.len() class WordCountFeature(BaseFeature): @property def name(self): return "word_count" def extract(self, df): return df["text"].str.split().str.len() class FeaturePipeline: def __init__(self, features: list[BaseFeature]): self.features = features def transform(self, df): for feature in self.features: df[feature.name] = feature.extract(df) return df

这样设计的好处是：每个特征可以独立测试、独立替换、独立优化。如果某个特征效果不好，直接把它从pipeline里去掉就行，不影响其他特征。

3.3 数据版本管理：DVC的实战用法

数据版本管理是AI工程里一个绕不开的话题。代码可以用Git管理，但数据文件通常很大，不适合直接放进Git仓库。DVC（Data Version Control）就是解决这个问题的工具。

DVC的工作原理是：数据文件本身存在本地或者远程存储（比如S3、OSS）上，Git仓库里只存一个很小的.dvc文件，记录数据的哈希值和存储路径。当你切换Git分支的时候，DVC会自动帮你切换到对应版本的数据。

# 初始化DVC dvc init # 添加数据文件 dvc add data/training_data.csv # 这会生成 data/training_data.csv.dvc 文件 # 把 .dvc 文件加入Git git add data/training_data.csv.dvc data/.gitignore git commit -m "Add training data v1" # 配置远程存储 dvc remote add -d myremote s3://my-bucket/dvc-store # 推送数据到远程 dvc push

提示：DVC的.dvc文件里记录的是数据的哈希值，所以即使数据文件很大，Git仓库也不会变大。切换分支的时候用dvc checkout就能拿到对应版本的数据。

我实际用下来，DVC最大的价值不是"版本管理"本身，而是它强制你把数据和代码的对应关系理清楚。每次实验用的哪版数据、哪版代码，都能追溯。这在团队协作的时候尤其重要。

4. 模型训练与实验管理的工程化实践

4.1 实验追踪：MLflow还是WandB

实验追踪工具我用过MLflow和WandB，各有优劣。MLflow是开源的，可以自己部署，数据完全掌握在自己手里；WandB是SaaS服务，界面更漂亮，协作功能更强，但数据要上传到他们的服务器。

如果你的团队对数据安全要求高，或者你想完全控制实验数据，选MLflow。如果你更看重易用性和协作体验，选WandB。我个人的选择是MLflow，因为我不太喜欢把实验数据传到第三方服务器上。

MLflow的核心概念很简单：一个experiment下面有多个run，每个run记录一组参数、指标和artifact。

import mlflow mlflow.set_experiment("text-classification") with mlflow.start_run(run_name="bert-base-lr2e5"): mlflow.log_params({ "model": "bert-base-chinese", "learning_rate": 2e-5, "batch_size": 32, "epochs": 5, }) for epoch in range(5): train_loss = train_one_epoch(model, train_loader) val_acc = evaluate(model, val_loader) mlflow.log_metrics({ "train_loss": train_loss, "val_acc": val_acc, }, step=epoch) mlflow.pytorch.log_model(model, "model")

这样每次实验的参数、指标、模型文件都会被记录下来，后面要对比不同实验的效果，直接在MLflow的UI里看就行。

4.2 训练脚本的配置化设计

训练脚本最容易犯的错误是把超参数硬编码在代码里。改一个学习率要改代码、重新提交、重新跑，效率极低。我的做法是用配置文件管理所有超参数，代码只负责逻辑。

# config/train_config.yaml model: name: bert-base-chinese num_labels: 3 dropout: 0.1 training: learning_rate: 2e-5 batch_size: 32 epochs: 5 warmup_ratio: 0.1 weight_decay: 0.01 data: train_path: data/train.csv val_path: data/val.csv max_length: 128 output: dir: outputs/bert-base-lr2e5 save_steps: 500

然后用一个配置加载器把YAML转成Python对象：

import yaml from dataclasses import dataclass @dataclass class TrainConfig: model_name: str learning_rate: float batch_size: int epochs: int # ... def load_config(path: str) -> TrainConfig: with open(path) as f: raw = yaml.safe_load(f) return TrainConfig( model_name=raw["model"]["name"], learning_rate=raw["training"]["learning_rate"], batch_size=raw["training"]["batch_size"], epochs=raw["training"]["epochs"], )

这样改超参数只需要改YAML文件，不用动代码。而且配置文件可以版本管理，每次实验用的什么配置一目了然。

4.3 模型检查点管理的坑

模型检查点管理有一个很容易被忽略的问题：保存模型的时候，你保存的到底是什么？

很多人只保存了模型的state_dict，但推理的时候需要的不只是权重，还有tokenizer、配置、标签映射等等。如果这些没有一起保存，加载模型的时候就会出问题。

我的做法是保存一个完整的"模型包"，包含所有推理需要的东西：

import os import json import torch def save_model_package(model, tokenizer, label_map, output_dir): os.makedirs(output_dir, exist_ok=True) # 保存模型权重 torch.save(model.state_dict(), os.path.join(output_dir, "model.pt")) # 保存tokenizer tokenizer.save_pretrained(output_dir) # 保存标签映射 with open(os.path.join(output_dir, "label_map.json"), "w") as f: json.dump(label_map, f) # 保存模型配置 config = { "model_name": model.config._name_or_path, "num_labels": model.config.num_labels, "max_length": 128, } with open(os.path.join(output_dir, "config.json"), "w") as f: json.dump(config, f) logger.info(f"Model package saved to {output_dir}")

这样加载的时候，一个目录就能恢复所有需要的东西，不会出现"忘了保存tokenizer"这种低级错误。

5. 模型部署与服务化的关键决策

5.1 推理服务的框架选型：FastAPI vs TorchServe vs Triton

推理服务的框架选择，取决于你的具体场景。我列一个对比表格：

我大多数时候选FastAPI，因为灵活。你可以完全控制请求的处理逻辑，想加什么预处理、后处理都行。性能方面，对于QPS在几百以内的场景，FastAPI完全够用。如果QPS上千，那就需要考虑Triton或者自己写C++扩展了。

一个典型的FastAPI推理服务：

from fastapi import FastAPI, HTTPException from pydantic import BaseModel import torch app = FastAPI() class PredictRequest(BaseModel): text: str class PredictResponse(BaseModel): label: str confidence: float model = None tokenizer = None @app.on_event("startup") def load_model(): global model, tokenizer model = torch.load("model/model.pt", map_location="cpu") model.eval() tokenizer = AutoTokenizer.from_pretrained("model/") @app.post("/predict", response_model=PredictResponse) def predict(request: PredictRequest): if not request.text.strip(): raise HTTPException(status_code=400, detail="Text cannot be empty") inputs = tokenizer( request.text, max_length=128, truncation=True, return_tensors="pt", ) with torch.no_grad(): outputs = model(**inputs) probs = torch.softmax(outputs.logits, dim=-1) pred = torch.argmax(probs, dim=-1).item() confidence = probs[0][pred].item() return PredictResponse( label=LABEL_MAP[pred], confidence=confidence, )

注意：@app.on_event("startup")在较新版本的FastAPI里已经被弃用了，推荐用lifespan上下文管理器。但很多教程还在用旧写法，你知道就行。

5.2 批处理与动态批处理的实现

推理服务性能优化的核心是批处理。单个请求推理一次，GPU利用率极低；把多个请求攒成一批一起推理，吞吐量能提升几倍甚至几十倍。

最简单的批处理是"固定窗口"：每隔100毫秒收集一次请求，凑成一批推理。但这种方式在低流量时会有延迟，高流量时批次可能太大。

更好的方式是"动态批处理"：设置一个最大批次大小和最大等待时间，当批次满了或者等待时间到了就触发推理。

import asyncio from collections import deque class DynamicBatcher: def __init__(self, max_batch_size=32, max_wait_ms=50): self.max_batch_size = max_batch_size self.max_wait_ms = max_wait_ms self.queue = deque() self.lock = asyncio.Lock() async def add_request(self, inputs): future = asyncio.Future() async with self.lock: self.queue.append((inputs, future)) if len(self.queue) >= self.max_batch_size: await self._process_batch() return await future async def _process_batch(self): batch = list(self.queue) self.queue.clear() inputs = [item[0] for item in batch] futures = [item[1] for item in batch] results = model_inference(inputs) for future, result in zip(futures, results): future.set_result(result)

这段代码的核心思路是：请求进来先放进队列，如果队列满了就立即处理；如果没满，等一个超时时间再处理。这样既能保证高流量时的吞吐量，又能保证低流量时的延迟。

5.3 模型版本管理与灰度发布

模型上线不是"替换文件"那么简单。你需要考虑：新模型和旧模型的兼容性、灰度发布的策略、回滚的机制。

我的做法是用一个模型注册表来管理所有模型版本：

class ModelRegistry: def __init__(self, base_dir: str): self.base_dir = Path(base_dir) self.models = {} self.active_version = None def register(self, version: str, model_path: str): self.models[version] = { "path": model_path, "loaded_at": datetime.now(), "status": "inactive", } def activate(self, version: str): if version not in self.models: raise ValueError(f"Version {version} not registered") if self.active_version: self.models[self.active_version]["status"] = "inactive" self.models[version]["status"] = "active" self.active_version = version logger.info(f"Activated model version: {version}") def get_active(self): if not self.active_version: raise RuntimeError("No active model") return self.models[self.active_version]

灰度发布的时候，可以按流量比例分流：90%的请求走旧模型，10%走新模型。观察一段时间，如果新模型表现正常，再逐步提高比例。如果发现问题，立即把流量切回旧模型。

6. 监控、运维与持续迭代——上线只是开始

6.1 模型性能监控的三个层次

模型上线之后，你需要监控三个层次的东西：

第一层：系统指标。CPU、内存、GPU利用率、请求延迟、QPS、错误率。这些是传统后端服务也需要监控的，工具用Prometheus + Grafana就行。

第二层：模型指标。推理延迟分布、批次大小分布、输入长度分布、输出标签分布。这些指标能帮你发现模型层面的问题。比如输入长度分布突然变了，可能是上游数据源出了问题；输出标签分布突然偏向某一类，可能是模型退化或者数据漂移。

第三层：业务指标。准确率、召回率、F1值、用户点击率、转化率。这些指标需要结合业务反馈来获取，通常有一定的延迟。

from prometheus_client import Histogram, Counter inference_latency = Histogram( "model_inference_latency_seconds", "Model inference latency", buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 5.0], ) prediction_counter = Counter( "model_predictions_total", "Total predictions", ["label"], ) def predict_with_metrics(text): with inference_latency.time(): result = model.predict(text) prediction_counter.labels(label=result.label).inc() return result

6.2 数据漂移检测的实操方法

数据漂移是模型性能下降的最常见原因。检测数据漂移的方法有很多，我常用的是PSI（Population Stability Index）和KL散度。

PSI的计算方式是：把训练数据和线上数据分别分桶，计算每个桶里样本比例的差异，然后加权求和。PSI小于0.1说明分布稳定，0.1到0.25说明有轻微漂移，大于0.25说明有显著漂移。

import numpy as np def calculate_psi(expected, actual, buckets=10): breakpoints = np.linspace(0, 100, buckets + 1) expected_percents = np.percentile(expected, breakpoints) actual_percents = np.percentile(actual, breakpoints) expected_counts = np.histogram(expected, bins=expected_percents)[0] / len(expected) actual_counts = np.histogram(actual, bins=expected_percents)[0] / len(actual) expected_counts = np.where(expected_counts == 0, 0.0001, expected_counts) actual_counts = np.where(actual_counts == 0, 0.0001, actual_counts) psi_values = (actual_counts - expected_counts) * np.log(actual_counts / expected_counts) return np.sum(psi_values)

提示：PSI对分桶方式比较敏感，分桶数量不同，结果可能差异很大。我的经验是分10到20个桶比较合适，太少不敏感，太多噪声大。

6.3 模型重训练的触发策略

模型什么时候需要重训练？我的策略是"定期 + 触发"结合：

• 定期重训练：每周或每月跑一次全量重训练，用最新的数据更新模型。
• 触发式重训练：当监控指标超过阈值时触发。比如PSI大于0.25、准确率下降超过5%、用户投诉增加等。

重训练不是简单地"用新数据跑一遍"，你需要考虑：新数据的时间范围、数据量、标注质量、和旧数据的混合比例。我的经验是新数据占比30%到50%比较合适，太少效果不明显，太多容易导致模型遗忘旧知识。

def should_retrain(metrics: dict) -> bool: if metrics["psi"] > 0.25: logger.warning(f"PSI {metrics['psi']} exceeds threshold, retrain needed") return True if metrics["accuracy_drop"] > 0.05: logger.warning(f"Accuracy dropped by {metrics['accuracy_drop']}, retrain needed") return True if metrics["days_since_last_train"] > 30: logger.info("Scheduled retraining triggered") return True return False

7. 一些让我少走弯路的经验总结

7.1 先跑通再优化，别一开始就追求完美架构

我见过太多人（包括我自己）在项目初期就花大量时间设计"完美的架构"，结果代码写了一堆抽象层，实际功能一个没跑通。AI工程的特点是变化快，你今天设计的架构，可能下周就因为需求变化要推翻重来。

我的建议是：先用最直接的方式把功能跑通，哪怕代码丑一点、硬编码多一点。等跑通了，你才知道真正的瓶颈在哪里，才知道哪些地方需要抽象、哪些地方不需要。过早优化是万恶之源，这句话在AI工程里尤其适用。

7.2 日志和监控要提前埋，不要等出问题再加

日志和监控这种东西，平时看起来没什么用，但一旦出问题，它就是你的救命稻草。我的习惯是在写第一行业务代码的时候就把日志埋好，每个关键步骤都打日志，包括输入、输出、耗时、异常。

监控也是同理。不要等上线了才想起来加监控，那时候你已经错过了很多关键信息。在开发阶段就把Prometheus的指标埋好，上线之后直接就能看到数据。

7.3 测试不是可选项，是必选项

AI工程的测试比传统软件测试更难，因为模型的行为是不确定的。但难不代表可以不做。我的做法是分三层测试：

• 单元测试：测试每个函数、每个类的行为。比如数据预处理函数，输入一组数据，输出是否符合预期。
• 集成测试：测试整个pipeline的行为。从原始数据到最终预测，端到端跑一遍，确保没有断裂。
• 回归测试：每次模型更新后，用一组固定的测试数据跑一遍，确保新模型没有在旧数据上表现变差。

def test_preprocessor_handles_nan(): df = pd.DataFrame({"feature": [1.0, np.nan, 3.0]}) preprocessor = FeaturePreprocessor({"feature_column": "feature"}) preprocessor.fit(df) result = preprocessor.transform(df) assert not result["feature_scaled"].isna().any() def test_model_output_shape(): model = load_model("model/") inputs = tokenizer("test", return_tensors="pt") outputs = model(**inputs) assert outputs.logits.shape == (1, 3)

这些测试看起来很简单，但它们能在你改代码的时候快速告诉你"有没有改坏东西"。没有测试的代码，改起来心里没底。

7.4 文档和注释：写给三个月后的自己

AI工程的项目周期通常比较长，你今天写的代码，可能三个月后才需要改。到时候你大概率已经忘了当时的思路。所以文档和注释非常重要。

我的习惯是：每个模块的入口函数写清楚输入输出、依赖关系、注意事项；每个关键决策点写清楚"为什么这么做"；每个坑写清楚"踩过什么坑、怎么解决的"。这些注释不是写给别人的，是写给三个月后的自己的。

7.5 保持学习，但不要盲目追新

AI领域的新技术、新框架、新模型层出不穷，很容易让人焦虑。我的态度是：保持关注，但不要盲目追新。新技术出来先看看它的核心思路是什么、解决了什么问题、适不适合我的场景。如果确实能解决我的问题，再花时间深入学；如果只是"看起来很美"，那就先放一放。

我见过太多人花大量时间学了一个新框架，结果发现它根本不适合自己的场景，白白浪费了时间。技术是工具，工具是为场景服务的，不要本末倒置。

最后分享一个我自己的小习惯：我会维护一个"踩坑笔记"，每次遇到问题、解决问题之后，都会把问题的现象、原因、解决方案记下来。这个笔记现在已经有几百条了，每次遇到类似的问题，先翻笔记，大部分时候都能找到答案。这个习惯帮我节省了大量重复排查的时间，也让我对AI工程的各个环节有了更系统的理解。