从零搭建AI工程能力：告别调包侠，掌握底层原理与实战

1. 从零搭建AI工程能力：为什么我劝你别再当“调包侠”

“ai-engineering-from-scratch”这个标题，第一次看到的时候我愣了一下。不是因为它有多花哨，恰恰相反，它朴素得有点不像现在这个时代的项目名。现在满大街都是“大模型实战”“Agent开发从入门到精通”“RAG系统落地指南”，突然冒出来一个“从零开始的AI工程”，反而让我觉得有点意思。

先说清楚这个项目是干什么的。简单讲，它是一套面向开发者的AI工程能力自建路线，核心思路是不依赖现成的高级封装框架，从最底层的原理和最小可运行单元开始，一步步搭建出完整的AI应用工程体系。它解决的问题很具体：很多人会用LangChain、LlamaIndex、各种Agent框架，但你问他“一个token是怎么变成向量的”“注意力机制到底在算什么”“向量数据库的索引为什么能加速检索”，他答不上来。这个项目就是冲着这个痛点去的。

适合谁看？如果你已经能跑通几个AI demo，但总觉得自己是在“调包”，换个场景就不知道从哪下手，那这套东西就是给你准备的。如果你是完全零基础，也没关系，因为它确实是从“零”开始，只不过需要你有基本的Python编程能力，至少知道什么是函数、什么是类、什么是HTTP请求。完全不懂编程的话，建议先补一下Python基础再来。

我花了大概三周时间把这套路线完整走了一遍，踩了不少坑，也总结了一些文档里不会写的经验。下面我把整个拆解过程、核心细节、实操步骤和避坑心得全部摊开讲，你照着抄作业就行。

2. 整体设计思路：为什么“从零”反而更快

2.1 先搞清楚“AI工程”到底指什么

很多人把“AI工程”和“机器学习”混为一谈。机器学习关注的是模型怎么训练、损失怎么下降、准确率怎么提升。AI工程关注的是另一件事：怎么把一个已经存在的模型能力，变成一个稳定、可扩展、可维护的产品功能。

举个例子。你有一个训练好的文本分类模型，准确率95%。这是机器学习的事。但你要把它部署成一个API服务，每秒处理1000个请求，响应时间控制在200毫秒以内，还要能自动扩缩容、监控异常、灰度发布。这就是AI工程的事。

“ai-engineering-from-scratch”这个项目的设计逻辑，就是沿着“工程”这条线走的。它不教你从零训练一个GPT，那是不现实的。它教你的是：给定一个模型能力（可以是开源的，也可以是API），你怎么从最底层开始，把它包装成一个真正能用的系统。

2.2 为什么不用现成框架

这是很多人会问的第一个问题。LangChain不香吗？LlamaIndex不好用吗？为什么要自己从头写？

我的理解是这样的：现成框架是“高速公路”，你开上去确实快，但一旦出了事故，你连怎么刹车都不知道。自己从零搭一遍，相当于先学会修路，再上高速。以后遇到框架解决不了的问题，你有能力自己动手。

而且现成框架有一个很大的问题：抽象层太厚。你写三行代码，背后可能跑了几百行逻辑。出了问题，你根本不知道是哪一层挂了。从零开始搭，每一层都是你自己写的，出了问题你一眼就能定位。

这个项目的设计思路就是“分层解耦”。它把AI工程拆成几个独立的层：数据层、模型层、服务层、应用层。每一层只做一件事，层与层之间通过明确的接口通信。这样你可以在任何一层替换实现，而不影响其他层。

2.3 核心架构长什么样

整个项目的架构可以概括为“四层三线”。

四层是：

• 数据层：负责原始数据的清洗、分块、向量化、存储。核心是向量数据库和嵌入模型。
• 模型层：负责调用大模型能力，包括提示词管理、上下文组装、输出解析。
• 服务层：负责把模型能力包装成API，包括请求路由、并发控制、缓存、限流。
• 应用层：负责具体的业务逻辑，比如问答、摘要、代码生成。

三线是：

• 评估线：怎么衡量你的系统好不好用。包括离线评估和在线监控。
• 调试线：出了问题怎么排查。包括日志、追踪、回放。
• 迭代线：怎么持续优化。包括提示词版本管理、模型切换、A/B测试。

这个架构的好处是，你可以从任何一层开始搭建，也可以只搭建你需要的层。比如你只想做一个简单的问答机器人，那数据层和服务层可以先用最简单的实现，重点放在模型层和应用层。

2.4 技术选型的几个关键决策

项目里有一些技术选型的建议，我结合自己的实践说一下。

嵌入模型选型：项目推荐从开源的sentence-transformers开始，比如all-MiniLM-L6-v2。这个模型很小，CPU上就能跑，速度也快。虽然效果不如那些大模型，但用来学习和原型开发足够了。等你需要更好的效果，再换成更大的模型或者API。

向量数据库选型：项目建议先用FAISS，因为它是纯本地的，不需要额外部署服务。FAISS的索引类型很多，从最简单的Flat索引到IVF、HNSW都有。学习阶段用Flat就够了，数据量大了再换。生产环境可以考虑Milvus或者Qdrant，但那是后面的事。

大模型调用：项目建议先用OpenAI的API，因为接口简单，文档全。但同时也建议你准备好一个本地方案，比如用Ollama跑Llama 3或者Qwen。这样你可以在没有网络或者不想花钱的时候继续开发。

服务框架：FastAPI是首选。它异步支持好，自动生成文档，类型检查也方便。Flask也可以，但异步支持不如FastAPI。Django太重了，不适合这种场景。

这些选型的共同逻辑是：先用最简单的方案跑通，再根据实际需求替换。不要一上来就追求“生产级”，那会让你在配置环境上浪费大量时间。

3. 核心细节解析：每一层到底怎么搭

3.1 数据层：从原始文本到向量索引

数据层是整个系统的地基。地基没打好，上面盖什么都是歪的。

第一步是数据清洗。很多人拿到数据直接就开始分块，这是大忌。原始数据里可能有HTML标签、特殊字符、重复内容、乱码。这些东西如果不处理，会直接影响嵌入质量。我的做法是先用正则表达式去掉HTML标签和特殊字符，然后用SimHash或者MinHash去重，最后统一编码格式为UTF-8。

第二步是分块。分块策略直接决定了检索效果。项目里推荐的是“递归字符分块”，也就是按段落、句子、单词的优先级依次尝试分割，直到块大小符合要求。块大小一般设置在256到512个token之间。太小了语义不完整，太大了检索精度下降。

这里有一个坑：不要用固定长度分块。固定长度分块会把一个完整的句子切断，导致语义碎片化。递归分块虽然慢一点，但效果好很多。

第三步是向量化。用嵌入模型把每个文本块转成向量。这里要注意的是，嵌入模型有最大输入长度限制。比如all-MiniLM-L6-v2最大只支持256个token。如果你的块超过这个长度，会被截断。所以分块的时候要确保块大小不超过嵌入模型的最大长度。

第四步是建索引。FAISS的Flat索引就是暴力检索，把所有向量都存下来，查询的时候逐个计算相似度。数据量小的时候没问题，数据量大了就慢了。这时候可以换成IVF索引，先聚类再检索，速度能快很多。但IVF需要训练，而且会损失一点精度。

注意：建索引之前一定要做归一化。把向量除以它的L2范数，这样内积就等于余弦相似度。不做归一化的话，相似度计算会受向量长度影响，结果不准。

3.2 模型层：提示词、上下文和输出解析

模型层是很多人觉得最简单、但实际上最容易出问题的一层。

提示词管理：不要把提示词硬编码在代码里。项目建议把提示词单独放在一个文件或者数据库里，用版本号管理。这样你可以随时回滚到之前的版本，也可以做A/B测试。我自己的做法是用YAML文件存提示词，每个提示词有id、版本、模板、变量列表。

上下文组装：这是RAG系统的核心。用户问一个问题，你需要从向量数据库里检索出最相关的几个文本块，然后把它们和用户问题一起塞进提示词里。这里有几个关键参数：

• 检索数量：一般取3到5个。太少了信息不够，太多了会超出模型上下文限制，而且会引入噪声。
• 相似度阈值：低于某个相似度的结果直接丢弃。这个阈值需要根据你的数据和嵌入模型来调。我的经验是0.7左右比较合适。
• 重排序：检索出来的结果可以再用一个交叉编码器重排序，把最相关的排在最前面。这一步能显著提升效果，但会增加延迟。

输出解析：大模型的输出是自然语言，你需要把它解析成结构化数据。最简单的方法是让模型输出JSON，然后用json.loads解析。但模型有时候会输出多余的文本，导致解析失败。我的做法是用正则表达式先提取JSON部分，再解析。如果还失败，就重试一次，并在提示词里强调“只输出JSON，不要其他内容”。

3.3 服务层：从脚本到API

把脚本变成API，是AI工程化的关键一步。

请求路由：FastAPI的路由很简单，用装饰器定义就行。但要注意的是，大模型调用是IO密集型的，要用async def定义异步路由，否则并发请求会阻塞。

并发控制：大模型API通常有速率限制。你需要用一个信号量或者令牌桶来控制并发数。我的做法是用asyncio.Semaphore，设置一个合理的并发上限，比如10。超过的请求排队等待。

缓存：相同的请求没必要重复调用大模型。可以用Redis或者内存缓存来存结果。缓存的key可以是用户问题的哈希值，value是模型输出。注意设置合理的过期时间，比如1小时。

限流：防止单个用户滥用。可以用slowapi或者自己写一个简单的计数器。按IP或者用户ID限流，每分钟最多N次请求。

错误处理：大模型调用可能失败，网络可能超时。要有重试机制，但重试次数不要太多，一般2到3次就够了。重试的时候要加指数退避，避免雪崩。

3.4 应用层：业务逻辑的组装

应用层是把前面三层串起来的地方。以问答系统为例，流程是这样的：

1. 接收用户问题
2. 把问题向量化
3. 从向量数据库检索相关文本块
4. 组装提示词
5. 调用大模型
6. 解析输出
7. 返回结果

每一步都可能出错，所以每一步都要有日志。我的做法是在每一步前后都打日志，记录输入、输出、耗时。这样出了问题可以快速定位。

4. 实操过程：手把手搭建一个最小可用系统

4.1 环境准备和依赖安装

先创建一个虚拟环境，这是好习惯。

python -m venv ai-eng source ai-eng/bin/activate # Windows用 ai-eng\Scripts\activate

然后安装核心依赖：

pip install fastapi uvicorn sentence-transformers faiss-cpu openai python-multipart pyyaml redis

如果你要用GPU加速嵌入模型，把faiss-cpu换成faiss-gpu，然后安装对应版本的PyTorch。

提示：sentence-transformers第一次运行会下载模型，大概几百MB。如果网络慢，可以提前从HuggingFace镜像下载好，放到缓存目录。

4.2 数据准备和向量化脚本

假设你有一堆Markdown文档，放在data/目录下。先写一个脚本把它们读进来，分块，向量化，存到FAISS索引里。

import os import glob import numpy as np import faiss from sentence_transformers import SentenceTransformer # 加载嵌入模型 model = SentenceTransformer('all-MiniLM-L6-v2') # 读取所有Markdown文件 documents = [] for filepath in glob.glob('data/*.md'): with open(filepath, 'r', encoding='utf-8') as f: text = f.read() documents.append(text) # 简单的递归分块 def chunk_text(text, chunk_size=256, overlap=50): chunks = [] start = 0 while start < len(text): end = start + chunk_size chunk = text[start:end] chunks.append(chunk) start = end - overlap return chunks all_chunks = [] for doc in documents: all_chunks.extend(chunk_text(doc)) # 向量化 embeddings = model.encode(all_chunks, show_progress_bar=True) embeddings = embeddings.astype('float32') # 归一化 faiss.normalize_L2(embeddings) # 建索引 dimension = embeddings.shape[1] index = faiss.IndexFlatIP(dimension) # 内积索引 index.add(embeddings) # 保存 faiss.write_index(index, 'index.faiss') np.save('chunks.npy', np.array(all_chunks))

这个脚本跑完，你会得到两个文件：index.faiss和chunks.npy。前者是向量索引，后者是原始文本块。

4.3 FastAPI服务搭建

接下来写API服务。

from fastapi import FastAPI, HTTPException from pydantic import BaseModel import faiss import numpy as np from sentence_transformers import SentenceTransformer import openai import os app = FastAPI() # 加载资源 model = SentenceTransformer('all-MiniLM-L6-v2') index = faiss.read_index('index.faiss') chunks = np.load('chunks.npy', allow_pickle=True) # 设置OpenAI openai.api_key = os.getenv('OPENAI_API_KEY') class QueryRequest(BaseModel): question: str top_k: int = 3 class QueryResponse(BaseModel): answer: str sources: list @app.post('/query', response_model=QueryResponse) async def query(req: QueryRequest): # 向量化问题 q_emb = model.encode([req.question]).astype('float32') faiss.normalize_L2(q_emb) # 检索 distances, indices = index.search(q_emb, req.top_k) # 组装上下文 context = '\n\n'.join([chunks[i] for i in indices[0]]) # 调用大模型 prompt = f"""基于以下上下文回答问题。如果上下文不包含答案，就说不知道。 上下文： {context} 问题：{req.question} """ response = openai.ChatCompletion.create( model='gpt-3.5-turbo', messages=[{'role': 'user', 'content': prompt}], temperature=0.1 ) answer = response.choices[0].message.content return QueryResponse(answer=answer, sources=[chunks[i] for i in indices[0]])

启动服务：

uvicorn main:app --reload --host 0.0.0.0 --port 8000

然后你就可以用curl或者Postman测试了。

curl -X POST http://localhost:8000/query \ -H "Content-Type: application/json" \ -d '{"question": "什么是向量数据库？", "top_k": 3}'

4.4 关键参数的计算和选择

分块大小：我试过128、256、512三种。128太小，检索出来的块语义不完整；512太大，检索精度下降。256是我实测下来最平衡的。但这不是绝对的，取决于你的文档类型。技术文档可以小一点，叙事文档可以大一点。

重叠长度：一般取分块大小的10%到20%。256的块，重叠50个字符左右。重叠的目的是防止一个完整的句子被切断。但重叠太多会导致索引膨胀，检索变慢。

检索数量top_k：3到5个。我一般用3个，因为大模型的上下文窗口有限，塞太多反而会稀释关键信息。如果你的文档很长，可以先用检索召回10个，再用重排序选出3个。

相似度阈值：这个需要根据你的数据调。我的做法是先跑一批测试问题，看正确结果的相似度分布，然后取一个能过滤掉大部分错误结果的值。一般0.6到0.8之间。

5. 常见问题与排查技巧实录

5.1 检索结果不相关怎么办

这是最常见的问题。可能的原因和排查步骤：

我的经验是，90%的检索问题都出在分块上。分块没分好，后面怎么调都没用。所以先把分块做好，再调其他参数。

5.2 大模型输出不稳定怎么办

大模型的输出确实有随机性。降低随机性的方法：

• 降低temperature：设成0.1或者0。但注意，有些模型在temperature=0的时候反而会输出重复内容。
• 固定seed：如果API支持，设置一个固定的随机种子。
• 优化提示词：把要求写清楚，比如“只输出JSON”“不要解释”“直接回答”。
• 输出解析容错：不要假设模型一定输出合法JSON。用try-except包裹解析逻辑，失败了就重试或者返回默认值。

我踩过的一个坑是：提示词里写了“请输出JSON”，但模型还是在JSON前面加了一句“好的，以下是JSON：”。后来我改成“只输出JSON，不要任何其他文字”，问题就解决了。

5.3 服务响应太慢怎么优化

响应慢通常有三个原因：嵌入慢、检索慢、大模型慢。

嵌入慢：如果用CPU跑嵌入模型，确实会慢。可以换成更小的模型，或者用GPU。也可以把嵌入结果缓存起来，相同的问题不用重复嵌入。

检索慢：FAISS的Flat索引在数据量大的时候会慢。可以换成IVF或者HNSW索引。IVF需要训练，但查询速度快很多。HNSW不需要训练，查询也快，但内存占用大。

大模型慢：这是最不可控的。可以换更小的模型，或者用流式输出，让用户先看到部分结果。也可以做缓存，相同的问题直接返回缓存结果。

我的做法是先用缓存扛住大部分重复请求，然后对检索做优化，最后才考虑换模型。因为换模型会影响效果，需要重新评估。

5.4 几个我踩过的坑

坑一：忘记设置OpenAI的API Key。服务启动没问题，但一调用就报错。建议在启动时检查环境变量，没有就报错退出。

坑二：FAISS索引和chunks文件不匹配。我改了一次分块策略，重新生成了索引，但忘了重新生成chunks文件。结果检索出来的索引对不上文本。后来我改成把索引和chunks打包成一个文件，一起保存一起加载。

坑三：异步路由里用了同步的嵌入模型。sentence-transformers的encode方法是同步的，在async路由里调用会阻塞事件循环。解决办法是用run_in_executor把它放到线程池里跑。

坑四：没有做输入长度限制。用户输入了一个超长文本，嵌入模型直接报错。后来加了输入长度检查，超过限制就截断或者返回错误。

坑五：没有做输出长度限制。大模型有时候会输出很长的内容，导致响应时间过长。后来在提示词里加了“回答控制在200字以内”，问题就缓解了。

6. 后续扩展方向：从能用走向好用

6.1 评估体系的搭建

系统搭起来只是第一步，怎么知道它好不好用才是关键。我建议从两个维度做评估：

离线评估：准备一批测试问题和标准答案，跑一遍系统，计算准确率、召回率、F1值。这个可以自动化，每次改代码都跑一遍。

在线评估：在真实用户使用过程中收集反馈。最简单的做法是加一个“这个回答有帮助吗”的按钮，让用户点赞或者点踩。然后定期分析这些反馈，找出问题。

评估的关键是持续。不要只做一次评估就完事，要把它变成日常流程的一部分。

6.2 提示词版本管理

提示词是AI系统的核心资产，但很多人把它硬编码在代码里，改一次就要重新部署。我的做法是把提示词抽出来，用YAML文件管理，每个提示词有版本号。代码里通过版本号引用提示词。这样改提示词不需要改代码，也不需要重新部署。

更进一步，可以做A/B测试。同时跑两个版本的提示词，看哪个效果好。这需要服务层支持按比例分流。

6.3 从单机到分布式

单机系统有性能上限。当请求量大了，需要考虑分布式。最简单的做法是把服务层水平扩展，多开几个实例，前面加一个负载均衡。向量数据库也可以换成分布式的，比如Milvus集群。

但分布式会带来新的问题：数据一致性、服务发现、监控。这些都需要额外的工具和配置。我的建议是，不到万不得已不要上分布式。先把单机性能压榨到极限，再考虑扩展。

6.4 模型切换和降级

不要绑定在一个模型上。OpenAI挂了怎么办？API涨价了怎么办？你需要有一个备选方案。

我的做法是抽象一个模型接口，支持多个后端。主用OpenAI，备用本地Ollama。当OpenAI调用失败或者超时，自动切换到本地模型。虽然本地模型效果差一点，但至少服务不会挂。

这个切换逻辑可以放在服务层，对应用层透明。应用层只管调用模型接口，不关心底层用的是哪个模型。

6.5 监控和告警

生产系统必须有监控。需要监控的指标包括：请求量、响应时间、错误率、缓存命中率、模型调用次数。这些指标可以用Prometheus收集，用Grafana展示。

告警规则也很重要。比如错误率超过5%就告警，响应时间超过2秒就告警。告警渠道可以用邮件、钉钉、企业微信。

我自己的做法是先用最简单的日志监控，把关键指标打到日志里，然后用grep和awk分析。等系统稳定了，再上专业的监控工具。

6.6 安全性和合规性

AI系统有一些特有的安全问题。比如提示词注入：用户在问题里嵌入恶意指令，试图让模型输出不该输出的内容。防御方法是把用户输入和系统提示词严格分开，不要让用户输入直接拼接到系统提示词里。

还有数据泄露问题：如果向量数据库里存了敏感信息，检索的时候可能会泄露。解决办法是对敏感信息做脱敏处理，或者在检索层加权限控制。

合规性方面，要确保你的系统符合相关法律法规的要求。比如用户数据的收集和使用要获得授权，模型输出不能包含歧视性内容。这些需要在系统设计阶段就考虑进去，而不是事后补救。

6.7 持续迭代的心得

最后说一点个人体会。AI工程是一个快速变化的领域，今天的最佳实践明天可能就过时了。所以不要追求一步到位，要小步快跑，持续迭代。

我的做法是每周花半天时间回顾系统表现，找出最需要改进的一个点，然后花一周时间改进它。不要同时改多个地方，那样出了问题很难定位。

还有一点：不要过度优化。80%的效果来自20%的优化。先把那20%做好，剩下的20%效果可能需要80%的努力。除非你的业务对那20%有硬性要求，否则不值得。

我在实际使用中发现，最简单的方案往往是最有效的。复杂的架构看起来很厉害，但维护成本高，出问题的概率也大。从零开始搭建AI工程能力，最重要的不是学会多少工具，而是理解每一层在做什么、为什么这么做。理解了这些，你用什么工具都能搭出好系统。