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_size1跑通然后逐步往上加同时用nvidia-smi观察显存占用找到显存占用80%左右的那个值。对于7B级别的模型量化后通常能跑到batch_size8左右。最大并发数要和批处理大小配合。如果批处理是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_size500, overlap50): # 按段落切分 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_embeddingsTrue) return embeddings注意normalize_embeddingsTrue这个参数它会把向量归一化到单位长度这样余弦相似度就等价于点积计算更快。4.3 检索与推理的串联逻辑检索的核心是计算查询向量和所有文档向量的相似度取Top-Kimport numpy as np def retrieve(query, chunks, embeddings, top_k3): query_embedding model.encode([query], normalize_embeddingsTrue)[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, modelgoogle/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_length200) 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都行我习惯用abab -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测试让不同模型版本在线对比。我个人在实际操作中的体会是从零搭建最大的价值不是省了多少钱而是让你对每个环节都有掌控感。当线上出问题时你知道该去看哪一层、该调哪个参数而不是对着框架文档干瞪眼。这种掌控感是调包调不出来的。最后分享一个小技巧每次改动只动一个变量改完立刻压测对比。我见过太多人一次改好几个地方结果效果变差了都不知道是哪个改动导致的。慢一点稳一点比什么都强。