从零构建AI工程能力：告别调包侠，掌握RAG与向量检索核心

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_size1000就完事了。这种做法在技术文档上勉强能用但遇到合同、论文、产品手册这类结构复杂的文本就会把完整的语义单元切碎。我举个例子一份采购合同里“付款方式”和“违约责任”是两个独立条款如果你按固定长度切很可能把两个条款混在一个块里检索时就会召回不相关的信息。我的做法是按文档结构切分再按语义合并。具体步骤先用解析器提取文档的标题层级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_onheaders_to_split_on, strip_headersFalse ) 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_size64max_retries3并在每批之间加0.5秒延迟。如果是本地模型直接用GPU批推理吞吐能提升10倍以上。from sentence_transformers import SentenceTransformer model SentenceTransformer(BAAI/bge-large-zh-v1.5) embeddings model.encode( texts, batch_size64, show_progress_barTrue, normalize_embeddingsTrue )normalize_embeddingsTrue这个参数很关键它把向量归一化到单位长度这样后续用余弦相似度检索时内积计算就等价于余弦相似度能省一次开方运算。数据量大的时候这点优化很可观。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 python3.10 conda activate ai-eng pip install langchain0.1.0 pip install qdrant-client1.7.0 pip install sentence-transformers2.3.0 pip install pypdf4.0.0 pip install fastapi0.109.0 pip install uvicorn0.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_size1000): 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_nameproduct_manual, vectors_configVectorParams( size1024, distanceDistance.COSINE ) ) points [] for idx, chunk in enumerate(chunks): vector model.encode(chunk).tolist() points.append(PointStruct( ididx, vectorvector, payload{text: chunk, source: manual.pdf} )) client.upsert(collection_nameproduct_manual, pointspoints)注意size1024要和你的嵌入模型输出维度一致。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_k20, top_n5): query_vector model.encode(query).tolist() results client.search( collection_nameproduct_manual, query_vectorquery_vector, limittop_k ) # 简单重排按分数排序取前top_n reranked sorted(results, keylambda x: x.score, reverseTrue)[: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, timeout10): try: # 这里替换成你实际的LLM调用 result await asyncio.wait_for(llm_call(prompt), timeouttimeout) 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。定义指标召回率RecallK、准确率PrecisionK、MRR。每次迭代后自动跑一遍对比指标变化。def evaluate(test_set, retrieve_func, k5): recalls [] for query, correct_ids in test_set: results retrieve_func(query, top_nk) 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-20BM25召回Top-20然后用RRF倒数排名融合合并取Top-10。实测下来混合检索在包含专有名词的场景下准确率比纯向量检索高出20%以上。6.3 缓存策略省下的都是真金白银如果你的系统有高频重复查询加一层缓存能大幅降低成本。我用Redis做查询缓存key是查询的哈希值value是检索结果。TTL设1小时因为文档更新频率通常没那么高。import hashlib import redis r redis.Redis(hostlocalhost, port6379) 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调用成功率、平均响应时间、缓存命中率。我用PrometheusGrafana搭监控面板设置告警阈值。比如检索延迟超过200ms就告警LLM调用失败率超过5%就告警。这些指标能帮你在用户投诉之前发现问题。我经历过一次Qdrant内存泄漏就是因为监控到检索延迟持续上升提前做了扩容避免了服务中断。7. 我个人在实际操作中的体会带过几个从零到一的AI项目后我最大的体会是AI工程的核心竞争力不在模型而在工程化能力。模型大家都能调但数据管道是否健壮、检索是否精准、评估是否可复现、监控是否到位这些才是拉开差距的地方。另外不要追求一步到位。我见过太多项目一开始就想搭一个“完美”的架构结果三个月过去了还在选型。正确的做法是先跑通最小闭环哪怕切分很粗糙、检索很简陋只要端到端能跑你就能拿到真实反馈然后针对性优化。这种迭代速度比任何架构设计都重要。最后分享一个小技巧每次改动检索或推理逻辑后一定要跑一遍评估集。我习惯把评估脚本做成命令行工具改完代码顺手跑一下指标掉了立刻回滚。这个习惯帮我避免了好几次线上事故。