从零搭建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 应用层业务逻辑的组装应用层是把前面三层串起来的地方。以问答系统为例流程是这样的接收用户问题把问题向量化从向量数据库检索相关文本块组装提示词调用大模型解析输出返回结果每一步都可能出错所以每一步都要有日志。我的做法是在每一步前后都打日志记录输入、输出、耗时。这样出了问题可以快速定位。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, encodingutf-8) as f: text f.read() documents.append(text) # 简单的递归分块 def chunk_text(text, chunk_size256, overlap50): 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_barTrue) 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_pickleTrue) # 设置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_modelQueryResponse) 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( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], temperature0.1 ) answer response.choices[0].message.content return QueryResponse(answeranswer, 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_k3到5个。我一般用3个因为大模型的上下文窗口有限塞太多反而会稀释关键信息。如果你的文档很长可以先用检索召回10个再用重排序选出3个。相似度阈值这个需要根据你的数据调。我的做法是先跑一批测试问题看正确结果的相似度分布然后取一个能过滤掉大部分错误结果的值。一般0.6到0.8之间。5. 常见问题与排查技巧实录5.1 检索结果不相关怎么办这是最常见的问题。可能的原因和排查步骤问题现象可能原因排查方法解决方案检索结果完全不相关嵌入模型不适合你的领域用几个典型问题测试看相似度分数换一个在你的领域上表现更好的嵌入模型检索结果部分相关分块策略有问题检查检索出来的块是否语义完整调整分块大小和重叠长度相似度分数普遍偏低向量没有归一化检查是否做了L2归一化在向量化后加faiss.normalize_L2相同问题每次结果不同索引没有持久化检查是否每次重启都重建索引把索引保存到磁盘启动时加载我的经验是90%的检索问题都出在分块上。分块没分好后面怎么调都没用。所以先把分块做好再调其他参数。5.2 大模型输出不稳定怎么办大模型的输出确实有随机性。降低随机性的方法降低temperature设成0.1或者0。但注意有些模型在temperature0的时候反而会输出重复内容。固定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工程能力最重要的不是学会多少工具而是理解每一层在做什么、为什么这么做。理解了这些你用什么工具都能搭出好系统。