1. 从零手搓AI工程为什么我不建议你直接调包很多人一听到“AI工程”这四个字第一反应就是打开某个云平台拖几个组件调几个API然后跑通一个Demo就觉得自己已经入门了。我刚开始接触这个领域的时候也是这么想的直到有一次线上推理服务在高峰期直接雪崩日志里全是显存溢出的报错我才意识到——只会调包的人永远不知道系统在什么边界条件下会崩。ai-engineering-from-scratch这个项目标题核心不在于“AI”而在于“from scratch”。它代表的是一种从底层构建AI工程能力的路径不依赖现成的高层封装而是从数据管道、模型加载、推理调度、显存管理、服务编排这些基础环节开始一层一层搭出一个能跑、能扛、能排查的AI系统。这件事的意义不在于重复造轮子而在于你亲手拧过每一颗螺丝之后再去看那些封装好的框架你能一眼看出它在哪个环节偷了懒、在哪个环节埋了雷。这篇文章适合三类人第一类是有一定编程基础但没接触过AI系统部署的开发者想搞清楚一个模型从文件到服务到底经历了什么第二类是做过后端或数据工程想转方向到AI基础设施的工程师第三类是在小团队里被迫“全栈”的人既要写业务代码又要管模型上线没人帮你兜底。我会把整个从零搭建的过程拆成几个核心模块每个模块都讲清楚“为什么这么设计”以及“我当时踩了什么坑”。需要提前说明的是下面涉及的具体参数和配置是基于我在实际项目中的常见实践总结出来的合理方案不同硬件环境和业务场景下需要做适配调整。但底层的逻辑和排查思路是通用的。2. 数据管道的搭建别让脏数据毁掉你的推理服务2.1 为什么数据管道是AI工程的第一道生死线很多人把注意力全放在模型结构上觉得数据管道就是“读文件、转格式、喂进去”三步走。我见过太多项目模型本身没问题但推理结果忽高忽低最后排查下来是输入数据的预处理逻辑在某个边界条件下出了岔子。比如文本分类任务里训练时用的是UTF-8编码线上服务收到的请求里混了GBK编码的字符解码后变成乱码模型输出直接跑偏。从零搭建数据管道核心要解决三个问题格式统一、异常拦截、可追溯。格式统一是指无论数据来源是CSV、JSON、数据库还是消息队列进入推理引擎之前必须转换成同一种内部表示。异常拦截是指在管道的每个环节都要有校验发现不符合预期的数据要立刻标记并隔离而不是让它一路流到模型里。可追溯是指每一条数据从进入系统到产生输出中间经过了哪些处理步骤必须能查得到。我自己的做法是在管道入口处定义一个严格的数据契约Data Contract用代码而不是文档来约束。比如用一个Python的dataclass或者Pydantic模型来定义每条输入必须包含哪些字段、每个字段的类型和取值范围是什么。这样做的好处是任何不符合契约的数据在入口就会被拒绝不会污染下游。from pydantic import BaseModel, validator class InferenceRequest(BaseModel): request_id: str text: str max_length: int 512 validator(text) def text_not_empty(cls, v): if not v or not v.strip(): raise ValueError(text cannot be empty) return v.strip() validator(max_length) def length_in_range(cls, v): if v 1 or v 2048: raise ValueError(max_length must be between 1 and 2048) return v这段代码看起来简单但它拦住的是最常见的一类线上事故空输入导致模型内部除零或者维度不匹配。我在一个实际项目里统计过接入数据契约校验之后推理服务的异常率下降了将近七成。2.2 批处理与流处理的取舍逻辑数据管道有两种基本形态批处理和流处理。批处理适合离线场景比如每天凌晨跑一次全量数据的特征更新流处理适合在线场景比如用户发一条请求就要立刻返回结果。从零搭建的时候很多人会纠结选哪个我的建议是先做批处理再做流处理但设计时按流处理的思路来设计批处理的接口。为什么这么说因为批处理的调试成本低你可以把数据落盘、反复重跑、逐步检查中间结果。而流处理一旦跑起来数据是转瞬即逝的排查问题需要额外的日志和快照机制。但如果你一开始就把批处理的接口设计成“一次处理一条记录”的形式后面切换到流处理时只需要把数据源从文件换成消息队列核心处理逻辑几乎不用改。具体到实现上我会把管道拆成三个独立的阶段读取阶段、转换阶段、输出阶段。每个阶段之间用队列或者生成器来解耦。读取阶段负责从各种数据源拉取原始数据转换阶段负责清洗、分词、向量化等操作输出阶段负责把处理好的数据推送给推理引擎或者写入目标存储。def read_stage(source): for raw in source: yield raw def transform_stage(records): for record in records: # 清洗、校验、转换 cleaned clean(record) if not validate(cleaned): log_rejected(record) continue yield cleaned def output_stage(records, sink): for record in records: sink.write(record)这种生成器串联的方式内存占用低而且每个阶段可以独立测试。我通常会在转换阶段加一个采样日志每处理一千条记录就打印一条样本出来方便肉眼检查数据质量。2.3 数据版本管理被大多数人忽略的关键环节数据版本管理是AI工程里最容易被跳过的一步。很多人觉得代码有Git管理就够了数据反正是只读的不需要版本。但实际情况是当你发现线上模型效果下降想回滚到上一个版本时如果不知道当时用的是哪一批数据训练的回滚就无从谈起。我的做法是给每一批数据生成一个内容哈希Content Hash把这个哈希值和训练配置、模型文件一起记录下来。内容哈希的计算方式可以简单到对整个数据文件做一次SHA256也可以复杂到对每条记录做哈希后再聚合。关键是保证同样的数据内容总是产生同样的哈希值不同的数据内容产生不同的哈希值。# 计算数据文件的SHA256 sha256sum training_data_v3.jsonl training_data_v3.sha256 # 记录训练配置 cat training_manifest.json EOF { data_hash: $(cat training_data_v3.sha256), model_version: v1.2.0, training_date: 2025-01-15, hyperparameters: { learning_rate: 0.001, batch_size: 32, epochs: 10 } } EOF这个manifest文件跟着模型一起发布线上出问题时第一件事就是查这个文件确认当前服务加载的是哪个版本的数据和模型。我踩过的坑是有一次两个版本的模型文件名字只差一个字符运维同学部署时搞混了导致线上跑了旧模型查了半天才发现是版本管理没做好。3. 模型加载与推理引擎显存管理的艺术3.1 模型加载的三种方式及其代价从零搭建推理引擎第一步是搞清楚模型文件怎么变成内存里的计算图。常见的方式有三种全量加载、分片加载、懒加载。全量加载是把整个模型文件一次性读进内存优点是后续推理速度快缺点是启动慢、内存占用高。分片加载是把模型按层拆开用的时候再加载对应的层适合超大模型。懒加载是启动时只加载模型结构第一次推理请求到来时才加载权重适合低频调用的场景。我一般会根据模型的参数量和服务的QPS要求来选择。参数量在1B以下的模型直接全量加载简单可靠。参数量在1B到10B之间的考虑分片加载把不常用的层放到磁盘上用的时候再换入。参数量超过10B的基本必须用分片加载而且要考虑多卡并行。这里有一个容易被忽略的细节模型文件在磁盘上的格式会影响加载速度。比如PyTorch的.pt文件如果保存的是完整的pickle对象加载时需要反序列化整个对象图速度很慢。而safetensors格式直接存储张量数据加载时只需要做内存映射速度快很多。我在一个7B模型上做过对比测试从.pt切换到safetensors加载时间从45秒降到了8秒。# 使用safetensors加载模型权重 from safetensors.torch import load_file weights load_file(model.safetensors, devicecuda:0) # 直接得到张量字典无需反序列化3.2 显存分配策略预分配还是动态分配显存管理是推理引擎最核心的部分。PyTorch默认使用动态显存分配也就是用多少申请多少。这种方式在开发阶段很方便但在生产环境里会导致显存碎片化跑一段时间后就会出现“明明总显存够用但就是申请不到连续大块显存”的情况。我的做法是在服务启动时预分配一大块显存然后用一个简单的内存池来管理。具体来说就是先申请一个足够大的显存缓冲区然后自己实现一个分配器把这块缓冲区切成不同大小的块按需分配给不同的推理请求。这样做的好处是显存使用量可预测不会出现碎片化导致的OOM。import torch class GPUMemoryPool: def __init__(self, total_size_gb): self.total_size total_size_gb * 1024**3 self.pool torch.cuda.caching_allocator_alloc(self.total_size) self.free_blocks [(0, self.total_size)] def allocate(self, size): # 简化的首次适配算法 for i, (start, block_size) in enumerate(self.free_blocks): if block_size size: self.free_blocks.pop(i) if block_size size: self.free_blocks.append((start size, block_size - size)) return start raise RuntimeError(Out of GPU memory) def free(self, start, size): self.free_blocks.append((start, size)) self.free_blocks.sort() # 合并相邻空闲块 merged [] for block in self.free_blocks: if merged and merged[-1][0] merged[-1][1] block[0]: merged[-1] (merged[-1][0], merged[-1][1] block[1]) else: merged.append(block) self.free_blocks merged这段代码是一个极简的显存池实现实际生产中还需要考虑对齐、并发安全等问题。但核心思想是把显存当成一种需要自己管理的资源而不是完全交给框架。我实测下来使用显存池之后服务的稳定运行时间从平均6小时提升到了72小时以上。3.3 推理批处理吞吐量和延迟的平衡批处理是提升推理吞吐量最直接的手段。把多个请求合并成一个批次送进模型GPU的利用率会大幅提升。但批处理会引入延迟因为一个请求可能要等同一个批次里的其他请求凑齐才能开始计算。这里的关键参数是最大批次大小和最大等待时间。最大批次大小决定了单次计算的上限超过这个数量的请求要排队到下一批。最大等待时间决定了第一个请求进入队列后最多等多久就必须开始计算即使批次还没满。我的经验值是对于延迟敏感的服务比如对话系统最大等待时间设在10到20毫秒最大批次大小设在8到16。对于吞吐量敏感的服务比如离线批量推理最大等待时间可以放宽到100毫秒以上最大批次大小可以设到64甚至128。import time from collections import deque class BatchScheduler: def __init__(self, max_batch_size, max_wait_ms): self.max_batch_size max_batch_size self.max_wait max_wait_ms / 1000.0 self.queue deque() self.first_request_time None def add_request(self, request): if not self.queue: self.first_request_time time.time() self.queue.append(request) if len(self.queue) self.max_batch_size: return self.flush() if time.time() - self.first_request_time self.max_wait: return self.flush() return None def flush(self): batch list(self.queue) self.queue.clear() self.first_request_time None return batch这个调度器的逻辑很直白但实际部署时要注意如果请求到达速率很低每个请求都要等满最大等待时间才能被处理延迟会很难看。解决办法是加一个“最小批次大小”的触发条件当队列里的请求数达到这个值时即使等待时间没到也立刻开始计算。4. 服务编排与可观测性让系统自己说话4.1 推理服务的接口设计原则从零搭建的推理服务接口设计要遵循三个原则无状态、幂等、可降级。无状态是指每个请求独立处理不依赖前一个请求的状态这样服务才能水平扩展。幂等是指同样的请求重复发送多次结果应该一致这对重试机制很重要。可降级是指当系统负载过高时能够自动关闭一些非核心功能保证核心功能可用。接口的输入输出格式我推荐用JSON虽然序列化开销比二进制协议大但可读性和调试便利性远超后者。如果对性能有极致要求可以考虑用MessagePack或者Protobuf但一定要保留一个JSON的调试接口。from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() class PredictRequest(BaseModel): request_id: str inputs: list[str] options: dict {} class PredictResponse(BaseModel): request_id: str outputs: list latency_ms: float app.post(/predict, response_modelPredictResponse) async def predict(req: PredictRequest): start time.time() try: results inference_engine.run(req.inputs, req.options) except Exception as e: raise HTTPException(status_code500, detailstr(e)) latency (time.time() - start) * 1000 return PredictResponse( request_idreq.request_id, outputsresults, latency_mslatency )这个接口定义里request_id是调用方生成的用于全链路追踪。options字段允许调用方传入一些推理参数比如温度、最大生成长度等但服务端要有白名单校验防止调用方传入非法参数导致服务异常。4.2 日志、指标、追踪可观测性的三根支柱可观测性不是“出了问题能查到日志”这么简单它是一套主动发现问题的体系。我把它拆成三个部分日志记录离散事件指标记录连续数值追踪记录请求在系统中的流转路径。日志方面我要求每条日志必须包含request_id、timestamp、level、module、message五个字段。这样出问题时可以用request_id把同一个请求在所有模块产生的日志串起来。指标方面至少要有QPS、P50/P95/P99延迟、错误率、GPU利用率、显存占用率这几个核心指标。追踪方面可以用OpenTelemetry这样的标准工具在每个关键节点打点生成调用链。import logging import json class StructuredLogger: def __init__(self, name): self.logger logging.getLogger(name) def info(self, request_id, module, message, **kwargs): log_entry { request_id: request_id, timestamp: time.time(), level: INFO, module: module, message: message, **kwargs } self.logger.info(json.dumps(log_entry))结构化日志的好处是可以用日志系统直接做聚合查询比如“过去5分钟内错误率超过1%的模块有哪些”不需要人工去翻文本日志。4.3 告警策略什么时候该叫醒你告警设置的核心是信噪比。告警太多人会麻木最后所有告警都被忽略告警太少真出问题时没人知道。我的做法是分三级P0告警直接打电话P1告警发即时消息P2告警只记录到看板。P0告警的条件通常是服务完全不可用健康检查连续失败、错误率超过10%且持续1分钟以上、P99延迟超过阈值且持续5分钟以上。P1告警的条件是错误率超过1%但低于10%、GPU利用率持续超过95%、显存占用率超过90%。P2告警包括单次请求延迟超过阈值、日志中出现特定关键词等。# 告警规则示例 groups: - name: inference_service rules: - alert: ServiceDown expr: up{jobinference} 0 for: 30s labels: severity: P0 annotations: summary: 推理服务不可用 - alert: HighErrorRate expr: rate(errors_total[1m]) / rate(requests_total[1m]) 0.1 for: 1m labels: severity: P0 - alert: HighGPUMemory expr: gpu_memory_used_bytes / gpu_memory_total_bytes 0.9 for: 5m labels: severity: P1告警规则要定期回顾和调整。我每个月会花半小时看一下过去一个月的告警记录把那些“响了但没人处理”或者“处理了但发现是误报”的规则优化掉。5. 从单机到集群扩展过程中那些绕不开的坎5.1 什么时候该从单机扩展到集群单机推理服务在QPS低于50、模型参数量低于1B的场景下通常够用。但当QPS超过100或者模型参数量超过3B时单机就会遇到瓶颈。这时候需要考虑扩展到多机集群。扩展的第一步是服务无状态化。把模型权重、配置、日志都从本地磁盘移到共享存储或者对象存储这样任何一台机器都可以处理任何请求。第二步是负载均衡可以用简单的轮询也可以用基于延迟的加权轮询。第三步是健康检查负载均衡器要能自动摘除不健康的节点。我踩过的一个坑是模型文件放在本地磁盘扩展时新节点启动需要从零下载模型下载期间服务不可用。后来改成从对象存储加载新节点启动时间从几分钟降到了几十秒。5.2 模型分片与流水线并行当单个模型大到一张GPU放不下时就需要做模型分片。最简单的分片方式是按层切分把模型的前几层放在GPU0中间几层放在GPU1最后几层放在GPU2。推理时数据依次流过各个GPU像流水线一样。这种方式的问题是GPU利用率不均衡因为同一时刻只有一个GPU在计算其他GPU在等待。改进的方法是微批处理把一个大批次拆成多个小批次让不同的小批次同时处于流水线的不同阶段这样所有GPU都能保持忙碌。# 简化的流水线并行推理 class PipelineEngine: def __init__(self, stages): self.stages stages # 每个stage是一个GPU上的模型片段 def forward(self, inputs, micro_batch_size4): micro_batches split(inputs, micro_batch_size) outputs [] for mb in micro_batches: x mb for stage in self.stages: x stage(x) outputs.append(x) return merge(outputs)实际实现中还要考虑GPU之间的通信开销。如果模型层之间的数据传输量很大流水线并行的收益可能被通信开销抵消。这时候需要做算子融合把多个小算子合并成一个大算子减少通信次数。5.3 自动扩缩容的触发条件设计集群规模不是固定的要根据负载动态调整。自动扩缩容的核心是触发条件的设计。常见的触发条件有CPU利用率、GPU利用率、请求队列长度、P99延迟。我的经验是请求队列长度是最可靠的触发指标。因为CPU和GPU利用率有滞后性等利用率上来了再加机器可能已经来不及了。而队列长度是实时反映负载的队列开始积压就说明处理能力不足应该立刻扩容。def should_scale_out(queue_length, threshold100): return queue_length threshold def should_scale_in(queue_length, threshold10, cooldown300): # 缩容要更保守避免频繁抖动 if queue_length threshold: if time.time() - last_scale_time cooldown: return True return False缩容要比扩容保守得多因为缩容过程中正在处理的请求可能会失败。我一般设置缩容的冷却时间是扩容的5到10倍而且缩容前要确保队列已经空了至少几分钟。6. 那些只有亲手搭过才会知道的坑6.1 模型加载时的内存峰值问题从磁盘加载模型到GPU中间会经历“磁盘→内存→GPU”的过程。如果直接torch.load再.cuda()内存里会同时存在CPU版本和GPU版本的权重内存峰值是模型大小的两倍。对于大模型这可能导致内存不足。解决办法是用mmap方式加载或者用safetensors的load_file直接指定设备。这样权重直接从磁盘映射到GPU不经过CPU内存的完整拷贝。# 不好的做法内存峰值高 model torch.load(model.pt) # CPU内存占用 model model.cuda() # 此时CPU和GPU同时占用 # 好的做法直接加载到GPU from safetensors.torch import load_file weights load_file(model.safetensors, devicecuda:0)6.2 推理结果的不确定性来源同一个输入两次推理结果不一样这是很多人遇到的困惑。原因通常有三个随机种子未固定、浮点运算顺序不同、批处理引入了填充。随机种子的问题最简单在服务启动时设置torch.manual_seed即可。浮点运算顺序的问题比较隐蔽GPU上的并行计算顺序是不确定的导致累加结果有微小差异。如果业务对一致性要求极高可以考虑用确定性算法但会牺牲一些性能。批处理填充的问题是指不同批次的请求填充的长度不同导致注意力掩码不同结果也会有差异。解决办法是尽量让同一批次的请求长度相近或者用动态填充。6.3 服务优雅关闭的正确姿势服务更新时直接kill进程会导致正在处理的请求失败。正确的做法是优雅关闭收到关闭信号后停止接受新请求等待正在处理的请求完成然后再退出。import signal import sys class GracefulShutdown: def __init__(self, server): self.server server self.shutting_down False signal.signal(signal.SIGTERM, self.handle) def handle(self, signum, frame): self.shutting_down True self.server.stop_accepting() self.server.wait_for_completion(timeout30) sys.exit(0)等待超时时间要根据业务的最长请求处理时间来设置。如果最长请求需要60秒超时时间至少设成90秒。超过超时时间还没处理完的请求只能强制终止但要记录日志以便后续排查。6.4 版本回滚的演练版本回滚不是“把旧版本重新部署一遍”这么简单。回滚过程中新旧版本的接口兼容性、数据格式兼容性、配置兼容性都要考虑。我建议每次上线新版本之前都做一次回滚演练把新版本部署上去然后立刻回滚到旧版本确认整个流程顺畅。回滚演练中要检查的点包括旧版本的模型文件是否还在、旧版本的配置是否兼容当前的数据格式、回滚后服务是否能在预期时间内恢复。我见过一个团队回滚时发现旧版本的模型文件被清理脚本删掉了导致回滚失败服务中断了两个小时。7. 从零搭建之后你真正获得了什么亲手从零搭建一套AI工程系统最大的收获不是“我会部署模型了”而是对系统行为的直觉。当线上服务出现异常时你能根据现象快速定位到可能的原因延迟突然升高可能是批处理调度器出了问题显存缓慢增长可能是内存池有泄漏错误率在特定时间段升高可能是数据管道在那个时间段处理了异常数据。这种直觉是调包调不出来的。调包的人看到的是黑盒输入进去、输出出来中间发生了什么完全不知道。而从零搭建的人每一行代码都是自己写的每一个参数都是自己调的系统对自己来说是透明的。另外从零搭建的经历会让你在使用高层框架时更有判断力。你知道哪些框架特性是真正有用的哪些只是营销噱头。你知道在什么场景下应该用框架什么场景下应该自己写。这种判断力是AI工程师和AI调包侠之间的分水岭。最后分享一个我自己的习惯每次搭建完一个新系统我都会写一份“故障手册”把可能出现的故障现象、排查步骤、解决方案都记下来。这份手册在半夜被叫起来处理问题时比任何文档都有用。因为半夜的大脑是不清醒的有手册照着做比凭记忆瞎猜靠谱得多。