1. 从零搭建AI工程能力:为什么我劝你别一上来就调包
这两年AI应用开发的门槛肉眼可见地在降低,一个刚入门的开发者,花一个下午就能用现成的框架搭出一个能跑通的对话机器人。但我在带团队和做技术评审的过程中发现一个很普遍的现象:很多人能跑通Demo,却说不清楚一次推理请求背后到底发生了什么,模型加载慢在哪里、显存为什么爆、Token是怎么被切分的、向量检索的召回率为什么上不去,全都是一笔糊涂账。ai-engineering-from-scratch这个方向,说白了就是冲着这个问题来的——它主张从最底层的原理和最小可运行单元出发,把AI工程里那些被框架封装掉的环节一个个拆开,自己动手实现一遍,再回头去用成熟工具。
这篇文章适合三类人看。第一类是刚转行做AI应用、只会调API但想搞懂底层机制的开发者;第二类是有一定后端或算法基础、想系统补齐AI工程链路的工程师;第三类是做技术管理、需要判断团队技术方案是否靠谱的负责人。我会围绕从零构建AI工程能力这条主线,把整体设计思路、核心环节的实操要点、完整落地流程以及踩坑经验都摊开讲清楚,尽量做到你看完就能照着复现。
需要先说明一点:从零实现不等于拒绝工具。我的观点一直是,你得先有能力手写一个最小版本,才有资格判断该不该用某个库、用哪个库。就像学开车,你可以开自动挡,但得知道离合和变速箱在干什么,否则车一抛锚你只能干瞪眼。AI工程也是这个道理。
2. 整体设计思路:先拆黑盒,再谈工程化
2.1 为什么选择“从零实现”这条路线
现在主流的AI应用开发,基本都建立在几层抽象之上:最底下是推理引擎,往上是模型服务框架,再往上是编排框架,最上面是各种应用模板。这套分层本身没问题,问题在于大多数人是直接从最上层入门的,导致对下面几层完全没有概念。一旦遇到性能瓶颈或者诡异Bug,排查起来就非常被动。
我举个真实的例子。之前有个同事做的RAG问答系统,用户反馈回答经常答非所问。他第一反应是换更大的模型,换了之后效果提升有限,成本还翻了几倍。后来我让他把检索环节单独拎出来看,发现是文本切分策略有问题——按固定字符数切分,把完整的语义段落切碎了,导致检索出来的片段本身就是残缺的。这个问题跟模型大小毫无关系,纯粹是工程细节没做好。如果他一开始就自己手写过切分和检索的逻辑,大概率不会犯这个错。
从零实现的核心价值,是让你建立一套完整的因果认知:每一个参数、每一个步骤,你都知道它为什么存在、改动了会有什么后果。这种认知是调包调不出来的。
2.2 分层拆解:AI工程到底包含哪些环节
我把一个典型的AI应用拆成五个层次,从下往上依次是:
| 层级 | 核心内容 | 从零实现的关键点 |
|---|---|---|
| 推理层 | 模型加载、前向计算、采样策略 | 理解张量运算、显存管理、解码逻辑 |
| 服务层 | 请求调度、批处理、并发控制 | 手写一个简单的推理服务 |
| 数据层 | 文本切分、向量化、索引构建 | 实现切分算法和相似度检索 |
| 编排层 | 提示词组装、多步调用、状态管理 | 手写一个链式调用流程 |
| 应用层 | 交互逻辑、结果后处理、评估 | 搭建端到端的最小闭环 |
这个分层不是绝对的,不同项目会有交叉,但它能帮你理清学习路径。我的建议是从推理层和数据层入手,因为这两层是AI工程区别于传统后端开发的核心,也是最容易被框架掩盖的部分。服务层和编排层更偏通用工程能力,有后端经验的人上手会快一些。
2.3 技术选型的取舍逻辑
从零实现的时候,语言和工具的选择很关键。我的建议是Python为主,原因很直接:AI生态的底层库基本都是Python优先,你在实现过程中需要对照官方实现来验证自己的理解,用Python能省掉大量翻译成本。
但有几个地方要注意。第一,不要一上来就用PyTorch或TensorFlow这种重型框架,先用NumPy手写矩阵运算和简单的神经网络前向传播,把张量、广播、梯度这些概念吃透。第二,向量检索先用NumPy的余弦相似度实现,别急着上FAISS或Milvus,等你理解了暴力检索的复杂度,再去看这些库的索引结构,会有豁然开朗的感觉。第三,服务层可以用FastAPI这种轻量框架,重点是理解请求生命周期,而不是框架本身的特性。
提示:从零实现的目标是建立认知,不是造轮子。每个环节手写一遍之后,生产环境该用成熟库还是用成熟库,这个判断力才是你真正获得的东西。
3. 核心环节实操要点:把每个黑盒拆开看
3.1 推理层:从矩阵乘法到文本生成
推理层是整个AI工程的地基。很多人觉得模型推理很神秘,其实拆开看就是一连串的矩阵运算加上一个采样策略。我建议你从最简单的开始:用NumPy实现一个单层的前馈网络,输入一个向量,做一次矩阵乘法,过一个激活函数,输出结果。这一步的目的是让你对“模型就是一个函数”这件事有体感。
接下来是理解自回归生成。语言模型生成文本的过程,本质上是每次预测下一个Token,然后把预测结果拼接到输入后面,再预测下一个,循环往复。你可以用NumPy写一个极简版本:给定一个概率分布,按照温度参数调整分布形状,然后用多项式采样选出一个Token。这个过程中你会直观感受到温度参数的作用——温度趋近于0时,采样退化成贪心搜索,输出最确定的那个;温度升高,分布变平,输出更随机。
显存管理是推理层的另一个重点。模型参数、激活值、KV Cache都要占显存。我实测过一个7B参数的模型,FP16精度下光参数就要占大约14GB显存,加上KV Cache和中间激活值,实际占用会更高。理解这些数字怎么来的,你才能在部署时做出合理的资源配置。计算公式很简单:参数量乘以每个参数的字节数。FP16是2字节,INT8是1字节,INT4是0.5字节。7B模型FP16就是7乘以10的9次方再乘以2,约等于14GB。
3.2 数据层:文本切分与向量检索的手写实践
数据层是RAG类应用的核心,也是最容易出问题的地方。先说文本切分。固定长度切分是最简单的,但效果往往最差,因为它会破坏语义完整性。我一般推荐按语义边界切分,比如按段落、按句子,然后在边界处做合并,控制每个片段在合理长度范围内。
具体怎么做?先用正则或标点符号把文本切成句子,然后从前往后累加句子,当累加长度接近目标长度时,就切一刀。目标长度怎么定?这取决于你的嵌入模型的最大输入长度和检索粒度。一般嵌入模型支持512个Token左右,那你的片段长度控制在256到384个Token比较合适,留出余量。片段太短,语义信息不足;片段太长,检索精度下降。
向量检索的手写实现更简单。把所有片段的向量存成一个矩阵,查询时计算查询向量和矩阵中每一行的余弦相似度,取Top-K。余弦相似度的公式是两向量点积除以各自模长的乘积。用NumPy实现就是几行代码的事。这个暴力检索的复杂度是O(n),n是片段数量。当n到十万级别时,单次查询可能要几百毫秒,这时候你才会真正理解为什么需要近似最近邻索引,也才能看懂FAISS那些索引结构在解决什么问题。
3.3 服务层:手写一个最小推理服务
服务层的核心是请求调度和并发控制。我建议用FastAPI写一个最简单的推理服务,接收请求、调用模型、返回结果。这个过程中你会遇到几个关键问题。
第一个是模型加载时机。模型应该在服务启动时加载一次,而不是每个请求都加载。这个道理很简单,但我在实际项目中见过有人在请求处理函数里加载模型,导致每次请求都要等几十秒。正确的做法是在应用启动事件里加载模型,存成全局变量。
第二个是并发控制。模型推理是计算密集型任务,同时处理多个请求会导致显存竞争和计算资源争抢。最简单的方案是加一个信号量或者队列,限制同时处理的请求数。更进一步的方案是批处理,把多个请求攒在一起做一次前向计算,能显著提升吞吐量。批处理的实现要点是设置一个等待窗口,比如10毫秒,窗口内的请求合并成一批。
第三个是超时和错误处理。推理可能因为各种原因失败,比如输入过长、显存不足、模型异常。服务层要做好捕获和降级,返回有意义的错误信息,而不是直接抛500。
3.4 编排层:链式调用与状态管理
编排层解决的是多步调用的组织问题。一个典型的RAG流程包含检索、组装提示词、调用模型、后处理这几个步骤。从零实现的时候,我建议先用最朴素的函数调用串起来,不要急着上LangChain这类框架。
手写编排的好处是你能清楚看到每一步的输入输出。检索返回了什么?提示词组装后的完整文本长什么样?模型原始输出是什么?后处理做了什么改动?这些中间状态在框架里往往被隐藏了,出问题时很难定位。我自己的习惯是在每个步骤加日志,把关键中间结果打印出来,调试效率会高很多。
状态管理是编排层的另一个要点。多轮对话需要维护历史消息,工具调用需要维护调用栈,这些状态怎么存、怎么传递、怎么清理,都需要设计。最简单的方案是用一个字典存会话状态,key是会话ID,value是消息列表。生产环境要考虑持久化和并发安全,但原理是一样的。
4. 完整实操流程:从零搭一个可运行的问答系统
4.1 环境准备与依赖安装
先把环境搭起来。我用的Python版本是3.10,这个版本在AI生态里兼容性最好。依赖方面,核心就是NumPy,其他都可以按需再加。
python -m venv ai-from-scratch source ai-from-scratch/bin/activate pip install numpy fastapi uvicornNumPy用来做矩阵运算和向量检索,FastAPI和uvicorn用来搭服务。暂时不需要装PyTorch,我们先用NumPy把原理跑通。等你对手写版本有感觉了,再换成真正的模型。
注意:不要在一个环境里装太多东西。从零实现的过程中,依赖越少,你对自己代码的掌控力越强。每加一个依赖,都要问自己:这个库帮我解决了什么问题?我能不能自己实现?
4.2 手写文本切分与向量化
先实现文本切分。我写了一个按句子边界切分的函数,核心逻辑是先用标点切句,再按长度合并。
import re def split_sentences(text): pattern = r'(?<=[。!?.!?])\s*' sentences = re.split(pattern, text) return [s.strip() for s in sentences if s.strip()] def chunk_text(text, max_tokens=300, overlap=50): sentences = split_sentences(text) chunks = [] current = [] current_len = 0 for sent in sentences: sent_len = len(sent) if current_len + sent_len > max_tokens and current: chunks.append(''.join(current)) current = current[-1:] if overlap > 0 else [] current_len = sum(len(s) for s in current) current.append(sent) current_len += sent_len if current: chunks.append(''.join(current)) return chunks这里用字符数近似Token数,实际项目中应该用真正的Tokenizer。overlap参数控制相邻片段的重叠长度,目的是避免语义在边界处被切断。我一般设置overlap为目标长度的15%到20%。
向量化部分,为了不依赖外部模型,我先用一个简单的哈希向量代替。原理是把每个词哈希到一个固定维度的向量空间,然后累加。这个向量没有语义信息,但能跑通流程。等你理解了整个链路,再换成真正的嵌入模型。
import numpy as np def hash_embed(text, dim=256): vec = np.zeros(dim) for char in text: idx = hash(char) % dim vec[idx] += 1 norm = np.linalg.norm(vec) return vec / norm if norm > 0 else vec4.3 实现余弦相似度检索
检索逻辑就是暴力计算余弦相似度,取Top-K。
def cosine_similarity(a, b): return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-8) def retrieve(query_vec, doc_vecs, top_k=3): scores = [cosine_similarity(query_vec, dv) for dv in doc_vecs] ranked = np.argsort(scores)[::-1][:top_k] return [(int(i), float(scores[i])) for i in ranked]这段代码很朴素,但它是所有向量数据库的核心。FAISS、Milvus这些库做的事情,本质上是在这个基础上加索引来加速。你先把这个版本跑通,记录一下当文档数量从100增加到10000时查询耗时的变化,就能直观感受到暴力检索的瓶颈在哪里。
4.4 组装提示词并调用模型
提示词组装就是把检索到的片段拼成上下文,加上用户问题和指令。
def build_prompt(query, contexts): context_text = '\n\n'.join([f'[片段{i+1}] {c}' for i, c in enumerate(contexts)]) prompt = f"""基于以下参考片段回答问题。如果片段中没有相关信息,请如实说明。 参考片段: {context_text} 问题:{query} 回答:""" return prompt调用模型这一步,从零实现的话可以先用一个规则函数模拟,比如返回检索到的最相关片段的前100个字符。等你把流程跑通,再替换成真正的模型调用。这个替换过程本身也很有价值,你会清楚知道模型在流程中扮演什么角色,输入输出格式是什么。
4.5 用FastAPI串起完整服务
最后把上面这些串成一个HTTP服务。
from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() doc_vecs = [] doc_chunks = [] class QueryRequest(BaseModel): question: str top_k: int = 3 @app.post('/ask') def ask(req: QueryRequest): q_vec = hash_embed(req.question) hits = retrieve(q_vec, doc_vecs, req.top_k) contexts = [doc_chunks[i] for i, _ in hits] prompt = build_prompt(req.question, contexts) answer = mock_generate(prompt) return {'answer': answer, 'sources': hits}启动命令是uvicorn main:app --reload。跑起来之后用curl或者Postman发个请求,整个链路就通了。这个版本很粗糙,但它包含了RAG系统的所有核心环节。你在这个基础上每优化一个环节,都能清楚知道优化了什么、为什么优化。
5. 常见问题与排查技巧实录
5.1 检索结果不相关怎么办
这是最常见的问题。排查顺序我一般是这样:先看切分是否合理,把检索到的片段打印出来,人工判断它是否包含答案。如果片段本身就是残缺的,问题在切分环节。如果片段完整但不相关,问题在向量化或相似度计算。
向量化的问题通常是嵌入模型不适合你的领域。通用嵌入模型在专业领域表现会下降,这时候要么换领域适配的模型,要么在检索前做查询改写。相似度计算的问题比较少见,但要注意归一化,如果向量没有归一化,余弦相似度和点积的结果会不一致。
还有一个容易被忽略的点是Top-K的选择。K太小可能漏掉相关片段,K太大又会引入噪声。我的经验是先用K=5跑一批测试问题,看召回情况再调整。如果相关片段经常排在5名之外,说明检索质量有问题,不是简单调K能解决的。
5.2 服务响应慢的排查思路
响应慢要分段计时。在检索、提示词组装、模型调用、后处理这几个环节分别打时间戳,看耗时分布。我遇到过的情况里,检索慢通常是因为文档量大且没建索引,模型调用慢是因为没做批处理或者模型太大,后处理慢往往是因为做了不必要的重复计算。
有一个隐蔽的坑是序列化开销。如果中间结果是大对象,频繁的JSON序列化和反序列化会吃掉不少时间。我一般建议在服务内部用原生对象传递,只在最终返回时序列化一次。
5.3 显存不足的常见原因
显存不足不一定是模型太大。常见原因有几个:批处理大小设置过大,KV Cache没有及时释放,中间激活值没有用梯度检查点优化,或者有内存泄漏导致显存逐渐被占满。排查的时候先用工具看显存占用曲线,是启动就高还是逐渐升高。启动就高说明是模型和配置问题,逐渐升高说明是泄漏。
提示:从零实现的过程中,我建议你刻意制造几次显存不足,观察报错信息和显存变化。这种“故意踩坑”的经历,比看十篇教程都管用。
5.4 常见问题速查表
| 问题现象 | 可能原因 | 排查方向 |
|---|---|---|
| 回答答非所问 | 切分破坏语义 | 检查片段完整性 |
| 检索召回率低 | 嵌入模型不匹配 | 换模型或做查询改写 |
| 服务响应慢 | 未做批处理 | 分段计时定位瓶颈 |
| 显存逐渐升高 | 内存泄漏 | 检查KV Cache释放 |
| 输出重复啰嗦 | 采样参数不当 | 调整温度和重复惩罚 |
| 并发请求失败 | 资源竞争 | 加信号量或队列 |
6. 从手写版本到生产环境的演进路径
手写版本跑通之后,下一步是逐步替换成生产级组件。这个替换过程要有序,一次只换一个环节,换完做对比测试,确认没有回退再换下一个。
第一步通常是替换嵌入模型。把哈希向量换成真正的嵌入模型,检索质量会有质的提升。替换后要重新评估切分策略,因为不同嵌入模型对输入长度的敏感度不同。第二步是替换向量检索,把暴力检索换成FAISS或类似库,重点观察召回率有没有变化,因为近似检索是有精度损失的。第三步是替换模型调用,接入真正的推理服务,这时候要重点关注批处理和并发控制。第四步是加评估体系,用一批标注好的问答对来量化每个环节的改动效果。
我自己的体会是,这个演进过程走一遍,你对AI工程的理解会完全不一样。你会知道每个组件的边界在哪里,什么情况下该用什么方案,出了问题该从哪里查。这种判断力,是直接调包永远得不到的。
最后分享一个小技巧:在手写版本的每个环节都留一个开关,可以随时切换到手写实现或生产组件。这样在排查问题时,你可以通过切换开关来快速定位是哪个环节出的问题。这个习惯帮我省了大量调试时间,你也可以试试。