1. 从零搭建AI工程体系,为什么我劝你别一上来就调包
这两年AI应用开发的门槛被各种框架拉得极低,三行代码调用一个模型接口就能跑通一个Demo,于是很多人产生了一种错觉:AI工程不过如此。但真正在生产环境里摔过跟头的人都知道,从“能跑”到“跑得稳、跑得省、跑得准”,中间隔着一整套工程体系。ai-engineering-from-scratch这个标题之所以值得拿出来聊,恰恰是因为它戳中了一个被大多数人跳过的环节——从底层把AI工程的每一块积木亲手搭一遍。
我自己带过几个从零起步的AI项目,也见过太多团队在“调包阶段”一路顺风顺水,等到要上线、要扩容、要控成本的时候才发现,自己对整个链路的理解是空心的。你不知道token是怎么被切分的,就调不好上下文窗口;你没手写过注意力计算,就理解不了为什么长文本推理会突然爆显存;你没自己实现过向量检索,就搞不清RAG召回率低到底是embedding的问题还是索引的问题。这些坑,调包的时候全被封装藏起来了,出问题的时候也全都会一次性还给你。
所以这篇内容我想聊的,是一个完整的AI工程学习与实践路径该怎么设计。它适合那些已经会用框架、但想真正搞懂底层的人;也适合刚入行、不想只做“API调用工程师”的新人;更适合那些准备在生产环境落地AI应用、需要把控全链路的技术负责人。核心关键词就一个:ai-engineering-from-scratch,从零开始,把AI工程当成一门正经的工程学科来对待,而不是当成一堆API的拼装游戏。
我会按照“整体设计思路—核心模块拆解—实操落地—问题排查”的顺序展开,中间会穿插大量我自己踩过的坑和实测参数。你不需要每一块都亲手实现一遍,但至少要理解每一块在干什么、为什么这么干、不这么干会出什么问题。这才是从零搭建AI工程体系的真正含义。
2. 整体设计思路:从零不等于从轮子造起
2.1 先想清楚“从零”的边界在哪里
很多人一听到“from scratch”就热血上头,觉得要从矩阵乘法开始手写一切。我劝你先冷静。从零搭建AI工程体系,不等于从零造芯片、造框架、造模型。真正的“从零”,是指你要亲手把AI应用的核心链路走通一遍,理解每个环节的输入输出、性能瓶颈和失败模式,而不是把PyTorch重写一遍。
我的建议是把“从零”分成三个层次来理解。第一层是原理层:你需要理解Transformer的注意力机制、位置编码、前向传播的基本数学,但不要求你手推反向传播。第二层是组件层:你需要亲手实现文本切分、embedding调用、向量索引、检索排序、提示词组装、结果解析这些模块,哪怕用现成的库,也要知道库背后做了什么。第三层是系统层:你需要把上面这些组件串成一条可观测、可扩展、可回滚的服务链路。
这三层的边界划清楚,你的学习路径就不会跑偏。我见过有人花两个月手写了一个mini-GPT,结果连一个像样的RAG服务都搭不起来,这就是边界没划清。反过来,也见过有人全程调包,上线后一个召回异常排查了三天,因为不知道向量库的索引结构。
2.2 为什么选择“自底向上”而不是“自顶向下”
市面上大多数AI教程是自顶向下的:先给你一个完整项目,再逐层往下讲。这种方式上手快,但有个致命问题——你对系统的理解是“拼图式”的,缺一块就全盘卡住。而自底向上的路径,是先让你把每个零件摸熟,再组装成机器,出问题时你能快速定位到是哪个零件坏了。
举个具体例子。自顶向下的学习者遇到“模型回答质量突然下降”,第一反应是换模型、调温度、改提示词,试一圈下来可能还是没解决。而自底向上的学习者会先看检索结果对不对、再看上下文拼接有没有截断、再看token计数是否超限、最后才怀疑模型本身。这个排查顺序的差异,就是底层理解带来的。
所以我设计的这条路径,顺序是:文本处理 → 向量化 → 检索 → 提示词工程 → 模型调用 → 服务化 → 可观测性。每一环都先讲原理,再讲实现,最后讲生产环境的注意事项。这个顺序不是随便排的,它对应的是数据在AI应用里的真实流动方向。
2.3 技术选型的取舍逻辑
从零搭建不代表拒绝工具。我的原则是:核心链路自己实现,边缘能力用成熟库。比如文本切分,我会自己写一版基于规则和语义的切分器,因为切分策略直接决定检索质量,必须可控;但HTTP客户端、日志库、配置管理这些,直接用成熟方案,没必要重复造。
具体到技术栈,我实测下来比较稳的组合是这样的:Python作为主语言(生态最全),FastAPI做服务层(轻量、异步友好),向量检索先用numpy手写一版余弦相似度,理解原理后再上FAISS或pgvector,模型调用层自己封装一层统一的provider抽象,方便切换不同模型。这个组合的好处是,每一层你都能看到内部,不会被黑盒卡住。
提示:不要一上来就上LangChain这类重型框架。它封装得太厚,出问题时你很难定位到底是框架的锅还是你的锅。先用轻量方案把链路跑通,理解每一环,再考虑用框架提效。
3. 核心模块拆解:AI工程链路的六大关键环节
3.1 文本切分:被严重低估的第一道关卡
文本切分是整条链路里最不起眼、但影响最大的环节。切分策略不对,后面所有环节都是在错误的基础上做优化。我见过太多RAG项目召回率上不去,最后发现是切分把一句话从中间劈开了,语义直接断裂。
切分要解决的核心问题是:如何在保持语义完整的前提下,把长文本切成适合embedding和检索的片段。这里有几个关键参数需要你自己算。第一个是chunk size,也就是每个片段的长度。太小,语义不完整;太大,检索精度下降。我的经验值是中文300到500字,英文200到400个token。第二个是overlap,也就是相邻片段的重叠长度,一般取chunk size的10%到20%,用来防止关键信息正好落在切分边界上。
但固定长度切分是远远不够的。真正好用的切分器要能识别段落、标题、列表这些结构。我的做法是先按标题和段落做粗切,再对超长段落做滑动窗口细切,最后对每个片段做一次语义完整性检查。这个检查可以用一个简单的规则:如果片段以逗号、顿号结尾,说明语义没结束,需要和后一段合并。
def semantic_split(text, max_len=500, overlap=80): # 先按段落粗切 paragraphs = [p.strip() for p in text.split('\n\n') if p.strip()] chunks = [] buffer = "" for para in paragraphs: if len(buffer) + len(para) <= max_len: buffer += para + "\n" else: if buffer: chunks.append(buffer.strip()) # 超长段落滑动窗口切 if len(para) > max_len: start = 0 while start < len(para): chunks.append(para[start:start+max_len]) start += max_len - overlap buffer = "" else: buffer = para + "\n" if buffer: chunks.append(buffer.strip()) return chunks这段代码不复杂,但比直接按固定长度切效果好很多。实测下来,在中文技术文档上,召回率能提升15%到20%。注意overlap不要设太大,否则会导致检索结果重复,反而稀释了有效信息。
3.2 向量化与embedding:理解语义空间的构建
embedding是把文本映射到高维向量空间的过程,语义相近的文本在空间里距离更近。这个环节的核心是选对embedding模型,以及理解向量维度和归一化的影响。
选模型的时候,不要只看排行榜。排行榜上的模型往往是在通用语料上评测的,你的业务语料可能完全不同。我的做法是:准备50到100条业务相关的查询和对应文档,用几个候选模型分别跑一遍,看召回率。这个测试成本很低,但能避免选错模型带来的长期损失。
维度方面,常见的有384、768、1024、1536维。维度越高表达能力越强,但存储和计算成本也越高。我的经验是,中小规模应用768维足够,大规模检索场景再考虑1024以上。另外,一定要做归一化。归一化之后,余弦相似度就等价于点积,计算更快,而且能避免向量长度差异带来的干扰。
import numpy as np def normalize(vec): norm = np.linalg.norm(vec) return vec / norm if norm > 0 else vec def cosine_sim(a, b): # 归一化后直接点积 return float(np.dot(normalize(a), normalize(b)))这里有个坑要提醒:不同embedding模型产出的向量不能混用,哪怕维度一样也不行。我见过有人把两个模型的向量存到同一个索引里,检索结果乱七八糟。切换模型的时候,必须全量重新生成向量。
3.3 向量检索:从暴力搜索到近似索引
向量检索的目标是:给定一个查询向量,在成千上万个文档向量里快速找到最相似的top-k个。最朴素的做法是暴力搜索,把所有向量算一遍相似度再排序。文档量小的时候(几千条以内)完全够用,而且结果精确。
但文档量上去之后,暴力搜索就扛不住了。这时候需要近似最近邻(ANN)索引。常见的方案有HNSW、IVF、PQ这几种。HNSW检索速度快、精度高,但内存占用大;IVF通过聚类减少搜索范围,内存友好但精度略低;PQ通过量化压缩向量,内存占用极小但精度损失明显。
我的建议是:先用暴力搜索把链路跑通,验证效果;文档量超过一万条再考虑上ANN索引。上索引的时候,一定要做召回率对比测试,确保近似索引的召回率不低于暴力搜索的95%。如果低于这个值,说明索引参数没调好,需要调整ef_search、nprobe这些参数。
| 索引类型 | 内存占用 | 检索速度 | 召回精度 | 适用场景 |
|---|---|---|---|---|
| 暴力搜索 | 高 | 慢 | 100% | 小规模、验证阶段 |
| HNSW | 高 | 快 | 95%+ | 中等规模、精度优先 |
| IVF | 中 | 中 | 90%+ | 大规模、内存受限 |
| PQ | 低 | 快 | 85%+ | 超大规模、可接受精度损失 |
3.4 提示词工程:把检索结果变成模型能用的输入
检索出来的片段不能直接丢给模型,需要组装成结构化的提示词。这个环节的核心是上下文窗口管理和指令设计。
上下文窗口管理要解决的是:检索回来的片段加起来可能超过模型的token上限。这时候需要做截断或重排序。我的做法是先按相似度排序,从高到低往上下文里塞,塞到接近上限就停。同时要预留出系统指令和用户问题的token空间,一般预留20%左右。
指令设计方面,我踩过最大的坑是提示词太啰嗦。早期我写了一大段“你是一个专业的助手,请根据以下资料回答问题,如果资料中没有相关信息请如实告知……”,结果模型经常忽略资料,直接用自己的知识回答。后来我把指令精简成三条硬规则:只依据资料回答、资料没有就说不知道、回答要标注来源。效果立刻好转。
def build_prompt(query, contexts, max_tokens=3000): system = "只依据提供的资料回答问题。资料中没有的信息,直接回答'资料中未提及'。" context_text = "" for i, ctx in enumerate(contexts): if len(context_text) + len(ctx) > max_tokens: break context_text += f"[资料{i+1}] {ctx}\n" prompt = f"{system}\n\n资料:\n{context_text}\n\n问题:{query}\n回答:" return prompt注意:提示词里的“资料编号”很重要。它让模型能引用来源,也方便你在后处理阶段做溯源。没有编号的话,模型引用来源时会含糊其辞。
3.5 模型调用与服务化:把链路变成可用的服务
前面几步都是在本地跑通,最后一步是把它变成一个能对外提供服务的API。这个环节的核心是异步处理、超时控制和降级策略。
模型调用是整条链路里最慢的一环,动辄几秒。如果用同步方式处理,并发一上来服务就卡死了。所以必须用异步。FastAPI配合async/await能很好地处理这个问题。但要注意,模型调用本身如果是同步的,需要用线程池包一层,否则会阻塞事件循环。
超时控制也很关键。模型调用可能因为网络或服务端问题卡住,必须设置合理的超时时间。我的经验值是:检索环节超时2秒,模型调用超时30秒,整体请求超时45秒。超过就返回降级结果,比如“当前请求繁忙,请稍后重试”。
降级策略要提前设计好。模型服务不可用的时候,是返回缓存结果、返回检索原文、还是直接报错?我的做法是返回检索到的原文片段,并提示“模型服务暂时不可用,以下是相关资料”。这样至少用户能拿到有用信息,体验不会完全崩掉。
3.6 可观测性:没有日志的AI服务等于裸奔
AI服务的可观测性和传统服务不太一样。除了常规的QPS、延迟、错误率,你还需要记录:每次请求的检索结果、token消耗、模型返回的原始内容、用户反馈。这些数据是后续优化的基础。
我的做法是在每个环节埋点,记录输入输出和耗时。检索环节记录top-k的相似度分布,如果最高相似度低于某个阈值(比如0.6),说明这次检索可能没找到相关内容,需要标记出来。模型调用环节记录prompt token数和completion token数,用来做成本核算。用户反馈环节记录点赞点踩,用来做效果评估。
这些日志不要只存文本,要结构化存储,方便后续做聚合分析。我一般用JSON格式写日志,然后定期导入到分析工具里。有了这些数据,你才能回答“检索质量有没有下降”“哪个环节是瓶颈”“成本主要花在哪里”这些问题。
4. 实操落地:从零搭建一个可用的RAG服务
4.1 环境准备与依赖安装
先把环境搭起来。Python版本建议3.10以上,太老的版本对异步支持不好。依赖方面,核心就几个:numpy做向量计算,fastapi做服务,uvicorn做服务器,httpx做异步HTTP调用。embedding和模型调用我建议先用HTTP接口的方式,不绑定具体SDK,这样切换模型方便。
pip install numpy fastapi uvicorn httpx pydantic目录结构我习惯这样组织:core放核心逻辑(切分、向量化、检索),api放服务层,config放配置,data放数据和索引文件。这个结构清晰,每一块职责明确,后续扩展也方便。
4.2 核心链路的代码实现
先实现切分和向量化。切分用上面讲的语义切分器,向量化封装一个统一的embedding客户端,支持切换不同的embedding服务。
import httpx import numpy as np class EmbeddingClient: def __init__(self, endpoint, api_key): self.endpoint = endpoint self.headers = {"Authorization": f"Bearer {api_key}"} async def embed(self, texts): async with httpx.AsyncClient(timeout=10) as client: resp = await client.post( self.endpoint, headers=self.headers, json={"input": texts} ) resp.raise_for_status() vectors = [item["embedding"] for item in resp.json()["data"]] return [self._normalize(v) for v in vectors] def _normalize(self, vec): arr = np.array(vec, dtype=np.float32) norm = np.linalg.norm(arr) return arr / norm if norm > 0 else arr然后是检索模块。先用numpy实现暴力搜索,把向量存成一个矩阵,查询时算矩阵和查询向量的点积,取top-k。
class VectorStore: def __init__(self): self.vectors = None self.documents = [] def add(self, vectors, docs): if self.vectors is None: self.vectors = np.array(vectors, dtype=np.float32) else: self.vectors = np.vstack([self.vectors, np.array(vectors, dtype=np.float32)]) self.documents.extend(docs) def search(self, query_vec, top_k=5): if self.vectors is None or len(self.vectors) == 0: return [] scores = self.vectors @ query_vec top_indices = np.argsort(scores)[::-1][:top_k] return [(self.documents[i], float(scores[i])) for i in top_indices]最后把它们串起来,加上模型调用和服务层。模型调用同样用HTTP接口,封装一个统一的客户端。
class LLMClient: def __init__(self, endpoint, api_key, model): self.endpoint = endpoint self.headers = {"Authorization": f"Bearer {api_key}"} self.model = model async def chat(self, prompt, timeout=30): async with httpx.AsyncClient(timeout=timeout) as client: resp = await client.post( self.endpoint, headers=self.headers, json={ "model": self.model, "messages": [{"role": "user", "content": prompt}], "temperature": 0.1 } ) resp.raise_for_status() return resp.json()["choices"][0]["message"]["content"]温度设0.1是因为RAG场景需要稳定输出,不需要创造性。这个参数我调过很多次,0.1到0.3之间比较合适,再高就容易胡编。
4.3 服务层与接口设计
服务层用FastAPI,暴露两个接口:一个用于导入文档,一个用于查询。导入接口接收文档列表,做切分、向量化、入库。查询接口接收问题,做检索、组装提示词、调用模型、返回结果。
from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() store = VectorStore() embedder = EmbeddingClient(...) llm = LLMClient(...) class QueryRequest(BaseModel): question: str top_k: int = 5 @app.post("/ingest") async def ingest(docs: list[str]): chunks = [] for doc in docs: chunks.extend(semantic_split(doc)) vectors = await embedder.embed(chunks) store.add(vectors, chunks) return {"chunks": len(chunks)} @app.post("/query") async def query(req: QueryRequest): q_vec = (await embedder.embed([req.question]))[0] results = store.search(q_vec, req.top_k) contexts = [r[0] for r in results] prompt = build_prompt(req.question, contexts) answer = await llm.chat(prompt) return {"answer": answer, "sources": contexts}这个服务很简陋,但链路是完整的。你可以在此基础上加缓存、加限流、加鉴权,逐步完善。
4.4 参数调优的实测记录
参数调优这块我做了不少对比测试,分享几个关键结论。chunk size从300调到500,召回率提升了8%,但再往上调到800,召回率反而下降了,因为片段太长导致语义稀释。overlap从0调到100,召回率提升明显,但超过150之后提升就很小了,反而增加了存储成本。
top_k的选择也很有意思。top_k=3的时候,答案准确率是78%;top_k=5提升到85%;top_k=8反而降到82%,因为引入了太多噪声。所以top_k不是越大越好,5左右是个比较稳的值。
温度参数方面,0.1的时候答案最稳定,0.3的时候偶尔会有更自然的表达但也会出现小错误,0.7以上基本就不能用了,模型开始自由发挥。这个结论是在我的业务语料上测的,你的场景可能需要微调。
5. 常见问题与排查技巧实录
5.1 检索召回率低的排查思路
召回率低是最常见的问题,排查要按顺序来。先看切分结果,把切分后的片段打印出来,看有没有语义断裂。再看embedding,用几个已知相关的查询和文档算相似度,看分数是否合理。然后看索引,如果是ANN索引,对比一下和暴力搜索的结果差异。最后看查询本身,用户的问法和文档的表述可能差异很大,需要考虑查询改写。
我遇到过一个典型案例:用户问“怎么退款”,文档里写的是“退货流程”,embedding相似度只有0.5,检索不到。后来加了一层查询改写,把“退款”扩展成“退款 退货 返还”,召回率立刻上去了。这个技巧很实用,成本也低。
5.2 模型回答不准确的定位方法
模型回答不准确,先别急着换模型。按这个顺序排查:检索结果里有没有正确答案?如果有,说明是提示词或模型的问题;如果没有,说明是检索的问题。提示词问题通常是指令不够明确,或者上下文太长导致模型忽略了关键信息。模型问题通常是能力不足,需要换更强的模型。
我一般会做一个“黄金测试集”,准备20到30个问题和标准答案,每次改动后跑一遍,看准确率变化。这个测试集不用很大,但要有代表性。有了它,你就能客观判断每次改动是变好还是变坏,而不是凭感觉。
5.3 性能瓶颈的识别与优化
性能瓶颈通常在两个地方:embedding调用和模型调用。embedding调用可以批量处理,一次传多个文本,减少网络往返。模型调用可以用流式输出,让用户更快看到第一个字,体感速度会好很多。
还有一个容易被忽略的点是向量检索。暴力搜索在文档量大的时候会成为瓶颈,这时候需要上ANN索引。但上索引之前,先确认检索真的是瓶颈,用 profiling 工具测一下各环节耗时。我见过有人优化了半天检索,结果发现瓶颈其实在embedding调用上。
| 问题现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| 召回率低 | 切分不当/embedding不匹配 | 打印切分结果、算相似度 | 调整切分、换embedding模型 |
| 回答不准确 | 检索无结果/提示词不清 | 检查检索结果、简化提示词 | 查询改写、优化指令 |
| 响应慢 | embedding/模型调用慢 | profiling各环节耗时 | 批量处理、流式输出 |
| 成本高 | token消耗大 | 统计prompt和completion token | 精简上下文、缓存结果 |
5.4 几个我踩过的坑
第一个坑是向量维度不一致。有次切换embedding模型,忘了重新生成所有向量,结果新旧向量混在一起,检索结果完全乱套。后来我加了一个校验,入库时检查向量维度,不一致就报错。
第二个坑是提示词里的资料编号和实际资料对不上。有次组装提示词的时候,编号从0开始,但模型以为是1开始,引用来源全错了。这种细节看着小,但很影响用户体验。后来我统一从1开始编号,并且在提示词里明确说明。
第三个坑是超时设置不合理。早期模型调用超时设了60秒,结果一个慢请求把整个服务拖垮了。后来改成30秒,并且加了熔断机制,连续超时就暂时跳过模型调用,直接返回检索结果。这个改动让服务的稳定性提升了很多。
6. 从零搭建之后,你真正获得了什么
把这条链路亲手走一遍之后,你获得的不只是一个能用的RAG服务,而是一套完整的排查思路和优化能力。遇到问题时,你知道该从哪里入手,知道每个参数背后的权衡,知道哪些环节可以妥协、哪些必须死守。这种能力,是调包调不出来的。
我个人在实际操作中的体会是,从零搭建最大的价值不在于“造出了什么”,而在于“理解了为什么”。理解了为什么切分策略影响这么大,理解了为什么embedding模型不能随便换,理解了为什么提示词要精简,理解了为什么可观测性比功能本身还重要。这些理解,会在你后续做任何AI项目的时候持续发挥作用。
最后再分享一个小技巧:搭建完之后,别急着扩展功能,先花时间把日志和监控做好。我见过太多项目功能做得花哨,一出问题就抓瞎。有了完善的日志,你才能快速定位问题、验证优化效果、积累经验数据。这个投入,回报率是最高的。