1. 从零搭建AI工程能力：为什么我劝你别再当“调包侠”
这两年AI应用层的岗位需求翻了不知道多少倍，但真正能扛住生产环境考验的工程师却始终稀缺。我面过不少人，简历上写着“精通LangChain”“熟悉RAG”，一问底层怎么切分文档、向量检索召回率怎么评估、推理延迟卡在哪一环，就开始含糊其辞。这就是典型的“调包侠”困境——会用工具，但不知道工具为什么这么设计，出了问题只能靠重启和玄学调试。
ai-engineering-from-scratch这个项目标题，核心讲的不是某个具体框架的教程，而是一套从底层原理出发、逐步构建AI工程能力的完整路径。它要解决的问题很明确：让开发者不再依赖黑盒式的API调用，而是真正理解数据管道、模型推理、检索增强、评估体系这些环节是怎么串起来的。适合谁来参考？我认为有三类人最该认真看：一是刚转行做AI应用、只会调接口的初中级工程师；二是有传统后端经验、想补齐AI工程链路的开发者；三是技术负责人，需要判断团队的技术选型到底靠不靠谱。
我自己带过三个从零起步的AI项目，踩过的坑包括但不限于：文档切分粒度太粗导致检索答非所问、向量库选型不当导致内存爆炸、没有评估集导致每次迭代都像开盲盒。这些问题的根源，都是因为一开始跳过了“从零理解”这一步，直接上了高级封装。所以这篇博文，我会按照一个真实项目的推进节奏，把AI工程能力拆成可落地、可复现的模块，每个模块都讲清楚“为什么这么做”和“不这么做会怎样”。
2. 整体设计思路：把AI工程拆成四层能力栈
2.1 为什么不能一上来就写业务代码
很多人做AI项目的第一个动作是pip install openai，然后写个循环就开始跑。这种做法在Demo阶段没问题，但一旦数据量上来、需求变复杂，整个系统就会变成一团乱麻。我习惯把AI工程能力分成四层：数据层、模型层、检索层、评估层。这四层不是随便分的，而是对应了AI应用从输入到输出的完整生命周期。
数据层负责原始文档的采集、清洗、切分和结构化。模型层负责推理服务的封装、批处理、并发控制和降级策略。检索层负责向量化、索引构建、召回排序。评估层负责构建测试集、定义指标、自动化回归。这四层之间是依赖关系：数据层没做好，检索层再强也白搭；评估层缺失，模型层改了什么你根本不知道好坏。
我见过太多项目把80%的时间花在模型层调参上，结果数据层用的是随手复制的切分脚本，评估层完全空白。最后上线效果不稳定，排查方向都找不到。
2.2 技术选型的核心原则：可控性优先于先进性
选型的时候，我坚持一个原则：优先选你能看懂源码、能改、能排查的工具。比如向量库，Milvus、Qdrant、Chroma、FAISS我都用过。Chroma上手最快，但生产环境我倾向Qdrant或Milvus，原因是它们的持久化机制和过滤查询更成熟，出问题时有日志可查。再比如推理框架，vLLM的吞吐确实高，但如果你的场景是低频调用、对延迟不敏感，直接用HuggingFace的pipeline反而更省心。
这里有个常见的误区：很多人觉得“先进”就等于“适合”。实际上，一个需要你花两周才能跑通的框架，和一个半天就能跑通但性能差20%的框架，在项目早期后者往往更划算。因为早期最重要的是验证链路，而不是压榨性能。等链路跑通了，再针对瓶颈做替换，这才是合理的演进路径。
2.3 从零构建的路线图
我把整个构建过程分成五个阶段，每个阶段都有明确的交付物：
| 阶段 | 核心任务 | 交付物 | 预计耗时 |
|---|---|---|---|
| 第一阶段 | 数据管道搭建 | 可复现的文档切分脚本 | 2-3天 |
| 第二阶段 | 向量检索实现 | 可查询的向量索引 | 3-5天 |
| 第三阶段 | 推理服务封装 | 带降级的API服务 | 3-5天 |
| 第四阶段 | 评估体系建立 | 自动化评估脚本 | 2-3天 |
| 第五阶段 | 端到端联调 | 完整可演示系统 | 3-5天 |
这个路线图的关键在于：每个阶段都能独立验证。你不需要等所有东西都做完才知道对不对。比如数据管道做完，你可以直接检查切分后的文本块是否语义完整；向量检索做完，你可以手动输入几个问题看召回结果。这种“小步验证”的习惯，能帮你省下大量返工时间。
3. 核心细节解析：数据管道与检索层的实操要点
3.1 文档切分：别再用固定长度硬切了
文档切分是RAG系统的地基，但很多人直接用RecursiveCharacterTextSplitter配个chunk_size=1000就完事了。这种做法在技术文档上勉强能用，但遇到合同、论文、产品手册这类结构复杂的文本，就会把完整的语义单元切碎。我举个例子：一份采购合同里，“付款方式”和“违约责任”是两个独立条款，如果你按固定长度切，很可能把两个条款混在一个块里，检索时就会召回不相关的信息。
我的做法是按文档结构切分，再按语义合并。具体步骤：
- 先用解析器提取文档的标题层级（H1/H2/H3）和段落边界。
- 以最小语义单元（如一个条款、一个段落）为切分单位。
- 如果相邻单元属于同一父标题，且合并后长度不超过阈值（我一般设800-1200字符），就合并。
- 对超长单元，再按句子边界二次切分。
from langchain.text_splitter import MarkdownHeaderTextSplitter headers_to_split_on = [ ("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3"), ] splitter = MarkdownHeaderTextSplitter( headers_to_split_on=headers_to_split_on, strip_headers=False ) chunks = splitter.split_text(document)这样切出来的块，每个都带有完整的标题路径信息，检索时可以把标题路径作为元数据一起存入向量库，召回时就能做过滤。实测下来，这种切分方式在技术文档上的召回准确率比固定长度切分高出30%以上。
注意：切分阈值不是拍脑袋定的。你要根据嵌入模型的最大输入长度来倒推。比如你用的嵌入模型最大支持512个token，那切分后的块最好控制在400个token以内，留出余量。
3.2 向量化：模型选择与批处理策略
嵌入模型的选择直接决定了检索质量。我测试过OpenAI的text-embedding-3-small、BGE系列、以及开源的gte-large。结论是：中文场景下BGE-large-zh-v1.5性价比最高，英文场景text-embedding-3-small足够用。如果你的数据涉及专业领域（如医疗、法律），建议在领域语料上做微调，或者至少用领域数据评估一下现成模型的表现。
向量化过程中最容易忽略的是批处理。很多人一条一条调API，速度慢不说，还容易触发限流。正确的做法是批量发送，但要注意两个参数：batch_size和max_retries。我一般设batch_size=64，max_retries=3，并在每批之间加0.5秒延迟。如果是本地模型，直接用GPU批推理，吞吐能提升10倍以上。
from sentence_transformers import SentenceTransformer model = SentenceTransformer('BAAI/bge-large-zh-v1.5') embeddings = model.encode( texts, batch_size=64, show_progress_bar=True, normalize_embeddings=True )normalize_embeddings=True这个参数很关键，它把向量归一化到单位长度，这样后续用余弦相似度检索时，内积计算就等价于余弦相似度，能省一次开方运算。数据量大的时候，这点优化很可观。
3.3 索引构建：HNSW参数怎么调
向量索引的核心是平衡召回率和查询速度。目前主流的选择是HNSW（分层可导航小世界图）。Qdrant和Milvus都支持，参数主要有三个：m、ef_construct、ef_search。
m：每个节点的最大连接数。越大召回率越高，但内存占用也越大。我一般设16-32。ef_construct：构建时的候选集大小。越大索引质量越好，但构建越慢。我一般设100-200。ef_search：查询时的候选集大小。越大召回率越高，但查询越慢。我一般设64-128。
这三个参数没有绝对的最优值，要根据你的数据规模和延迟要求来调。我的经验是：先用默认值跑通，然后用一批标注好的查询-文档对来测召回率，逐步调整ef_search，直到召回率满足要求，再看延迟是否可接受。
一个容易踩的坑：索引构建完成后，如果你新增了文档，HNSW需要增量插入。频繁的小批量插入会导致图结构退化，召回率下降。建议积累到一定量（比如1000条）再批量插入，或者定期重建索引。
4. 实操过程：从零搭建一个可用的RAG系统
4.1 环境准备与依赖安装
我习惯用conda管理环境，因为AI相关的依赖版本冲突太常见了。以下是基础环境配置：
conda create -n ai-eng python=3.10 conda activate ai-eng pip install langchain==0.1.0 pip install qdrant-client==1.7.0 pip install sentence-transformers==2.3.0 pip install pypdf==4.0.0 pip install fastapi==0.109.0 pip install uvicorn==0.27.0这里我固定了版本号，因为LangChain的API变动非常频繁，不固定版本的话，今天跑通的代码明天可能就报错。Qdrant客户端也是，1.7.0是我实测比较稳定的版本。
如果你用GPU，还需要装对应版本的PyTorch。建议去PyTorch官网查好CUDA版本再装，别直接
pip install torch，很容易装成CPU版。
4.2 数据管道完整实现
假设我们有一批PDF格式的产品手册，需要构建检索系统。完整流程如下：
第一步：PDF解析与文本提取
from pypdf import PdfReader def extract_text_from_pdf(pdf_path): reader = PdfReader(pdf_path) full_text = [] for page_num, page in enumerate(reader.pages): text = page.extract_text() if text.strip(): full_text.append({ "page": page_num + 1, "content": text }) return full_text这里我保留了页码信息，因为后续检索时，用户可能想定位到具体页面。元数据在RAG系统里非常重要，不要只存文本内容。
第二步：按结构切分
def split_by_structure(text, max_chunk_size=1000): paragraphs = text.split("\n\n") chunks = [] current_chunk = "" for para in paragraphs: if len(current_chunk) + len(para) <= max_chunk_size: current_chunk += para + "\n\n" else: if current_chunk: chunks.append(current_chunk.strip()) current_chunk = para + "\n\n" if current_chunk: chunks.append(current_chunk.strip()) return chunks这个切分逻辑比固定长度切分更符合语义边界，因为它是按段落合并的。max_chunk_size设1000是经验值，你可以根据嵌入模型的能力调整。
第三步：向量化与入库
from qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams, PointStruct client = QdrantClient(path="./qdrant_data") client.recreate_collection( collection_name="product_manual", vectors_config=VectorParams( size=1024, distance=Distance.COSINE ) ) points = [] for idx, chunk in enumerate(chunks): vector = model.encode(chunk).tolist() points.append(PointStruct( id=idx, vector=vector, payload={"text": chunk, "source": "manual.pdf"} )) client.upsert(collection_name="product_manual", points=points)注意size=1024要和你的嵌入模型输出维度一致。BGE-large-zh-v1.5的输出维度就是1024。如果维度对不上，入库会直接报错。
4.3 检索服务封装
检索服务的核心是召回+重排。先用向量检索召回Top-K（比如20条），再用重排模型精排，取Top-N（比如5条）返回给生成模型。
from qdrant_client.models import Filter, FieldCondition, MatchValue def retrieve(query, top_k=20, top_n=5): query_vector = model.encode(query).tolist() results = client.search( collection_name="product_manual", query_vector=query_vector, limit=top_k ) # 简单重排：按分数排序，取前top_n reranked = sorted(results, key=lambda x: x.score, reverse=True)[:top_n] return [{"text": r.payload["text"], "score": r.score} for r in reranked]如果要做更精细的重排，可以引入bge-reranker模型。实测下来，重排能把Top-5的准确率再提升15%左右。但重排会增加延迟，所以要根据你的场景权衡。
4.4 推理服务与降级策略
推理服务我推荐用FastAPI封装，因为它的异步支持好，适合处理并发请求。关键是要加超时控制和降级策略。
from fastapi import FastAPI, HTTPException import asyncio app = FastAPI() async def call_llm(prompt, timeout=10): try: # 这里替换成你实际的LLM调用 result = await asyncio.wait_for(llm_call(prompt), timeout=timeout) return result except asyncio.TimeoutError: return "抱歉，当前请求较多，请稍后重试。" @app.post("/query") async def query_endpoint(query: str): contexts = retrieve(query) prompt = build_prompt(query, contexts) answer = await call_llm(prompt) return {"answer": answer, "sources": contexts}降级策略的意思是：当LLM调用超时或失败时，不要直接报错，而是返回一个兜底回复，或者只返回检索到的原文片段。这样用户体验不会断崖式下跌。
5. 常见问题与排查技巧实录
5.1 检索召回不准的排查思路
召回不准是最常见的问题，排查要按顺序来：
| 排查项 | 检查方法 | 常见原因 |
|---|---|---|
| 切分质量 | 人工检查切分后的文本块 | 语义被切碎或混入无关内容 |
| 嵌入模型 | 用标注数据测召回率 | 模型与领域不匹配 |
| 索引参数 | 调大ef_search看是否改善 | HNSW参数过于保守 |
| 查询改写 | 对比原始查询和改写后查询 | 用户查询太短或歧义 |
我的经验是，80%的召回问题出在切分环节。所以遇到召回不准，先别急着换模型，把切分后的文本块打印出来看看，往往问题一目了然。
5.2 推理延迟过高的优化手段
延迟高通常有三个来源：检索慢、LLM推理慢、网络传输慢。排查方法：
- 检索慢：看Qdrant的查询日志，如果单次查询超过100ms，考虑降低
ef_search或减少Top-K。 - LLM推理慢：如果是API调用，看是不是网络问题；如果是本地模型，看GPU利用率，如果利用率低，说明批处理没做好。
- 网络传输慢：把检索服务和推理服务部署在同一内网，减少跨网络调用。
我实测过一个优化案例：把ef_search从128降到64，召回率只掉了2%，但查询延迟从80ms降到了35ms。这种权衡在生产环境非常值得做。
5.3 评估体系怎么建才不流于形式
很多团队的评估就是找几个人手动问几个问题，看看回答对不对。这种做法不可复现，也无法量化。我的做法是建一个黄金测试集：
- 收集100-200个真实用户查询。
- 为每个查询标注正确的文档块ID。
- 定义指标：召回率（Recall@K）、准确率（Precision@K）、MRR。
- 每次迭代后自动跑一遍，对比指标变化。
def evaluate(test_set, retrieve_func, k=5): recalls = [] for query, correct_ids in test_set: results = retrieve_func(query, top_n=k) retrieved_ids = [r["id"] for r in results] hit = len(set(retrieved_ids) & set(correct_ids)) recalls.append(hit / len(correct_ids)) return sum(recalls) / len(recalls)这个评估脚本不到20行，但能帮你把迭代从“凭感觉”变成“看数据”。我强烈建议在项目早期就把这个建起来，哪怕测试集只有50条。
5.4 几个我踩过的坑
坑一：向量维度不匹配。有次我换了嵌入模型，忘了改Qdrant的size参数，结果入库全部失败。排查了半天才发现是维度问题。所以换模型时，一定要同步检查向量库配置。
坑二：元数据丢失。早期我只存了文本内容，没存来源和页码。后来用户问“这个信息在哪一页”，我完全答不上来。元数据一定要在切分阶段就带上，后面补很麻烦。
坑三：忽略并发写入。Qdrant支持并发写入，但如果多个进程同时写同一个collection，可能导致数据不一致。生产环境建议用单写入进程，或者加分布式锁。
坑四：评估集泄露。有次我把测试集里的查询也放进了训练数据，导致评估指标虚高。后来我严格分开训练集和测试集，确保没有重叠。
6. 从能跑到好用：进阶优化方向
6.1 查询改写与多路召回
用户输入的查询往往很短，比如“怎么退款”。这种查询直接拿去检索，召回效果很差。我的做法是先用LLM做查询改写，生成多个相关查询，然后多路召回、合并去重。
def rewrite_query(query): prompt = f"请将以下查询改写成3个语义相同但表达不同的查询，用换行分隔：\n{query}" rewritten = llm_call(prompt) return [query] + rewritten.strip().split("\n")多路召回的好处是能覆盖更多表达方式，提升召回率。代价是检索次数增加，延迟会上升。所以适合对召回率要求高、对延迟容忍度高的场景。
6.2 混合检索：向量+关键词
纯向量检索有个天然缺陷：对精确匹配不敏感。比如用户搜“型号X200”，向量检索可能召回“型号X100”的文档，因为语义相近。这时候就需要结合关键词检索（BM25）。
我的做法是：向量检索召回Top-20，BM25召回Top-20，然后用RRF（倒数排名融合）合并，取Top-10。实测下来，混合检索在包含专有名词的场景下，准确率比纯向量检索高出20%以上。
6.3 缓存策略：省下的都是真金白银
如果你的系统有高频重复查询，加一层缓存能大幅降低成本。我用Redis做查询缓存，key是查询的哈希值，value是检索结果。TTL设1小时，因为文档更新频率通常没那么高。
import hashlib import redis r = redis.Redis(host='localhost', port=6379) def cached_retrieve(query): key = hashlib.md5(query.encode()).hexdigest() cached = r.get(key) if cached: return json.loads(cached) result = retrieve(query) r.setex(key, 3600, json.dumps(result)) return result这个优化在客服场景特别有效，因为用户问来问去就是那些问题。我有个项目加了缓存后，LLM调用量直接降了40%。
6.4 监控与告警：上线只是开始
系统上线后，必须监控几个核心指标：检索延迟、LLM调用成功率、平均响应时间、缓存命中率。我用Prometheus+Grafana搭监控面板，设置告警阈值。比如检索延迟超过200ms就告警，LLM调用失败率超过5%就告警。
这些指标能帮你在用户投诉之前发现问题。我经历过一次Qdrant内存泄漏，就是因为监控到检索延迟持续上升，提前做了扩容，避免了服务中断。
7. 我个人在实际操作中的体会
带过几个从零到一的AI项目后，我最大的体会是：AI工程的核心竞争力不在模型，而在工程化能力。模型大家都能调，但数据管道是否健壮、检索是否精准、评估是否可复现、监控是否到位，这些才是拉开差距的地方。
另外，不要追求一步到位。我见过太多项目一开始就想搭一个“完美”的架构，结果三个月过去了还在选型。正确的做法是先跑通最小闭环，哪怕切分很粗糙、检索很简陋，只要端到端能跑，你就能拿到真实反馈，然后针对性优化。这种迭代速度，比任何架构设计都重要。
最后分享一个小技巧：每次改动检索或推理逻辑后，一定要跑一遍评估集。我习惯把评估脚本做成命令行工具，改完代码顺手跑一下，指标掉了立刻回滚。这个习惯帮我避免了好几次线上事故。
