1. 从零搭建AI工程能力:为什么我劝你别一上来就调包
这两年AI应用开发的门槛肉眼可见地降低了,随便拉个框架、调个API就能跑出一个能对话的Demo。但我带过的新人里,十个有八个在真正要上线一个AI功能时卡住——不是模型不会用,而是不知道数据怎么流转、推理服务怎么部署、延迟和成本怎么平衡、线上出问题怎么排查。ai-engineering-from-scratch这个方向之所以值得聊,恰恰是因为它反其道而行:不教你调哪个库,而是让你亲手把一条从数据到推理再到服务的链路搭一遍,理解每一层在干什么。
我自己是从传统后端转过来的,早期也走过“能跑就行”的弯路。后来踩的坑多了才明白,AI工程和普通后端工程的差别,不在于会不会用模型,而在于你要同时处理不确定性输入、算力资源约束、模型版本迭代这三件事。这篇内容就是把我从零搭建一套AI工程能力的过程拆开讲,包括整体思路、核心环节、实操步骤和踩过的坑。适合有一定编程基础、想真正搞懂AI系统怎么落地的人,也适合已经在做AI应用但总觉得“知其然不知其所以然”的开发者。
2. 整体设计思路:先搭骨架,再填血肉
2.1 为什么选择“从零手写”而不是直接上框架
市面上成熟的AI应用框架已经很多了,从模型推理到服务编排都有现成方案。但我坚持认为,想真正掌握AI工程能力,第一步应该是用最原始的方式把核心链路跑通一遍。原因很简单:框架帮你屏蔽了细节,但线上出问题时,恰恰是这些细节在作祟。
举个例子,你用某个高层框架搭了一个RAG应用,本地测试一切正常,上线后用户反馈“回答越来越慢”。你去查框架文档,发现它内部做了向量检索缓存,但缓存策略和你的数据更新频率不匹配,导致旧结果一直被命中。如果你自己手写过检索层,就会立刻意识到问题出在缓存失效逻辑上,而不是盲目地去调模型参数。
从零搭建的核心思路是:把AI系统拆成数据层、模型层、服务层三个独立模块,每层先用最简实现跑通,再逐步替换为生产级方案。这样做的好处是,每一层的输入输出边界清晰,出问题时能快速定位是哪一层的锅。
2.2 三层架构的职责划分与选型考量
我把这套从零搭建的AI工程分成三层,每层的职责和初期选型如下:
| 层级 | 核心职责 | 初期选型 | 生产级替换方向 |
|---|---|---|---|
| 数据层 | 数据清洗、分块、向量化、存储 | 本地文件+简单分块 | 分布式存储+增量索引 |
| 模型层 | 推理、微调、版本管理 | 本地小模型+API调用 | 推理集群+模型注册中心 |
| 服务层 | 请求路由、限流、监控、日志 | 单进程HTTP服务 | 容器编排+网关 |
这个划分不是拍脑袋定的。数据层放在最前面,是因为AI系统的上限由数据质量决定,模型再强,喂进去一堆噪声也白搭。模型层独立出来,是为了让推理逻辑和业务逻辑解耦,方便后续换模型不影响上层服务。服务层单独一层,是因为AI服务的流量特征和普通Web服务差别很大——请求耗时波动大、GPU资源有限、需要处理超时和降级。
注意:初期不要追求每一层都做到完美。数据层先用本地文件跑通,模型层先用小模型验证流程,服务层先用单进程扛住。等整条链路通了,再针对瓶颈逐层优化。我见过太多人一上来就搞分布式向量库,结果连数据分块策略都没调明白。
2.3 从零搭建的阶段性目标设定
整个搭建过程我分了四个阶段,每个阶段有明确的验收标准:
- 阶段一:单条数据跑通。手动输入一段文本,经过分块、向量化、检索、推理,输出一个回答。验收标准是能正确回答一个基于给定文本的问题。
- 阶段二:批量数据+简单服务。把一批文档灌入系统,起一个HTTP服务,能通过接口提问。验收标准是服务能稳定运行,连续请求不崩溃。
- 阶段三:性能与成本优化。引入缓存、批处理、模型量化等手段,把单次请求延迟和成本降下来。验收标准是延迟降低50%以上,成本可控。
- 阶段四:可观测与可维护。加上日志、监控、告警,能快速定位线上问题。验收标准是任意一次请求都能追溯完整链路。
这四个阶段不是线性的,实际做的时候经常要回头调整。比如阶段三做量化时发现精度下降太多,就得回到阶段二重新选模型。但这种反复恰恰是从零搭建的价值所在——你被迫理解每个决策的代价。
3. 核心细节解析:数据层、模型层、服务层的关键实现
3.1 数据层:分块策略比向量模型更重要
很多人一提到数据层就想到“用哪个Embedding模型”,但我的经验是:分块策略对最终效果的影响,往往比换一个更强的向量模型更大。原因在于,检索的质量取决于“检索单元”是否包含了完整的语义信息。如果分块切得稀碎,再好的向量模型也拼不出完整答案。
我试过三种分块策略,对比如下:
| 策略 | 做法 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|---|
| 固定长度 | 按字符数硬切 | 实现简单 | 容易切断语义 | 结构规整的文本 |
| 按段落 | 按换行符切 | 保留自然语义 | 段落长度不均 | 文档类内容 |
| 语义分块 | 按语义相似度合并 | 语义完整 | 计算开销大 | 高质量问答 |
实测下来,按段落分块+最大长度限制是性价比最高的方案。具体做法是:先按换行符切分,如果某个段落超过阈值(比如500字符),再按句子边界二次切分。这样既保留了自然语义,又避免了单块过大导致检索精度下降。
分块之后是向量化。初期我建议用轻量级模型,比如all-MiniLM-L6-v2这类,维度低、速度快,适合快速验证流程。等流程跑通、数据量上来之后,再考虑换更强的模型。这里有个坑:不同向量模型产出的向量维度不同,一旦换了模型,之前存的向量全部要重新生成。所以初期选模型时,最好选一个维度适中、社区活跃的,避免后期迁移成本过高。
实操心得:分块的时候加一点重叠(比如相邻块重叠50字符),能有效缓解“答案刚好被切断”的问题。这个技巧在长文档问答里特别管用,代价只是多存一点向量。
3.2 模型层:推理服务的三个关键参数
模型层最容易踩的坑是“本地跑得好好的,一上服务就崩”。问题通常出在三个参数上:批处理大小、最大并发数、超时时间。
批处理大小决定了单次推理能处理多少请求。设太小,GPU利用率低;设太大,显存容易爆。我的经验值是:先用batch_size=1跑通,然后逐步往上加,同时用nvidia-smi观察显存占用,找到显存占用80%左右的那个值。对于7B级别的模型,量化后通常能跑到batch_size=8左右。
最大并发数要和批处理大小配合。如果批处理是8,并发数设成16,那就有16个请求在排队,实际同时推理的还是8个。并发数设太高会导致请求排队时间过长,用户感知就是“卡”。我一般把并发数设成批处理大小的1.5到2倍,留一点缓冲。
超时时间是最容易被忽略的。AI推理的耗时波动很大,同样的输入,第一次可能要3秒,第二次只要0.5秒(缓存命中)。如果超时设得太短,正常请求会被误杀;设得太长,异常请求会拖垮整个服务。我的做法是:先统计P99耗时,然后把超时设成P99的1.5倍。比如P99是4秒,超时就设6秒。
# 推理服务的关键参数配置示例 INFERENCE_CONFIG = { "batch_size": 8, # 根据显存调整 "max_concurrency": 16, # 批处理大小的1.5-2倍 "timeout_seconds": 6, # P99耗时的1.5倍 "max_retries": 1, # 超时后重试一次 }3.3 服务层:请求生命周期与降级策略
服务层的核心任务是管理请求的完整生命周期。一个AI请求从进来到出去,要经过:鉴权、限流、路由、推理、后处理、返回。每个环节都可能出问题,所以每个环节都要有降级策略。
鉴权环节,初期可以用简单的API Key,但要注意Key的存储和轮换。我见过有人把Key硬编码在代码里,结果代码一泄露,Key就被人盗用了。正确做法是把Key放在环境变量或配置中心,定期轮换。
限流环节,AI服务和普通Web服务的限流策略不一样。普通服务按QPS限流就行,AI服务还要考虑Token消耗速率。因为不同请求消耗的算力差异很大,一个长文本请求可能顶十个短请求。我的做法是双维度限流:QPS限制请求数量,Token速率限制算力消耗。
路由环节,要根据请求类型分发到不同的推理后端。比如简单问答走小模型,复杂推理走大模型。这样既能保证效果,又能控制成本。
降级策略是服务层的最后一道防线。当推理服务过载或超时时,不能直接给用户报错,而是要有兜底方案。常见的降级策略有:返回缓存结果、返回简化版回答、返回“稍后再试”的提示。我一般会准备一个静态的FAQ缓存,当推理服务不可用时,至少能回答一些常见问题。
注意:降级策略一定要在测试环境验证过。我踩过一次坑,线上触发降级后,发现降级逻辑本身有bug,导致整个服务雪崩。从那以后,我每次上线前都会专门跑一遍降级测试。
4. 实操过程:从零搭建一条完整的AI推理链路
4.1 环境准备与依赖安装
先说明一下,这套环境是在一台带GPU的Linux机器上搭建的,如果你手头只有CPU机器,也能跑通流程,只是推理速度会慢一些。我建议初期用CPU验证逻辑,等流程通了再上GPU。
基础环境需要Python 3.10以上,以及以下核心依赖:
# 创建虚拟环境 python -m venv ai-env source ai-env/bin/activate # 安装核心依赖 pip install torch transformers sentence-transformers fastapi uvicorn numpy这里解释一下每个依赖的作用:torch和transformers负责模型加载和推理,sentence-transformers负责向量化,fastapi和uvicorn负责起HTTP服务,numpy负责向量运算。初期不需要装向量数据库,用numpy做余弦相似度就够了。
实操心得:装
torch的时候注意选对版本。如果你用的是CUDA 11.8,就装对应版本的torch,否则会报“CUDA不可用”。我一般去PyTorch官网复制安装命令,比直接pip install torch靠谱。
4.2 数据分块与向量化的完整代码实现
数据分块我写了一个简单的函数,核心逻辑是“先按段落切,再按句子切,最后加重叠”:
import re def split_text(text, max_chunk_size=500, overlap=50): # 按段落切分 paragraphs = text.split('\n\n') chunks = [] for para in paragraphs: para = para.strip() if not para: continue # 如果段落小于阈值,直接作为一个块 if len(para) <= max_chunk_size: chunks.append(para) else: # 按句子切分 sentences = re.split(r'(?<=[。!?.!?])', para) current_chunk = "" for sentence in sentences: if len(current_chunk) + len(sentence) <= max_chunk_size: current_chunk += sentence else: if current_chunk: chunks.append(current_chunk) current_chunk = sentence if current_chunk: chunks.append(current_chunk) # 加重叠 overlapped_chunks = [] for i, chunk in enumerate(chunks): if i > 0: prev_tail = chunks[i-1][-overlap:] overlapped_chunks.append(prev_tail + chunk) else: overlapped_chunks.append(chunk) return overlapped_chunks向量化用sentence-transformers,几行代码就能搞定:
from sentence_transformers import SentenceTransformer model = SentenceTransformer('all-MiniLM-L6-v2') def embed_chunks(chunks): embeddings = model.encode(chunks, normalize_embeddings=True) return embeddings注意normalize_embeddings=True这个参数,它会把向量归一化到单位长度,这样余弦相似度就等价于点积,计算更快。
4.3 检索与推理的串联逻辑
检索的核心是计算查询向量和所有文档向量的相似度,取Top-K:
import numpy as np def retrieve(query, chunks, embeddings, top_k=3): query_embedding = model.encode([query], normalize_embeddings=True)[0] scores = np.dot(embeddings, query_embedding) top_indices = np.argsort(scores)[-top_k:][::-1] return [chunks[i] for i in top_indices]推理部分,初期我建议先用一个小的生成模型,比如flan-t5-small,跑通流程再说:
from transformers import pipeline generator = pipeline('text2text-generation', model='google/flan-t5-small') def generate_answer(query, context_chunks): context = "\n".join(context_chunks) prompt = f"基于以下信息回答问题:\n{context}\n\n问题:{query}\n回答:" result = generator(prompt, max_length=200) return result[0]['generated_text']把这三步串起来,就是一个最简的RAG流程。虽然简陋,但每一行代码你都知道在干什么,出问题也能快速定位。
4.4 起一个HTTP服务并压测
用FastAPI起服务,核心代码不到20行:
from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class QueryRequest(BaseModel): question: str @app.post("/ask") def ask(request: QueryRequest): chunks = retrieve(request.question, all_chunks, all_embeddings) answer = generate_answer(request.question, chunks) return {"answer": answer}启动命令:uvicorn main:app --host 0.0.0.0 --port 8000。
压测用ab或wrk都行,我习惯用ab:
ab -n 100 -c 10 -p query.json -T application/json http://localhost:8000/ask压测时重点看三个指标:平均延迟、P99延迟、错误率。如果P99延迟远高于平均延迟,说明有长尾请求,通常是某些输入触发了更长的推理路径。这时候就要考虑加超时和降级了。
实操心得:压测的时候一定要用真实数据,别用“你好”“测试”这种短查询。真实用户的查询长度和复杂度差异很大,用假数据压测出来的结果没有参考价值。
5. 常见问题与排查技巧实录
5.1 检索结果不相关怎么办
这是最常见的问题,排查思路按优先级排列:
| 可能原因 | 排查方法 | 解决方案 |
|---|---|---|
| 分块太碎 | 打印检索到的块,看是否语义完整 | 增大分块大小或加重叠 |
| 向量模型不匹配 | 换一个更强的向量模型对比 | 换模型或微调 |
| 查询太短 | 看查询是否只有几个词 | 查询扩展或改写 |
| Top-K太小 | 增大Top-K看是否包含正确答案 | 调整Top-K值 |
我的经验是,先调分块,再调Top-K,最后才考虑换模型。因为换模型的成本最高,而且很多时候问题根本不在模型上。
5.2 推理速度慢的优化路径
推理慢的优化要分情况讨论。如果是首次请求慢,那是模型加载的问题,解决办法是服务启动时预热,先跑几个假请求把模型加载到显存。如果是所有请求都慢,那可能是模型太大或批处理没开,解决办法是量化模型或开批处理。如果是偶尔慢,那可能是长尾请求或资源竞争,解决办法是加超时和限流。
我实测下来,量化对速度的提升最明显。一个7B模型从FP16量化到INT8,显存占用减半,推理速度提升30%以上,精度损失通常在1%以内。对于大多数应用场景,这个 trade-off 是划算的。
5.3 服务上线后内存泄漏的排查
内存泄漏是AI服务上线后最容易遇到的问题。表现是服务跑一段时间后,内存占用越来越高,最后OOM崩溃。常见原因有三个:缓存没设上限、请求对象没释放、日志文件没轮转。
排查方法是:在服务里加一个定时任务,每隔一段时间打印一次内存占用和对象数量。如果发现某个对象的数量持续增长,那就是泄漏点。我遇到过一次,是因为缓存字典只增不减,后来改成LRU缓存就解决了。
注意:Python的垃圾回收对循环引用处理得不好,如果代码里有循环引用,内存就不会释放。可以用
gc模块手动触发回收,或者用weakref打破循环引用。
5.4 模型更新后效果变差的回滚策略
模型更新是好事,但更新后效果变差就是事故了。我的做法是:每次模型更新都保留旧版本,新版本先跑灰度,对比效果后再全量。灰度的方法是:把10%的流量切到新模型,对比两边的回答质量和延迟。如果新模型在关键指标上不如旧模型,就自动回滚。
回滚策略要提前写好,不能等出事了再临时想。我一般会准备一个model_registry,记录每个模型的版本、路径、指标。切换模型只需要改一个配置项,重启服务即可。
6. 从零搭建之后:下一步可以怎么扩展
这套从零搭建的AI工程链路跑通之后,你会发现很多地方可以继续深挖。数据层可以引入增量索引,让新数据不用全量重建;模型层可以加微调流程,让模型更懂你的业务;服务层可以加A/B测试,让不同模型版本在线对比。
我个人在实际操作中的体会是,从零搭建最大的价值不是省了多少钱,而是让你对每个环节都有掌控感。当线上出问题时,你知道该去看哪一层、该调哪个参数,而不是对着框架文档干瞪眼。这种掌控感,是调包调不出来的。
最后分享一个小技巧:每次改动只动一个变量,改完立刻压测对比。我见过太多人一次改好几个地方,结果效果变差了都不知道是哪个改动导致的。慢一点,稳一点,比什么都强。