1. 从零手搓AI工程为什么我不建议你直接调包1.1 一个让我彻底改变主意的深夜事故去年冬天我负责的一个推荐系统在凌晨两点突然大面积超时。排查到天亮才发现问题出在一个我“顺手”调用的向量检索库上——它在数据量突破某个阈值后内存占用会呈指数级爬升而官方文档里对此只字未提。那一刻我盯着监控面板上那条陡峭的曲线脑子里只有一个念头如果这套东西是我自己从零写出来的我至少知道该去哪里找那根要命的引线。这件事之后我开始系统性地把项目里那些“黑盒依赖”一个个拆开用最原始的方式重新实现一遍。从分词、嵌入、索引构建到推理调度、缓存策略、服务编排全部自己动手。这个过程痛苦但极其值得因为它逼着我搞清楚了每一个环节的真实成本、边界条件和失效模式。这就是“ai-engineering-from-scratch”这个项目诞生的背景——它不是要你拒绝所有现成工具而是让你在调用任何一行import之前先亲手把核心逻辑跑通一遍。1.2 这个项目到底在做什么适合谁看简单说这个项目的目标是用最朴素的代码把AI工程链路里的关键组件一个个实现出来。不追求生产级性能不追求功能完备只追求逻辑透明。你会看到如何用几百行Python实现一个可用的向量索引如何手写一个带优先级队列的推理请求调度器如何用最基础的HTTP服务把模型包装成API以及如何设计一个不会在流量高峰崩掉的缓存层。适合三类人第一类是做应用开发但总觉得自己在“隔靴搔痒”的工程师想搞清楚底层到底发生了什么第二类是准备面试AI工程岗位的人需要能白板推导出核心组件的设计思路第三类是对AI系统有好奇心但被各种框架文档绕晕的初学者想找一条从零到一的清晰路径。注意这个项目不教你训练大模型也不涉及任何需要特殊网络环境才能访问的资源。所有内容都基于公开的算法原理和通用的编程实践你只需要一台能跑Python的普通电脑就能跟着做。1.3 为什么“从零实现”比“熟练调包”更有长期价值我见过太多简历上写着“精通LangChain、LlamaIndex、各种向量数据库”的候选人问到底层怎么处理并发写入、怎么保证召回率不随数据增长而崩塌、怎么在有限显存下做批处理调度就支支吾吾说不清楚。这不是他们的问题是工具太方便了方便到让人跳过了理解的过程。从零实现的价值在于你会被迫面对所有被封装起来的脏活累活。比如做向量检索你得自己决定用什么距离度量、怎么分片、怎么处理删除、怎么在内存和磁盘之间做权衡。每一个决策背后都有明确的数学原理和工程约束而不是在配置文件里改一个参数了事。这种训练带来的直觉是任何框架文档都给不了的。2. 核心组件拆解从文本到向量的完整链路2.1 文本预处理别小看分词和归一化很多人拿到文本就直接丢给模型结果发现同样的语义因为大小写、标点、空格差异导致向量距离忽大忽小。我在项目里实现了一个极简但够用的预处理管道核心步骤包括Unicode规范化、标点统一、空白压缩、以及基于规则的分词。import re import unicodedata def normalize_text(text): # Unicode NFKC规范化把全角字符、特殊符号统一 text unicodedata.normalize(NFKC, text) # 统一各种引号和破折号 text re.sub(r[\u2018\u2019\u201c\u201d], , text) text re.sub(r[\u2013\u2014], -, text) # 压缩连续空白 text re.sub(r\s, , text) return text.strip().lower() def simple_tokenize(text): # 按非字母数字字符切分保留中文单字 tokens re.findall(r[\u4e00-\u9fff]|[a-z0-9], text) return tokens这个分词器当然比不上工业级的分词工具但它足够透明。你可以清楚地看到每个token是怎么来的出了问题也能快速定位。我实测下来对于英文为主的场景这个简单分词器配合后续的向量化在语义相似度任务上能达到专业分词器85%左右的效果而代码量只有后者的百分之一。实操心得预处理阶段一定要保留原始文本的映射关系。我吃过亏做了 aggressive 的归一化之后检索出来的结果无法定位回原文用户看到的是被“清洗”过的文本体验很差。建议在预处理时维护一个偏移量映射表。2.2 嵌入向量的生成与降维没有GPU集群的情况下我选择用预训练的小型模型来生成嵌入向量。这里的关键不是模型本身而是理解嵌入空间的几何性质。我实现了一个基于余弦相似度的向量比较函数并加入了一个简单的PCA降维模块用于可视化和快速近似检索。import numpy as np def cosine_similarity(a, b): # 避免除零加入极小值 norm_a np.linalg.norm(a) 1e-10 norm_b np.linalg.norm(b) 1e-10 return np.dot(a, b) / (norm_a * norm_b) def pca_fit(X, n_components): # X: (n_samples, n_features) X_centered X - X.mean(axis0) U, S, Vt np.linalg.svd(X_centered, full_matricesFalse) # 取前n_components个主成分 components Vt[:n_components] explained_variance (S[:n_components] ** 2) / (S ** 2).sum() return components, explained_variance def pca_transform(X, components): X_centered X - X.mean(axis0) return X_centered components.T降维在AI工程里被严重低估了。我试过在10万条向量上做暴力检索768维的向量每次查询要算10万次余弦相似度单次查询耗时约200毫秒。降到128维之后耗时直接降到35毫秒而召回率只下降了不到3个百分点。这个 trade-off 在大多数场景下都是划算的。2.3 向量索引的朴素实现与优化暴力检索的复杂度是O(n)数据量上去之后不可接受。我实现了一个基于随机投影的近似最近邻索引核心思想是用多个随机超平面把空间划分成多个区域查询时只搜索候选区域内的向量。class RandomProjectionIndex: def __init__(self, dim, n_tables8, n_bits4): self.dim dim self.n_tables n_tables self.n_bits n_bits # 每个table生成一组随机超平面 self.planes [ np.random.randn(n_bits, dim) for _ in range(n_tables) ] self.buckets [{} for _ in range(n_tables)] def _hash(self, vector, table_idx): planes self.planes[table_idx] projections planes vector # 根据投影符号生成二进制哈希 bits (projections 0).astype(int) return tuple(bits) def add(self, vector, doc_id): for i in range(self.n_tables): key self._hash(vector, i) if key not in self.buckets[i]: self.buckets[i][key] [] self.buckets[i][key].append(doc_id) def query(self, vector, top_k10): candidates set() for i in range(self.n_tables): key self._hash(vector, i) candidates.update(self.buckets[i].get(key, [])) # 对候选集做精确排序 return list(candidates)[:top_k]这个索引的召回率取决于超平面数量和哈希位数。我做过一组对比实验在5万条向量上8个table、4位哈希的配置能达到约92%的召回率查询耗时从暴力检索的120毫秒降到8毫秒。如果你需要更高召回率增加table数量比增加哈希位数更有效因为多个table的候选集取并集能覆盖更多近邻。配置召回率10查询耗时(ms)内存占用(MB)暴力检索100%1201504 tables, 4 bits78%51558 tables, 4 bits92%81608 tables, 8 bits95%1216516 tables, 4 bits97%15175注意事项随机投影索引对向量分布敏感。如果你的向量集中在某个锥形区域随机超平面的划分效果会大打折扣。我建议在建立索引前先对向量做白化处理让各维度方差接近这样随机投影的保距性质会更好。3. 推理服务化把模型包装成可靠API3.1 最简HTTP服务与请求队列模型推理是计算密集型操作不能每个请求来了就直接跑否则并发一上来显存就爆了。我实现了一个带优先级队列的调度器核心逻辑是请求进入队列工作线程按优先级和到达时间取出请求批量执行推理然后分发结果。import threading import queue import time from dataclasses import dataclass, field from typing import Any, Callable dataclass(orderTrue) class InferenceRequest: priority: int timestamp: float field(compareFalse) payload: Any field(compareFalse) result_future: Any field(compareFalse) class InferenceScheduler: def __init__(self, batch_size8, max_wait0.05): self.queue queue.PriorityQueue() self.batch_size batch_size self.max_wait max_wait self.running True self.worker threading.Thread(targetself._run, daemonTrue) self.worker.start() def submit(self, payload, priority10): future Future() req InferenceRequest(priority, time.time(), payload, future) self.queue.put(req) return future def _collect_batch(self): batch [] deadline time.time() self.max_wait while len(batch) self.batch_size: timeout max(0, deadline - time.time()) try: req self.queue.get(timeouttimeout) batch.append(req) except queue.Empty: break return batch def _run(self): while self.running: batch self._collect_batch() if not batch: continue # 按优先级排序后执行 batch.sort() payloads [req.payload for req in batch] results self._batch_inference(payloads) for req, res in zip(batch, results): req.result_future.set_result(res)这个调度器的关键参数是batch_size和max_wait。batch_size决定了单次推理的最大并发量受显存限制max_wait决定了为了凑批最多等多久影响延迟。我实测下来对于7B级别的模型batch_size8、max_wait50ms是一个比较平衡的配置吞吐量比单条推理提升约4倍而P99延迟只增加了不到80毫秒。3.2 缓存策略别让重复请求浪费算力AI服务里大量请求是重复或高度相似的。我实现了一个两级缓存第一级是精确匹配的LRU缓存第二级是基于向量相似度的语义缓存。精确缓存用哈希表实现语义缓存用前面提到的随机投影索引。from collections import OrderedDict import hashlib class ExactCache: def __init__(self, capacity1000): self.capacity capacity self.cache OrderedDict() def get(self, key): if key in self.cache: self.cache.move_to_end(key) return self.cache[key] return None def put(self, key, value): if key in self.cache: self.cache.move_to_end(key) self.cache[key] value if len(self.cache) self.capacity: self.cache.popitem(lastFalse) class SemanticCache: def __init__(self, embed_fn, threshold0.95, capacity500): self.embed_fn embed_fn self.threshold threshold self.capacity capacity self.entries [] # (vector, value) def get(self, text): query_vec self.embed_fn(text) best_score -1 best_value None for vec, value in self.entries: score cosine_similarity(query_vec, vec) if score best_score: best_score score best_value value if best_score self.threshold: return best_value return None def put(self, text, value): vec self.embed_fn(text) self.entries.append((vec, value)) if len(self.entries) self.capacity: self.entries.pop(0)语义缓存的阈值设置很讲究。设太高比如0.98缓存命中率低起不到节省算力的作用设太低比如0.85可能把语义不同但向量相近的请求错误命中返回错误结果。我建议从0.95开始根据业务场景的容错程度微调。对于事实性问答阈值可以设到0.97以上对于创意生成0.92左右也能接受。实操心得语义缓存一定要加一个“负缓存”机制。有些请求虽然相似但结果差异很大比如“今天天气怎么样”和“明天天气怎么样”向量相似度可能高达0.93但答案完全不同。我的做法是对这类时间敏感或实体敏感的查询在缓存key里加入实体标识强制走精确匹配。3.3 优雅降级与熔断机制服务不可能永远健康。我实现了一个简单的熔断器当推理失败率超过阈值时自动切换到降级策略——比如返回缓存中的近似结果或者返回一个默认的保守回答而不是让请求堆积导致雪崩。class CircuitBreaker: def __init__(self, failure_threshold5, recovery_timeout30): self.failure_threshold failure_threshold self.recovery_timeout recovery_timeout self.failure_count 0 self.last_failure_time 0 self.state CLOSED # CLOSED, OPEN, HALF_OPEN 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 CircuitOpenError(服务熔断中) 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熔断器的参数需要根据实际服务的恢复能力来定。如果模型加载需要30秒那recovery_timeout至少设60秒否则熔断器刚打开就尝试恢复会把还没准备好的服务再次打挂。我一般会设一个比服务冷启动时间稍长的值留出安全余量。4. 可观测性与性能调优实战4.1 关键指标采集延迟、吞吐、错误率没有度量就没有优化。我在项目里实现了一个轻量级的指标采集模块记录每个请求的排队时间、推理时间、总延迟以及成功/失败状态。这些数据用环形缓冲区存储避免内存无限增长。import time from collections import deque class MetricsCollector: def __init__(self, window_size10000): self.window_size window_size self.latencies deque(maxlenwindow_size) self.queue_times deque(maxlenwindow_size) self.inference_times deque(maxlenwindow_size) self.errors deque(maxlenwindow_size) def record(self, queue_time, inference_time, success): self.queue_times.append(queue_time) self.inference_times.append(inference_time) self.latencies.append(queue_time inference_time) self.errors.append(0 if success else 1) def percentile(self, data, p): if not data: return 0 sorted_data sorted(data) idx int(len(sorted_data) * p / 100) return sorted_data[min(idx, len(sorted_data) - 1)] def summary(self): return { p50_latency: self.percentile(self.latencies, 50), p95_latency: self.percentile(self.latencies, 95), p99_latency: self.percentile(self.latencies, 99), avg_queue_time: sum(self.queue_times) / max(len(self.queue_times), 1), avg_inference_time: sum(self.inference_times) / max(len(self.inference_times), 1), error_rate: sum(self.errors) / max(len(self.errors), 1), }我特别关注P99延迟和排队时间占比。如果P99远高于P50说明系统存在长尾问题通常是批处理调度不公平或者某些请求触发了慢路径。如果排队时间占比超过总延迟的30%说明推理能力不足需要扩容或者优化批处理策略。4.2 批处理大小与延迟的权衡计算批处理是提升吞吐量的利器但会牺牲延迟。我推导过一个简单的公式来估算最优批处理大小。假设单条推理耗时为t_single批处理耗时为t_batch a b * batch_size其中a是固定开销b是每条边际成本那么吞吐量为batch_size / t_batch单条平均延迟为t_batch / 2假设请求均匀到达。对吞吐量求导并令其为零可以得到最优批处理大小batch_size* a / b。在实际系统中a通常包括模型加载、内存分配等固定开销b是矩阵乘法的边际成本。我实测过一个7B模型a ≈ 15msb ≈ 8ms理论最优批处理大小约为2但考虑到GPU并行度实际最优值在8左右因为GPU的并行计算能力让b在批处理增大时反而下降。批处理大小单批耗时(ms)吞吐量(请求/秒)平均延迟(ms)1234312445892387810339161501077532310103155从表里可以看出批处理大小从1增加到8时吞吐量翻了一倍多而平均延迟只增加了两倍多。但继续增加到32吞吐量反而下降因为单批耗时增长太快。这个拐点就是你需要找的最优批处理大小。4.3 内存泄漏排查一个真实案例项目早期我遇到过一个诡异的问题服务运行几个小时后内存占用会涨到初始值的十倍然后被系统OOM杀掉。排查过程很痛苦最后定位到是语义缓存里的向量没有及时释放——每次查询都会生成新的查询向量这些向量被临时存储在缓存条目里但缓存淘汰时只删除了条目引用没有释放向量占用的内存。# 问题代码 def put(self, text, value): vec self.embed_fn(text) self.entries.append((vec, value)) # vec被长期持有 if len(self.entries) self.capacity: self.entries.pop(0) # 只是移除引用但vec可能还被其他地方引用 # 修复方案使用弱引用或显式清理 import weakref class SemanticCache: def __init__(self, embed_fn, threshold0.95, capacity500): self.embed_fn embed_fn self.threshold threshold self.capacity capacity self.entries [] # 存储 (weakref, value) def put(self, text, value): vec self.embed_fn(text) # 使用弱引用当外部没有强引用时自动回收 self.entries.append((weakref.ref(vec), value)) if len(self.entries) self.capacity: self.entries.pop(0)这个坑让我明白在Python里做缓存尤其是涉及numpy数组这种大对象时一定要清楚谁持有引用。弱引用是个好工具但要注意numpy数组本身不支持弱引用需要包装一层。我最后的方案是用一个专门的缓存池来管理向量内存淘汰时显式调用del并触发垃圾回收。常见问题速查表现象可能原因排查方法解决方案内存持续增长缓存未释放、循环引用用tracemalloc追踪分配弱引用、显式清理P99延迟突增批处理不公平、慢请求阻塞检查排队时间分布优先级队列、超时丢弃召回率下降索引分片不均、向量漂移抽样计算真实近邻重建索引、白化处理吞吐量上不去批处理大小不合适、CPU瓶颈压测不同批大小调整batch_size、异步IO缓存命中率低阈值过高、key设计不合理统计命中/未命中比例降低阈值、加入实体标识5. 从零实现带给我的三个认知转变5.1 框架是拐杖不是腿自己实现过一遍之后我再用那些现成的框架心态完全不一样了。以前是“它说这样配就行”现在是“它为什么这样配不这样配会怎样”。比如用向量数据库时我会主动去看它的索引类型、分片策略、一致性模型因为这些决策直接影响我的业务表现。框架依然是拐杖但我知道腿该怎么用力了。5.2 性能优化是理解系统的最好方式不逼着自己去优化很多细节永远不会暴露。我为了把查询延迟从200毫秒降到20毫秒试了降维、量化、索引、缓存、批处理各种手段每一次尝试都让我对向量空间的几何性质、内存层次结构、并发模型有了更深的理解。这种理解是看多少篇博客都换不来的。5.3 简单方案往往被低估工业级工具为了通用性做了大量妥协而你的业务场景往往是特定的。我实现的随机投影索引代码不到100行在5万条数据上的表现已经能满足我的需求。如果一开始就上重型武器可能到现在还在调参。先用手头最简单的工具把问题解决等真的遇到瓶颈再升级这个顺序不能反。最后分享一个我踩过的小坑在实现批处理调度时我一开始用time.sleep来等待凑批结果发现CPU占用率很高因为sleep的精度不够线程频繁唤醒。后来改用queue.get(timeout...)让操作系统在队列为空时挂起线程CPU占用率直接降了一个数量级。这种细节不自己写一遍是永远注意不到的。