尧图网站开发 
1. 从零手搓AI工程为什么我不建议你直接调包很多人一听到“AI工程”这四个字第一反应就是打开某个云平台拖几个组件调一下API跑通一个Demo然后发个朋友圈说“今天又搞定了一个AI项目”。我刚开始也是这么干的直到有一次线上模型推理延迟突然从80毫秒飙到2秒日志里全是超时而我盯着控制台里那几个黑盒组件完全不知道从哪下手。那一刻我才意识到只会调包的人在AI工程这条路上是走不远的。ai-engineering-from-scratch这个标题核心不是教你如何快速拼凑一个能跑的AI应用而是带你从最底层开始把AI工程里那些被封装层藏起来的关键环节一个一个拆开亲手实现一遍。它适合那些已经会用Python、了解一点机器学习基础但每次遇到性能瓶颈、部署故障、数据管道断裂就束手无策的开发者。说白了就是让你从“会用工具的人”变成“懂工具为什么这样设计的人”。我打算按一个真实项目的推进节奏来写先搞清楚AI工程到底包含哪些环节再逐个环节手搓最小可用版本最后把它们串成一条完整的流水线。整个过程不依赖任何重型框架能用标准库就用标准库必须用第三方库的地方我会解释为什么选它、有没有替代方案、坑在哪里。你跟着走一遍至少能获得三个能力第一看到任何AI工程架构图都能说出每个模块在干什么第二遇到线上问题能快速定位是数据、模型还是服务层的问题第三能根据自己的业务场景做合理的技术选型而不是别人用什么你就用什么。2. 先把AI工程的地基挖清楚数据管道不是“读个CSV”那么简单2.1 数据管道的三个隐藏层次大部分教程讲数据管道就是pd.read_csv()然后train_test_split()完事。但在真实工程里数据管道至少分三层采集层、清洗层、特征层。采集层负责从各种源头把数据弄进来可能是数据库、日志文件、消息队列也可能是第三方接口清洗层处理缺失值、异常值、格式统一、去重特征层做归一化、编码、特征交叉、时序窗口统计。为什么要把这三层分开因为它们的变更频率完全不同。采集层的源头可能每天新增一个清洗规则可能每周调整一次特征工程可能每个模型版本都要改。如果全写在一个脚本里改一处就要全量重跑调试成本极高。我手搓的第一个版本就是把所有逻辑塞进一个process.py结果每次调特征都要重新解析原始日志一次跑二十分钟一天下来光等脚本就跑掉了半天。正确的做法是每一层输出独立的中间结果用文件系统或对象存储做持久化。采集层输出原始快照清洗层输出标准化后的数据集特征层输出模型可直接消费的特征矩阵。这样调特征的时候只需要从清洗层的结果开始跑时间从二十分钟降到两分钟。2.2 手搓一个带背压的采集器采集层最容易忽略的问题是背压。什么叫背压简单说就是下游处理不过来的时候上游要能感知到并降低速度而不是无限制地往内存里塞数据。我见过一个项目采集器从消息队列里疯狂拉数据清洗层处理速度只有采集速度的三分之一结果内存直接爆掉进程被系统杀掉。手搓一个带背压的采集器核心是维护一个固定大小的缓冲区当缓冲区满了就阻塞采集动作。用Python实现的话可以用queue.Queue(maxsizeN)来做缓冲采集线程往队列里put清洗线程从队列里get。当队列满时put会自动阻塞天然实现了背压。import queue import threading import time raw_queue queue.Queue(maxsize1000) def collector(source): for record in source: raw_queue.put(record) # 队列满时自动阻塞 raw_queue.put(None) # 结束信号 def cleaner(): while True: record raw_queue.get() if record is None: break process(record) raw_queue.task_done()这个模式看起来简单但有几个细节要注意。第一结束信号None必须在所有数据之后放入否则清洗线程可能提前退出。第二如果采集过程中出现异常要确保结束信号仍然被发送否则清洗线程会永远阻塞。第三队列大小需要根据内存和处理速度做权衡太小会导致采集频繁阻塞太大会占用过多内存。我一般从1000开始试观察内存占用和吞吐量再调整。2.3 清洗层的幂等性设计清洗层最怕的是什么是重跑。因为网络抖动、进程崩溃、人为误操作清洗任务可能需要重跑。如果清洗逻辑不是幂等的重跑就会产生重复数据或者覆盖掉正确结果。幂等的意思是同样的输入跑一次和跑十次输出结果完全一样。实现幂等清洗的关键是给每条记录一个唯一标识清洗后的结果按标识存储重跑时先检查是否已经处理过。唯一标识可以用原始数据的哈希值也可以用业务主键加时间戳。我通常会在清洗层维护一个processed_ids的集合持久化到磁盘每次启动时加载。处理新记录前先查这个集合如果已经处理过就跳过。import hashlib import json import os def get_record_id(record): content json.dumps(record, sort_keysTrue) return hashlib.md5(content.encode()).hexdigest() def load_processed_ids(path): if os.path.exists(path): with open(path) as f: return set(json.load(f)) return set() def save_processed_ids(ids, path): with open(path, w) as f: json.dump(list(ids), f)这个方案有个代价需要存储所有已处理记录的ID数据量大了之后内存和磁盘都会吃紧。折中方案是只存储最近一段时间比如最近七天的ID更早的数据认为不会再重跑。或者用布隆过滤器做概率性判断牺牲一点准确性换取空间效率。具体选哪种取决于你的数据量和重跑频率。3. 模型训练之外的事把实验管理当回事3.1 为什么你的实验总是无法复现我敢打赌每个做AI工程的人都遇到过这种情况上周跑出一个效果很好的模型这周想复现一下结果怎么跑都跑不出那个指标了。代码没改数据没改但结果就是不一样。原因通常藏在三个地方随机种子、环境依赖、数据版本。随机种子不只是random.seed(42)和np.random.seed(42)就完事了。深度学习框架有自己的随机源CUDA也有自己的随机行为。如果你用了GPU还需要设置torch.manual_seed(42)、torch.cuda.manual_seed_all(42)并且在某些操作上启用确定性算法。但要注意启用确定性算法可能会降低性能所以只在需要精确复现的实验里开。环境依赖的问题更隐蔽。你本地是numpy 1.24服务器上是numpy 1.21某些函数的默认行为可能不一样。更别说CUDA版本、cuDNN版本、甚至不同型号的GPU都可能导致数值差异。解决办法是把环境完整记录下来用pip freeze导出依赖列表用容器镜像固定系统级依赖。数据版本是最容易被忽略的。很多人觉得数据放在那里又不会变但实际上数据可能被清洗脚本重新跑过、被新采集的数据追加过、被人工修正过。没有数据版本管理复现就是碰运气。最简单的做法是给每次数据变更打一个版本号训练时记录用了哪个版本。3.2 手搓一个轻量实验追踪器市面上的实验追踪工具很多但我想让你先手搓一个最小版本理解它到底在追踪什么。核心就三件事记录参数、记录指标、关联产物。import json import os import time import uuid class ExperimentTracker: def __init__(self, base_direxperiments): self.base_dir base_dir self.exp_id str(uuid.uuid4())[:8] self.exp_dir os.path.join(base_dir, self.exp_id) os.makedirs(self.exp_dir, exist_okTrue) self.metrics [] self.params {} def log_params(self, params): self.params params with open(os.path.join(self.exp_dir, params.json), w) as f: json.dump(params, f, indent2) def log_metric(self, name, value, stepNone): self.metrics.append({ name: name, value: value, step: step, timestamp: time.time() }) with open(os.path.join(self.exp_dir, metrics.json), w) as f: json.dump(self.metrics, f, indent2) def log_artifact(self, filepath, artifact_name): import shutil dest os.path.join(self.exp_dir, artifact_name) shutil.copy(filepath, dest) return dest这个追踪器把每次实验的参数、指标、产物都放在一个独立目录里用UUID做标识。参数在实验开始时记录一次指标可以多次记录比如每个epoch记录一次loss产物在训练结束后保存。这样你随时可以回到任何一个实验看到当时用了什么参数、指标怎么变化的、模型文件在哪里。实际使用中我建议在参数里额外记录几个东西代码的git commit hash、数据版本号、开始时间、机器标识。这些信息在排查“为什么这次和上次不一样”的时候非常有用。代码hash可以用subprocess调git命令获取数据版本号从数据管道的输出里读机器标识用platform.node()。3.3 指标记录的粒度选择指标记录太粗会丢失信息太细会拖慢训练。我的经验是分三层记录全局指标、阶段指标、批次指标。全局指标是整个训练结束后才有的比如最终测试集准确率阶段指标是每个epoch或每个验证周期记录的比如验证集loss批次指标是每个batch记录的比如训练loss。批次指标的数据量很大一个epoch有几千个batch的话记录所有批次指标会产生大量数据。我的做法是批次指标只保留最近N个或者做滑动平均后再记录。阶段指标全量保留因为一个epoch才一条数据量可控。全局指标单独存一个文件方便快速查看。另外指标的名字要有统一的命名规范。我见过一个项目同一个人在不同脚本里用了val_acc、val_accuracy、validation_accuracy三种写法最后汇总的时候对不上。建议在项目开始就定好命名规范比如统一用{split}_{metric}的格式train_loss、val_loss、test_accuracy。4. 模型服务化从pickle文件到可用的推理接口4.1 模型加载的性能陷阱训练完的模型保存成pickle文件服务启动时加载看起来很简单。但这里有个性能陷阱如果模型文件很大比如几个GB加载时间可能长达几十秒甚至几分钟。在这段时间里服务是不可用的。如果服务重启频繁或者需要动态加载多个模型这个问题会非常致命。我踩过的坑是这样的一个推荐模型文件有3GB服务每次重启要加载两分钟。后来做AB测试需要同时加载两个模型启动时间变成四分钟。再后来做多路召回要加载五个模型启动时间直接超过十分钟运维那边已经开始报警了。解决办法有几个方向。第一用更高效的序列化格式比如onnx或torchscript加载速度比pickle快很多。第二做模型预热服务启动后在后台异步加载模型加载完成前返回一个默认结果或降级结果。第三用内存映射文件让操作系统按需加载模型参数而不是一次性全部读入内存。import mmap import pickle def load_model_mmap(path): with open(path, rb) as f: mm mmap.mmap(f.fileno(), 0, accessmmap.ACCESS_READ) model pickle.loads(mm) return model内存映射的好处是启动时只建立映射关系不实际读取数据。真正用到某个参数时操作系统才从磁盘加载对应的页。对于大模型来说启动时间可以从分钟级降到秒级。但要注意内存映射的文件不能被修改否则会导致未定义行为。4.2 手搓一个带批处理的推理服务推理服务的核心指标是吞吐量和延迟。单条推理的吞吐量很低因为每次都要做一遍前向传播的计算图调度。批处理可以把多条请求合并成一批一次性做前向传播大幅提升吞吐量。但批处理会增加延迟因为要等凑够一批才能处理。手搓一个带批处理的推理服务核心是维护一个请求队列和一个批处理循环。请求到达时先放入队列批处理循环从队列里取出最多N条请求组成一个batch送给模型然后把结果分发给对应的请求。import queue import threading import numpy as np class BatchInferenceServer: def __init__(self, model, max_batch_size32, max_wait_ms10): self.model model self.max_batch_size max_batch_size self.max_wait_ms max_wait_ms self.request_queue queue.Queue() self.results {} self.lock threading.Lock() self._start_worker() def _start_worker(self): def worker(): while True: batch [] try: # 阻塞等待第一条请求 item self.request_queue.get(timeout1) batch.append(item) # 在max_wait_ms内尽量多收集请求 deadline time.time() self.max_wait_ms / 1000 while len(batch) self.max_batch_size: remaining deadline - time.time() if remaining 0: break try: item self.request_queue.get(timeoutremaining) batch.append(item) except queue.Empty: break self._process_batch(batch) except queue.Empty: continue t threading.Thread(targetworker, daemonTrue) t.start() def _process_batch(self, batch): inputs np.stack([item[input] for item in batch]) outputs self.model(inputs) for item, output in zip(batch, outputs): with self.lock: self.results[item[id]] output def predict(self, input_data): req_id str(uuid.uuid4()) self.request_queue.put({id: req_id, input: input_data}) while True: with self.lock: if req_id in self.results: return self.results.pop(req_id) time.sleep(0.001)这个实现里有两个关键参数max_batch_size和max_wait_ms。前者决定一批最多处理多少条后者决定最多等多久。这两个参数需要根据实际场景调优。如果请求量大且延迟要求不严格可以增大max_batch_size和max_wait_ms如果请求量小且延迟敏感就减小这两个值。4.3 服务降级与熔断推理服务不可能永远稳定。模型可能因为输入异常而报错依赖的特征服务可能超时GPU可能被其他任务占满。如果没有降级和熔断机制一个小的异常就可能拖垮整个服务。降级的思路是当主逻辑失败时返回一个兜底结果。比如推荐服务如果模型推理失败可以返回热门榜单如果特征服务超时可以用默认特征代替。兜底结果的质量肯定不如正常结果但至少服务是可用的。熔断的思路是当某个依赖的失败率超过阈值时暂时切断对该依赖的调用直接走降级逻辑避免大量请求堆积导致雪崩。熔断器一般有三个状态关闭正常调用、打开直接降级、半开允许少量请求试探是否恢复。class CircuitBreaker: def __init__(self, failure_threshold5, recovery_timeout30): self.failure_threshold failure_threshold self.recovery_timeout recovery_timeout self.failure_count 0 self.state closed self.last_failure_time None def call(self, func, *args, **kwargs): if self.state open: if time.time() - self.last_failure_time self.recovery_timeout: self.state half-open else: raise Exception(Circuit is open) try: result func(*args, **kwargs) if self.state half-open: self.state closed self.failure_count 0 return result except Exception as e: self.failure_count 1 self.last_failure_time time.time() if self.failure_count self.failure_threshold: self.state open raise e这个熔断器实现比较粗糙但足以说明原理。实际使用中失败计数应该按时间窗口统计而不是简单的累加。比如统计最近60秒内的失败率超过50%就熔断。另外半开状态下应该只允许一个请求通过成功后再完全恢复。5. 监控与排错线上问题不会给你慢慢调试的机会5.1 推理服务的四个黄金指标线上推理服务最需要关注的四个指标是延迟、吞吐量、错误率、资源利用率。延迟分P50、P95、P99P99延迟高说明有少量请求体验很差可能是长尾问题。吞吐量是每秒处理的请求数反映服务容量。错误率包括业务错误和系统错误业务错误比如输入格式不对系统错误比如超时、内存溢出。资源利用率包括CPU、GPU、内存、网络用来判断是否需要扩容。这四个指标要一起看单独看一个容易误判。比如延迟升高但吞吐量没变可能是某个依赖变慢了吞吐量下降但延迟没变可能是请求量本身减少了错误率升高但延迟正常可能是上游传入了异常数据。我习惯在服务启动时就埋好这些指标的采集点用prometheus_client暴露出来然后用Grafana做面板。如果不想引入这么重的方案至少要把指标写到日志里用grep和awk做简单分析。from prometheus_client import Histogram, Counter, Gauge import time REQUEST_LATENCY Histogram(request_latency_seconds, Request latency) REQUEST_COUNT Counter(request_count, Total requests, [status]) INFLIGHT_REQUESTS Gauge(inflight_requests, In-flight requests) def handle_request(input_data): INFLIGHT_REQUESTS.inc() start time.time() try: result model.predict(input_data) REQUEST_COUNT.labels(statussuccess).inc() return result except Exception as e: REQUEST_COUNT.labels(statuserror).inc() raise e finally: REQUEST_LATENCY.observe(time.time() - start) INFLIGHT_REQUESTS.dec()5.2 一次线上延迟飙升的完整排查过程说一个我亲身经历的案例。某天下午推荐服务的P99延迟从120毫秒突然涨到800毫秒持续了大约十分钟后恢复。没有报警是用户反馈“推荐刷不出来”才发现的。第一步看监控面板。延迟飙升的时间段内CPU利用率从40%涨到90%GPU利用率从60%涨到95%内存正常网络正常。初步判断是计算资源不够。第二步看请求量。同一时间段内请求量并没有明显增加QPS稳定在2000左右。排除流量突增。第三步看错误日志。没有大量报错只有零星几条超时日志。排除代码异常。第四步看依赖服务。特征服务的延迟从20毫秒涨到200毫秒但特征服务本身没有报警。深入查特征服务的日志发现它在同一时间段内做了一次全量缓存刷新刷新期间所有请求都穿透到数据库数据库响应变慢导致特征服务变慢进而拖慢了推理服务。第五步定位根因。特征服务的缓存刷新策略是每天下午三点全量刷新刷新时先清空缓存再重新加载。清空到加载完成之间有大约五分钟的窗口期所有请求都打到数据库。之前数据库压力小没出问题那天数据库上正好有另一个任务在跑叠加起来就扛不住了。修复方案有两个第一缓存刷新改成增量刷新不清空旧缓存新数据加载完成后再原子替换第二给特征服务加熔断数据库响应慢时直接返回默认特征避免拖垮上游。两个方案都上线后再没出现过类似问题。这个案例的教训是监控不能只看自己服务的指标还要看依赖服务的指标。依赖服务的微小抖动可能在上游被放大成严重故障。5.3 日志里应该记录什么日志是排错的第一手资料但很多人的日志要么太少只有“开始处理”“处理完成”要么太多每个变量都打一遍。我的经验是记录五类信息请求标识、关键输入、关键输出、耗时、异常堆栈。请求标识用于串联一次请求在所有服务里的日志。可以用UUID也可以用上游传下来的trace_id。关键输入不是把整个输入都打出来可能很大而是打摘要比如输入的shape、关键字段的值、哈希值。关键输出同理。耗时记录每个阶段的耗时比如特征获取耗时、模型推理耗时、后处理耗时。异常堆栈在出错时打印但要注意脱敏不要把用户敏感信息打出来。import logging import uuid import time logger logging.getLogger(__name__) def handle_request(input_data): req_id str(uuid.uuid4())[:8] logger.info(freq_id{req_id} stagestart input_shape{input_data.shape}) t0 time.time() features get_features(input_data) t1 time.time() logger.info(freq_id{req_id} stagefeatures cost{t1-t0:.3f}s) result model.predict(features) t2 time.time() logger.info(freq_id{req_id} stageinference cost{t2-t1:.3f}s) return result日志格式建议用结构化日志比如JSON格式方便后续用工具做聚合分析。如果日志量很大要注意采样不是所有请求都打全量日志可以按比例采样或者只对慢请求打详细日志。6. 把碎片串起来一条最小可用的AI工程流水线6.1 流水线的目录结构设计把前面几部分串起来需要一个清晰的目录结构。我习惯按功能划分目录而不是按文件类型。比如data/放数据相关脚本model/放模型相关脚本service/放服务相关脚本monitor/放监控相关脚本。每个目录下再按阶段细分。ai-pipeline/ ├── data/ │ ├── collector.py │ ├── cleaner.py │ └── featurizer.py ├── model/ │ ├── trainer.py │ ├── evaluator.py │ └── exporter.py ├── service/ │ ├── server.py │ ├── batch_inference.py │ └── circuit_breaker.py ├── monitor/ │ ├── metrics.py │ └── dashboard.py ├── experiments/ │ └── {exp_id}/ │ ├── params.json │ ├── metrics.json │ └── model.pkl └── configs/ ├── data.yaml ├── model.yaml └── service.yaml这个结构的好处是每个模块职责清晰可以独立开发和测试。数据管道可以单独跑不依赖模型模型训练可以单独跑不依赖服务服务可以单独跑不依赖训练。模块之间通过文件和配置解耦。配置用YAML文件管理不同环境开发、测试、生产用不同的配置文件。配置里放路径、参数、开关不要硬编码在代码里。我见过一个项目把数据库密码写在代码里换环境的时候忘了改连到了生产库差点出大事。6.2 从数据到服务的完整流程完整流程分六步采集、清洗、特征、训练、导出、服务。每一步的输出是下一步的输入中间结果持久化到磁盘。采集步骤从数据源拉取原始数据存到data/raw/目录按日期分文件夹。清洗步骤读取原始数据做标准化处理存到data/clean/目录。特征步骤读取清洗后的数据生成特征矩阵存到data/features/目录。训练步骤读取特征矩阵训练模型把模型文件和实验记录存到experiments/目录。导出步骤把模型转换成服务需要的格式存到service/models/目录。服务步骤加载模型启动推理接口。每一步都可以单独重跑只要保证输入数据存在。比如特征工程改了只需要从特征步骤开始重跑不需要重新采集和清洗。训练改了只需要从训练步骤开始重跑。这种设计让迭代速度大幅提升。# 完整流程 python data/collector.py --date 2024-01-01 python data/cleaner.py --date 2024-01-01 python data/featurizer.py --date 2024-01-01 python model/trainer.py --config configs/model.yaml python model/exporter.py --exp_id latest python service/server.py --config configs/service.yaml6.3 我踩过的三个集成坑第一个坑是路径问题。开发的时候用相对路径跑得好好的部署到服务器上用绝对路径全挂了。解决办法是统一用相对于项目根目录的路径在代码里通过os.path.dirname(__file__)定位项目根目录然后拼接路径。第二个坑是依赖版本冲突。数据管道用pandas 1.5模型训练用pandas 2.0两个模块跑在同一个环境里就冲突了。解决办法是用虚拟环境隔离每个模块有自己的requirements.txt或者用容器把每个模块打包成独立的镜像。第三个坑是时区问题。数据采集用UTC时间模型训练用本地时间服务日志用另一个时区排查问题的时候时间对不上。解决办法是统一用UTC时间只在展示层做时区转换。所有日志、指标、数据文件的时间戳都用UTC避免混乱。7. 一些让我少走弯路的经验关于数据管道我的建议是宁可多存中间结果也不要为了省磁盘空间把多个步骤合并。磁盘便宜时间贵。中间结果保留至少最近一个月的方便回溯和调试。关于实验管理不要等到实验多了才想起来做追踪。从第一个实验开始就用追踪器哪怕只是一个简单的JSON文件。后期补记录的成本远高于前期养习惯的成本。关于模型服务批处理是提升吞吐量最有效的手段但批处理大小和等待时间需要根据实际流量调优。流量大的时候增大批处理流量小的时候减小等待时间。不要一套参数用到底。关于监控不要只监控自己服务的指标。上下游的指标同样重要因为故障往往是从上下游传导过来的。至少要把直接依赖的服务的核心指标纳入监控面板。关于排错养成先看监控、再看日志、最后看代码的习惯。大部分问题在监控和日志层面就能定位不需要读代码。直接读代码容易陷入细节忽略全局。最后说一个心态上的体会。AI工程是一个系统工程涉及数据、模型、服务、运维多个领域没有人能一开始就全部精通。我的做法是每个领域都了解个大概然后选一两个领域深入。遇到问题的时候知道大概是什么方向的问题知道找谁帮忙知道查什么资料这比什么都懂一点但什么都不精要有效得多。 
1. 从零手搓AI工程为什么我不建议你直接调包很多人一听到“AI工程”这四个字第一反应就是打开某个云平台拖几个组件调一下API跑通一个Demo然后发个朋友圈说“今天又搞定了一个AI项目”。我刚开始也是这么干的直到有一次线上模型推理延迟突然从80毫秒飙到2秒日志里全是超时而我盯着控制台里那几个黑盒组件完全不知道从哪下手。那一刻我才意识到只会调包的人在AI工程这条路上是走不远的。ai-engineering-from-scratch这个标题核心不是教你如何快速拼凑一个能跑的AI应用而是带你从最底层开始把AI工程里那些被封装层藏起来的关键环节一个一个拆开亲手实现一遍。它适合那些已经会用Python、了解一点机器学习基础但每次遇到性能瓶颈、部署故障、数据管道断裂就束手无策的开发者。说白了就是让你从“会用工具的人”变成“懂工具为什么这样设计的人”。我打算按一个真实项目的推进节奏来写先搞清楚AI工程到底包含哪些环节再逐个环节手搓最小可用版本最后把它们串成一条完整的流水线。整个过程不依赖任何重型框架能用标准库就用标准库必须用第三方库的地方我会解释为什么选它、有没有替代方案、坑在哪里。你跟着走一遍至少能获得三个能力第一看到任何AI工程架构图都能说出每个模块在干什么第二遇到线上问题能快速定位是数据、模型还是服务层的问题第三能根据自己的业务场景做合理的技术选型而不是别人用什么你就用什么。2. 先把AI工程的地基挖清楚数据管道不是“读个CSV”那么简单2.1 数据管道的三个隐藏层次大部分教程讲数据管道就是pd.read_csv()然后train_test_split()完事。但在真实工程里数据管道至少分三层采集层、清洗层、特征层。采集层负责从各种源头把数据弄进来可能是数据库、日志文件、消息队列也可能是第三方接口清洗层处理缺失值、异常值、格式统一、去重特征层做归一化、编码、特征交叉、时序窗口统计。为什么要把这三层分开因为它们的变更频率完全不同。采集层的源头可能每天新增一个清洗规则可能每周调整一次特征工程可能每个模型版本都要改。如果全写在一个脚本里改一处就要全量重跑调试成本极高。我手搓的第一个版本就是把所有逻辑塞进一个process.py结果每次调特征都要重新解析原始日志一次跑二十分钟一天下来光等脚本就跑掉了半天。正确的做法是每一层输出独立的中间结果用文件系统或对象存储做持久化。采集层输出原始快照清洗层输出标准化后的数据集特征层输出模型可直接消费的特征矩阵。这样调特征的时候只需要从清洗层的结果开始跑时间从二十分钟降到两分钟。2.2 手搓一个带背压的采集器采集层最容易忽略的问题是背压。什么叫背压简单说就是下游处理不过来的时候上游要能感知到并降低速度而不是无限制地往内存里塞数据。我见过一个项目采集器从消息队列里疯狂拉数据清洗层处理速度只有采集速度的三分之一结果内存直接爆掉进程被系统杀掉。手搓一个带背压的采集器核心是维护一个固定大小的缓冲区当缓冲区满了就阻塞采集动作。用Python实现的话可以用queue.Queue(maxsizeN)来做缓冲采集线程往队列里put清洗线程从队列里get。当队列满时put会自动阻塞天然实现了背压。import queue import threading import time raw_queue queue.Queue(maxsize1000) def collector(source): for record in source: raw_queue.put(record) # 队列满时自动阻塞 raw_queue.put(None) # 结束信号 def cleaner(): while True: record raw_queue.get() if record is None: break process(record) raw_queue.task_done()这个模式看起来简单但有几个细节要注意。第一结束信号None必须在所有数据之后放入否则清洗线程可能提前退出。第二如果采集过程中出现异常要确保结束信号仍然被发送否则清洗线程会永远阻塞。第三队列大小需要根据内存和处理速度做权衡太小会导致采集频繁阻塞太大会占用过多内存。我一般从1000开始试观察内存占用和吞吐量再调整。2.3 清洗层的幂等性设计清洗层最怕的是什么是重跑。因为网络抖动、进程崩溃、人为误操作清洗任务可能需要重跑。如果清洗逻辑不是幂等的重跑就会产生重复数据或者覆盖掉正确结果。幂等的意思是同样的输入跑一次和跑十次输出结果完全一样。实现幂等清洗的关键是给每条记录一个唯一标识清洗后的结果按标识存储重跑时先检查是否已经处理过。唯一标识可以用原始数据的哈希值也可以用业务主键加时间戳。我通常会在清洗层维护一个processed_ids的集合持久化到磁盘每次启动时加载。处理新记录前先查这个集合如果已经处理过就跳过。import hashlib import json import os def get_record_id(record): content json.dumps(record, sort_keysTrue) return hashlib.md5(content.encode()).hexdigest() def load_processed_ids(path): if os.path.exists(path): with open(path) as f: return set(json.load(f)) return set() def save_processed_ids(ids, path): with open(path, w) as f: json.dump(list(ids), f)这个方案有个代价需要存储所有已处理记录的ID数据量大了之后内存和磁盘都会吃紧。折中方案是只存储最近一段时间比如最近七天的ID更早的数据认为不会再重跑。或者用布隆过滤器做概率性判断牺牲一点准确性换取空间效率。具体选哪种取决于你的数据量和重跑频率。3. 模型训练之外的事把实验管理当回事3.1 为什么你的实验总是无法复现我敢打赌每个做AI工程的人都遇到过这种情况上周跑出一个效果很好的模型这周想复现一下结果怎么跑都跑不出那个指标了。代码没改数据没改但结果就是不一样。原因通常藏在三个地方随机种子、环境依赖、数据版本。随机种子不只是random.seed(42)和np.random.seed(42)就完事了。深度学习框架有自己的随机源CUDA也有自己的随机行为。如果你用了GPU还需要设置torch.manual_seed(42)、torch.cuda.manual_seed_all(42)并且在某些操作上启用确定性算法。但要注意启用确定性算法可能会降低性能所以只在需要精确复现的实验里开。环境依赖的问题更隐蔽。你本地是numpy 1.24服务器上是numpy 1.21某些函数的默认行为可能不一样。更别说CUDA版本、cuDNN版本、甚至不同型号的GPU都可能导致数值差异。解决办法是把环境完整记录下来用pip freeze导出依赖列表用容器镜像固定系统级依赖。数据版本是最容易被忽略的。很多人觉得数据放在那里又不会变但实际上数据可能被清洗脚本重新跑过、被新采集的数据追加过、被人工修正过。没有数据版本管理复现就是碰运气。最简单的做法是给每次数据变更打一个版本号训练时记录用了哪个版本。3.2 手搓一个轻量实验追踪器市面上的实验追踪工具很多但我想让你先手搓一个最小版本理解它到底在追踪什么。核心就三件事记录参数、记录指标、关联产物。import json import os import time import uuid class ExperimentTracker: def __init__(self, base_direxperiments): self.base_dir base_dir self.exp_id str(uuid.uuid4())[:8] self.exp_dir os.path.join(base_dir, self.exp_id) os.makedirs(self.exp_dir, exist_okTrue) self.metrics [] self.params {} def log_params(self, params): self.params params with open(os.path.join(self.exp_dir, params.json), w) as f: json.dump(params, f, indent2) def log_metric(self, name, value, stepNone): self.metrics.append({ name: name, value: value, step: step, timestamp: time.time() }) with open(os.path.join(self.exp_dir, metrics.json), w) as f: json.dump(self.metrics, f, indent2) def log_artifact(self, filepath, artifact_name): import shutil dest os.path.join(self.exp_dir, artifact_name) shutil.copy(filepath, dest) return dest这个追踪器把每次实验的参数、指标、产物都放在一个独立目录里用UUID做标识。参数在实验开始时记录一次指标可以多次记录比如每个epoch记录一次loss产物在训练结束后保存。这样你随时可以回到任何一个实验看到当时用了什么参数、指标怎么变化的、模型文件在哪里。实际使用中我建议在参数里额外记录几个东西代码的git commit hash、数据版本号、开始时间、机器标识。这些信息在排查“为什么这次和上次不一样”的时候非常有用。代码hash可以用subprocess调git命令获取数据版本号从数据管道的输出里读机器标识用platform.node()。3.3 指标记录的粒度选择指标记录太粗会丢失信息太细会拖慢训练。我的经验是分三层记录全局指标、阶段指标、批次指标。全局指标是整个训练结束后才有的比如最终测试集准确率阶段指标是每个epoch或每个验证周期记录的比如验证集loss批次指标是每个batch记录的比如训练loss。批次指标的数据量很大一个epoch有几千个batch的话记录所有批次指标会产生大量数据。我的做法是批次指标只保留最近N个或者做滑动平均后再记录。阶段指标全量保留因为一个epoch才一条数据量可控。全局指标单独存一个文件方便快速查看。另外指标的名字要有统一的命名规范。我见过一个项目同一个人在不同脚本里用了val_acc、val_accuracy、validation_accuracy三种写法最后汇总的时候对不上。建议在项目开始就定好命名规范比如统一用{split}_{metric}的格式train_loss、val_loss、test_accuracy。4. 模型服务化从pickle文件到可用的推理接口4.1 模型加载的性能陷阱训练完的模型保存成pickle文件服务启动时加载看起来很简单。但这里有个性能陷阱如果模型文件很大比如几个GB加载时间可能长达几十秒甚至几分钟。在这段时间里服务是不可用的。如果服务重启频繁或者需要动态加载多个模型这个问题会非常致命。我踩过的坑是这样的一个推荐模型文件有3GB服务每次重启要加载两分钟。后来做AB测试需要同时加载两个模型启动时间变成四分钟。再后来做多路召回要加载五个模型启动时间直接超过十分钟运维那边已经开始报警了。解决办法有几个方向。第一用更高效的序列化格式比如onnx或torchscript加载速度比pickle快很多。第二做模型预热服务启动后在后台异步加载模型加载完成前返回一个默认结果或降级结果。第三用内存映射文件让操作系统按需加载模型参数而不是一次性全部读入内存。import mmap import pickle def load_model_mmap(path): with open(path, rb) as f: mm mmap.mmap(f.fileno(), 0, accessmmap.ACCESS_READ) model pickle.loads(mm) return model内存映射的好处是启动时只建立映射关系不实际读取数据。真正用到某个参数时操作系统才从磁盘加载对应的页。对于大模型来说启动时间可以从分钟级降到秒级。但要注意内存映射的文件不能被修改否则会导致未定义行为。4.2 手搓一个带批处理的推理服务推理服务的核心指标是吞吐量和延迟。单条推理的吞吐量很低因为每次都要做一遍前向传播的计算图调度。批处理可以把多条请求合并成一批一次性做前向传播大幅提升吞吐量。但批处理会增加延迟因为要等凑够一批才能处理。手搓一个带批处理的推理服务核心是维护一个请求队列和一个批处理循环。请求到达时先放入队列批处理循环从队列里取出最多N条请求组成一个batch送给模型然后把结果分发给对应的请求。import queue import threading import numpy as np class BatchInferenceServer: def __init__(self, model, max_batch_size32, max_wait_ms10): self.model model self.max_batch_size max_batch_size self.max_wait_ms max_wait_ms self.request_queue queue.Queue() self.results {} self.lock threading.Lock() self._start_worker() def _start_worker(self): def worker(): while True: batch [] try: # 阻塞等待第一条请求 item self.request_queue.get(timeout1) batch.append(item) # 在max_wait_ms内尽量多收集请求 deadline time.time() self.max_wait_ms / 1000 while len(batch) self.max_batch_size: remaining deadline - time.time() if remaining 0: break try: item self.request_queue.get(timeoutremaining) batch.append(item) except queue.Empty: break self._process_batch(batch) except queue.Empty: continue t threading.Thread(targetworker, daemonTrue) t.start() def _process_batch(self, batch): inputs np.stack([item[input] for item in batch]) outputs self.model(inputs) for item, output in zip(batch, outputs): with self.lock: self.results[item[id]] output def predict(self, input_data): req_id str(uuid.uuid4()) self.request_queue.put({id: req_id, input: input_data}) while True: with self.lock: if req_id in self.results: return self.results.pop(req_id) time.sleep(0.001)这个实现里有两个关键参数max_batch_size和max_wait_ms。前者决定一批最多处理多少条后者决定最多等多久。这两个参数需要根据实际场景调优。如果请求量大且延迟要求不严格可以增大max_batch_size和max_wait_ms如果请求量小且延迟敏感就减小这两个值。4.3 服务降级与熔断推理服务不可能永远稳定。模型可能因为输入异常而报错依赖的特征服务可能超时GPU可能被其他任务占满。如果没有降级和熔断机制一个小的异常就可能拖垮整个服务。降级的思路是当主逻辑失败时返回一个兜底结果。比如推荐服务如果模型推理失败可以返回热门榜单如果特征服务超时可以用默认特征代替。兜底结果的质量肯定不如正常结果但至少服务是可用的。熔断的思路是当某个依赖的失败率超过阈值时暂时切断对该依赖的调用直接走降级逻辑避免大量请求堆积导致雪崩。熔断器一般有三个状态关闭正常调用、打开直接降级、半开允许少量请求试探是否恢复。class CircuitBreaker: def __init__(self, failure_threshold5, recovery_timeout30): self.failure_threshold failure_threshold self.recovery_timeout recovery_timeout self.failure_count 0 self.state closed self.last_failure_time None def call(self, func, *args, **kwargs): if self.state open: if time.time() - self.last_failure_time self.recovery_timeout: self.state half-open else: raise Exception(Circuit is open) try: result func(*args, **kwargs) if self.state half-open: self.state closed self.failure_count 0 return result except Exception as e: self.failure_count 1 self.last_failure_time time.time() if self.failure_count self.failure_threshold: self.state open raise e这个熔断器实现比较粗糙但足以说明原理。实际使用中失败计数应该按时间窗口统计而不是简单的累加。比如统计最近60秒内的失败率超过50%就熔断。另外半开状态下应该只允许一个请求通过成功后再完全恢复。5. 监控与排错线上问题不会给你慢慢调试的机会5.1 推理服务的四个黄金指标线上推理服务最需要关注的四个指标是延迟、吞吐量、错误率、资源利用率。延迟分P50、P95、P99P99延迟高说明有少量请求体验很差可能是长尾问题。吞吐量是每秒处理的请求数反映服务容量。错误率包括业务错误和系统错误业务错误比如输入格式不对系统错误比如超时、内存溢出。资源利用率包括CPU、GPU、内存、网络用来判断是否需要扩容。这四个指标要一起看单独看一个容易误判。比如延迟升高但吞吐量没变可能是某个依赖变慢了吞吐量下降但延迟没变可能是请求量本身减少了错误率升高但延迟正常可能是上游传入了异常数据。我习惯在服务启动时就埋好这些指标的采集点用prometheus_client暴露出来然后用Grafana做面板。如果不想引入这么重的方案至少要把指标写到日志里用grep和awk做简单分析。from prometheus_client import Histogram, Counter, Gauge import time REQUEST_LATENCY Histogram(request_latency_seconds, Request latency) REQUEST_COUNT Counter(request_count, Total requests, [status]) INFLIGHT_REQUESTS Gauge(inflight_requests, In-flight requests) def handle_request(input_data): INFLIGHT_REQUESTS.inc() start time.time() try: result model.predict(input_data) REQUEST_COUNT.labels(statussuccess).inc() return result except Exception as e: REQUEST_COUNT.labels(statuserror).inc() raise e finally: REQUEST_LATENCY.observe(time.time() - start) INFLIGHT_REQUESTS.dec()5.2 一次线上延迟飙升的完整排查过程说一个我亲身经历的案例。某天下午推荐服务的P99延迟从120毫秒突然涨到800毫秒持续了大约十分钟后恢复。没有报警是用户反馈“推荐刷不出来”才发现的。第一步看监控面板。延迟飙升的时间段内CPU利用率从40%涨到90%GPU利用率从60%涨到95%内存正常网络正常。初步判断是计算资源不够。第二步看请求量。同一时间段内请求量并没有明显增加QPS稳定在2000左右。排除流量突增。第三步看错误日志。没有大量报错只有零星几条超时日志。排除代码异常。第四步看依赖服务。特征服务的延迟从20毫秒涨到200毫秒但特征服务本身没有报警。深入查特征服务的日志发现它在同一时间段内做了一次全量缓存刷新刷新期间所有请求都穿透到数据库数据库响应变慢导致特征服务变慢进而拖慢了推理服务。第五步定位根因。特征服务的缓存刷新策略是每天下午三点全量刷新刷新时先清空缓存再重新加载。清空到加载完成之间有大约五分钟的窗口期所有请求都打到数据库。之前数据库压力小没出问题那天数据库上正好有另一个任务在跑叠加起来就扛不住了。修复方案有两个第一缓存刷新改成增量刷新不清空旧缓存新数据加载完成后再原子替换第二给特征服务加熔断数据库响应慢时直接返回默认特征避免拖垮上游。两个方案都上线后再没出现过类似问题。这个案例的教训是监控不能只看自己服务的指标还要看依赖服务的指标。依赖服务的微小抖动可能在上游被放大成严重故障。5.3 日志里应该记录什么日志是排错的第一手资料但很多人的日志要么太少只有“开始处理”“处理完成”要么太多每个变量都打一遍。我的经验是记录五类信息请求标识、关键输入、关键输出、耗时、异常堆栈。请求标识用于串联一次请求在所有服务里的日志。可以用UUID也可以用上游传下来的trace_id。关键输入不是把整个输入都打出来可能很大而是打摘要比如输入的shape、关键字段的值、哈希值。关键输出同理。耗时记录每个阶段的耗时比如特征获取耗时、模型推理耗时、后处理耗时。异常堆栈在出错时打印但要注意脱敏不要把用户敏感信息打出来。import logging import uuid import time logger logging.getLogger(__name__) def handle_request(input_data): req_id str(uuid.uuid4())[:8] logger.info(freq_id{req_id} stagestart input_shape{input_data.shape}) t0 time.time() features get_features(input_data) t1 time.time() logger.info(freq_id{req_id} stagefeatures cost{t1-t0:.3f}s) result model.predict(features) t2 time.time() logger.info(freq_id{req_id} stageinference cost{t2-t1:.3f}s) return result日志格式建议用结构化日志比如JSON格式方便后续用工具做聚合分析。如果日志量很大要注意采样不是所有请求都打全量日志可以按比例采样或者只对慢请求打详细日志。6. 把碎片串起来一条最小可用的AI工程流水线6.1 流水线的目录结构设计把前面几部分串起来需要一个清晰的目录结构。我习惯按功能划分目录而不是按文件类型。比如data/放数据相关脚本model/放模型相关脚本service/放服务相关脚本monitor/放监控相关脚本。每个目录下再按阶段细分。ai-pipeline/ ├── data/ │ ├── collector.py │ ├── cleaner.py │ └── featurizer.py ├── model/ │ ├── trainer.py │ ├── evaluator.py │ └── exporter.py ├── service/ │ ├── server.py │ ├── batch_inference.py │ └── circuit_breaker.py ├── monitor/ │ ├── metrics.py │ └── dashboard.py ├── experiments/ │ └── {exp_id}/ │ ├── params.json │ ├── metrics.json │ └── model.pkl └── configs/ ├── data.yaml ├── model.yaml └── service.yaml这个结构的好处是每个模块职责清晰可以独立开发和测试。数据管道可以单独跑不依赖模型模型训练可以单独跑不依赖服务服务可以单独跑不依赖训练。模块之间通过文件和配置解耦。配置用YAML文件管理不同环境开发、测试、生产用不同的配置文件。配置里放路径、参数、开关不要硬编码在代码里。我见过一个项目把数据库密码写在代码里换环境的时候忘了改连到了生产库差点出大事。6.2 从数据到服务的完整流程完整流程分六步采集、清洗、特征、训练、导出、服务。每一步的输出是下一步的输入中间结果持久化到磁盘。采集步骤从数据源拉取原始数据存到data/raw/目录按日期分文件夹。清洗步骤读取原始数据做标准化处理存到data/clean/目录。特征步骤读取清洗后的数据生成特征矩阵存到data/features/目录。训练步骤读取特征矩阵训练模型把模型文件和实验记录存到experiments/目录。导出步骤把模型转换成服务需要的格式存到service/models/目录。服务步骤加载模型启动推理接口。每一步都可以单独重跑只要保证输入数据存在。比如特征工程改了只需要从特征步骤开始重跑不需要重新采集和清洗。训练改了只需要从训练步骤开始重跑。这种设计让迭代速度大幅提升。# 完整流程 python data/collector.py --date 2024-01-01 python data/cleaner.py --date 2024-01-01 python data/featurizer.py --date 2024-01-01 python model/trainer.py --config configs/model.yaml python model/exporter.py --exp_id latest python service/server.py --config configs/service.yaml6.3 我踩过的三个集成坑第一个坑是路径问题。开发的时候用相对路径跑得好好的部署到服务器上用绝对路径全挂了。解决办法是统一用相对于项目根目录的路径在代码里通过os.path.dirname(__file__)定位项目根目录然后拼接路径。第二个坑是依赖版本冲突。数据管道用pandas 1.5模型训练用pandas 2.0两个模块跑在同一个环境里就冲突了。解决办法是用虚拟环境隔离每个模块有自己的requirements.txt或者用容器把每个模块打包成独立的镜像。第三个坑是时区问题。数据采集用UTC时间模型训练用本地时间服务日志用另一个时区排查问题的时候时间对不上。解决办法是统一用UTC时间只在展示层做时区转换。所有日志、指标、数据文件的时间戳都用UTC避免混乱。7. 一些让我少走弯路的经验关于数据管道我的建议是宁可多存中间结果也不要为了省磁盘空间把多个步骤合并。磁盘便宜时间贵。中间结果保留至少最近一个月的方便回溯和调试。关于实验管理不要等到实验多了才想起来做追踪。从第一个实验开始就用追踪器哪怕只是一个简单的JSON文件。后期补记录的成本远高于前期养习惯的成本。关于模型服务批处理是提升吞吐量最有效的手段但批处理大小和等待时间需要根据实际流量调优。流量大的时候增大批处理流量小的时候减小等待时间。不要一套参数用到底。关于监控不要只监控自己服务的指标。上下游的指标同样重要因为故障往往是从上下游传导过来的。至少要把直接依赖的服务的核心指标纳入监控面板。关于排错养成先看监控、再看日志、最后看代码的习惯。大部分问题在监控和日志层面就能定位不需要读代码。直接读代码容易陷入细节忽略全局。最后说一个心态上的体会。AI工程是一个系统工程涉及数据、模型、服务、运维多个领域没有人能一开始就全部精通。我的做法是每个领域都了解个大概然后选一两个领域深入。遇到问题的时候知道大概是什么方向的问题知道找谁帮忙知道查什么资料这比什么都懂一点但什么都不精要有效得多。