1. 从零搭建AI工程能力为什么“手搓一遍”比调包更值钱这两年AI应用层的工具链成熟得吓人LangChain、LlamaIndex、各种Agent框架几乎把能封装的都封装了。打开文档三行代码就能跑一个RAG问答复制一段示例十分钟就能搭出一个能对话的机器人。表面上看门槛被拉到了地板上谁都能说自己“做过AI项目”。但真到了线上出问题的时候差距就出来了检索召回率突然掉了一半没人知道是切分策略的问题还是向量模型换了版本模型输出开始胡言乱语排查半天发现是上下文窗口被悄悄截断成本一个月翻了三倍翻遍代码也找不到token到底花在了哪里。这就是我特别想聊“ai-engineering-from-scratch”这个方向的原因。它不是一个具体的库或者框架而是一种做事的方式——把AI工程里那些被封装层藏起来的关键环节自己动手实现一遍。从文本切分、向量化、索引构建、检索排序到提示词组装、上下文管理、输出解析、评测回归每一块都亲手写一遍最小可用版本。做完这一轮你再回头看那些框架就不再是“黑盒调参”而是能一眼看穿它在哪一层做了什么取舍。这篇文章适合三类人一是刚入行做AI应用、只会调API但说不清底层原理的开发者二是想从传统后端/算法转AI工程、需要补齐工程链路的同学三是带团队的技术负责人想搞清楚自己项目里哪些环节是真正的风险点。我会按照一个完整的RAGAgent混合项目的思路把每个环节的设计考量、实操细节、踩坑经验都摊开讲。全程不依赖重型框架核心逻辑用Python手写能跑通、能复现、能直接抄作业。先说清楚一个前提手搓不是为了反对用框架而是为了在需要的时候有能力替换框架里的任何一块。框架是加速器但你得知道加速的是什么。下面进入正题。2. 整体架构设计与技术选型思路2.1 为什么选择“最小依赖”路线我见过太多项目一上来就pip install一堆框架结果环境依赖冲突、版本锁定困难、升级一次崩一片。更麻烦的是当你想改某个环节的行为时发现框架的抽象层根本不给你插手的口子。所以“from scratch”的第一个原则就是核心链路只依赖最基础的库。具体来说我的选型是这样的环节选型理由文本处理原生Python regex切分逻辑自己控制不引入额外抽象向量化sentence-transformers 或直接调API本地模型可控API方案省事按场景选向量存储numpy 本地文件 / faiss小规模用numpy足够大规模再上faiss检索排序自己实现BM25 向量余弦理解混合检索的本质提示词组装字符串模板 手动token计数不依赖模板引擎逻辑透明评测自己写指标函数准确率、召回率、MRR手算一遍这套组合的好处是任何一个环节出问题你都能在几十行代码里定位到。坏处是前期要多写一些代码但这点投入在调试阶段会加倍还回来。提示如果你的项目数据量在十万条以内numpy做向量检索完全够用。别一上来就上向量数据库多一个组件就多一个故障点。2.2 核心链路的模块划分我把整个系统拆成六个模块每个模块有明确的输入输出契约文档加载与清洗把各种格式的原始文档转成纯文本去掉页眉页脚、乱码、重复段落。文本切分按语义或固定长度切分成chunk这是最容易被低估的一步。向量化与索引把chunk转成向量建立可快速检索的索引结构。检索与重排给定query召回候选chunk再精排。上下文组装与生成把检索结果塞进提示词调模型生成答案。评测与回归用固定测试集衡量每一版改动的效果。这六个模块之间通过明确的数据结构传递比如chunk用dataclass定义包含id、文本、向量、元数据。这样任何一块都能单独替换、单独测试。2.3 一个容易被忽略的设计原则可观测性优先很多项目做到最后变成“玄学调参”根本原因是中间过程不可见。所以我在设计阶段就强制要求每个模块都要能dump中间结果。切分后的chunk要能导出成jsonl看检索的候选列表要能打印分数提示词组装后的完整文本要能保存。这些在调试期是救命稻草在上线后是排查问题的依据。我一般会在项目根目录建一个debug/文件夹每次跑流程时把关键中间态写进去用时间戳区分。别嫌麻烦等你遇到“为什么昨天还好今天就不行”的时候会感谢自己留了这些痕迹。3. 核心细节解析与实操要点3.1 文本切分决定检索质量的第一道关切分看起来简单实际上是最影响效果的一步。切太大检索出来的chunk包含太多无关信息模型容易被干扰切太小语义不完整检索命中率下降。我的经验是按语义边界切用长度做兜底。具体做法是先用段落分隔符连续换行切如果某段超过阈值比如500字符再按句子切句子还超就按逗号切最后才硬切。这样能最大程度保留语义完整性。import re from dataclasses import dataclass, field dataclass class Chunk: id: str text: str metadata: dict field(default_factorydict) def split_text(text, max_len500, overlap50): # 先按段落切 paragraphs re.split(r\n\s*\n, text) chunks [] buffer for para in paragraphs: para para.strip() if not para: continue if len(buffer) len(para) max_len: buffer (\n if buffer else ) para else: if buffer: chunks.append(buffer) # 段落本身超长按句子切 if len(para) max_len: sentences re.split(r(?[。.!?])\s*, para) sub for s in sentences: if len(sub) len(s) max_len: sub s else: if sub: chunks.append(sub) sub s if sub: chunks.append(sub) buffer else: buffer para if buffer: chunks.append(buffer) # 加overlap final [] for i, c in enumerate(chunks): if i 0 and overlap 0: c chunks[i-1][-overlap:] c final.append(Chunk(idfchunk_{i}, textc)) return final这段代码有几个细节值得说。第一overlap的作用是防止关键信息正好落在切分边界上被割裂但overlap太大会导致重复内容多、检索冗余50字符左右是个经验值。第二中文和英文的句子边界不同正则里要同时覆盖中英文标点。第三切分后的chunk要保留原始位置信息比如来自哪个文档、第几段方便后续溯源。注意如果你的文档里有大量表格或代码块纯文本切分会破坏结构。这种情况建议先把表格转成markdown格式再切或者单独处理。3.2 向量化模型选择与批量处理向量化这一步核心决策是“用本地模型还是调API”。本地模型如bge、m3e系列的好处是免费、可控、无网络依赖坏处是占显存、首次加载慢。API方案省事但按量计费、有延迟、数据要出本地。我的建议是开发调试阶段用本地小模型上线根据数据敏感度和成本决定。本地模型用sentence-transformers加载几行代码就能跑from sentence_transformers import SentenceTransformer import numpy as np model SentenceTransformer(BAAI/bge-small-zh-v1.5) def embed_chunks(chunks, batch_size32): texts [c.text for c in chunks] vectors model.encode( texts, batch_sizebatch_size, normalize_embeddingsTrue, # 归一化后余弦相似度点积 show_progress_barTrue ) for c, v in zip(chunks, vectors): c.metadata[vector] v return chunks这里normalize_embeddingsTrue是个关键设置。归一化之后两个向量的余弦相似度就等于点积检索时直接做矩阵乘法速度快很多。另外batch_size要根据显存调太小了慢太大了OOM32或64是常见起点。还有一个坑query和document要用同一个模型编码而且有些模型对query和document有不同的前缀要求。比如bge系列query前面要加“为这个句子生成表示以用于检索相关文章”document不用。这个细节不注意检索效果会差一大截。3.3 检索混合检索比单一向量检索稳得多纯向量检索的问题是它对关键词匹配不敏感。比如用户搜一个专有名词向量模型可能把它编码到语义相近但实际不相关的区域。这时候BM25这种基于词频的检索就能补上。所以我的做法是向量检索和BM25各召回一批然后融合排序。BM25的实现不复杂核心是计算词频和逆文档频率import math from collections import Counter class BM25: def __init__(self, chunks, k11.5, b0.75): self.chunks chunks self.k1 k1 self.b b self.doc_len [len(c.text) for c in chunks] self.avg_len sum(self.doc_len) / len(self.doc_len) self.doc_freqs [] self.idf {} self._build() def _tokenize(self, text): # 简单按字符和空格切中文场景可换jieba return list(text) def _build(self): df Counter() for c in self.chunks: tokens set(self._tokenize(c.text)) for t in tokens: df[t] 1 N len(self.chunks) for t, freq in df.items(): self.idf[t] math.log((N - freq 0.5) / (freq 0.5) 1) for c in self.chunks: self.doc_freqs.append(Counter(self._tokenize(c.text))) def score(self, query, idx): tokens self._tokenize(query) score 0 for t in tokens: if t not in self.idf: continue tf self.doc_freqs[idx].get(t, 0) denom tf self.k1 * (1 - self.b self.b * self.doc_len[idx] / self.avg_len) score self.idf[t] * tf * (self.k1 1) / denom return score def search(self, query, top_k10): scores [(self.score(query, i), i) for i in range(len(self.chunks))] scores.sort(reverseTrue) return scores[:top_k]融合排序用RRFReciprocal Rank Fusion最简单有效不需要调权重def rrf_fusion(rank_lists, k60): scores {} for ranks in rank_lists: for rank, idx in enumerate(ranks): scores[idx] scores.get(idx, 0) 1 / (k rank 1) return sorted(scores.items(), keylambda x: -x[1])RRF的好处是只看排名不看分数避免了不同检索器分数量纲不一致的问题。k取60是论文里的经验值实际用50到100都行。3.4 上下文组装token预算要精打细算检索回来一堆chunk不能全塞进提示词得按token预算裁剪。我的做法是先按相关性排序从高到低往上下文里塞直到接近预算上限。预算怎么定模型上下文窗口减去提示词模板长度、再减去预留的输出长度剩下的就是检索内容的额度。def build_context(query, ranked_chunks, max_tokens3000): context_parts [] used 0 for idx, score in ranked_chunks: chunk chunks[idx] # 粗略估算中文1字符≈1token英文4字符≈1token est len(chunk.text) if used est max_tokens: break context_parts.append(f[来源{idx}]\n{chunk.text}) used est return \n\n.join(context_parts)这里加[来源X]标记是为了让模型能引用出处也方便后续做归因分析。token估算用字符数近似就够了真要精确可以用tiktoken但会增加依赖。实操心得上下文不是塞得越满越好。我实测下来塞到预算的70%左右效果最稳塞满了反而容易让模型忽略中间部分的内容所谓的“lost in the middle”现象。4. 完整实操流程与关键环节实现4.1 环境准备与依赖安装先把基础环境搭起来。我习惯用conda建独立环境避免污染系统Pythonconda create -n ai-scratch python3.10 -y conda activate ai-scratch pip install numpy sentence-transformers jieba tqdm如果要用faiss做大规模检索再加pip install faiss-cpu。GPU版本按官方文档装。整个依赖列表控制在5个以内保持轻量。4.2 数据加载与清洗的实操细节假设我们处理一批markdown文档。加载逻辑要处理编码问题、去除无关内容import os import re def load_documents(root_dir): docs [] for dirpath, _, filenames in os.walk(root_dir): for fn in filenames: if not fn.endswith((.md, .txt)): continue path os.path.join(dirpath, fn) with open(path, r, encodingutf-8, errorsignore) as f: raw f.read() # 去掉代码块可选看场景 # raw re.sub(r.*?, , raw, flagsre.S) # 去掉多余空行 raw re.sub(r\n{3,}, \n\n, raw) docs.append({path: path, text: raw.strip()}) return docs清洗这一步的取舍要看场景。如果是技术文档问答代码块要保留如果是通用知识问答代码块可能是噪声。我一般会保留原始文本在切分阶段再决定是否过滤。4.3 索引构建与持久化把切分、向量化、索引构建串起来import pickle def build_index(root_dir, index_pathindex.pkl): docs load_documents(root_dir) all_chunks [] for doc in docs: chunks split_text(doc[text]) for c in chunks: c.metadata[source] doc[path] all_chunks.extend(chunks) print(f共切分 {len(all_chunks)} 个chunk) all_chunks embed_chunks(all_chunks) # 构建向量矩阵 vectors np.array([c.metadata[vector] for c in all_chunks]) bm25 BM25(all_chunks) index { chunks: all_chunks, vectors: vectors, bm25: bm25 } with open(index_path, wb) as f: pickle.dump(index, f) return index持久化用pickle最简单但要注意版本兼容。如果chunk数量大pickle加载会慢可以改用numpy的npz存向量、jsonl存文本分开管理。4.4 检索与生成的完整链路把前面所有模块串成一个完整的问答函数def retrieve(query, index, top_k5): chunks index[chunks] vectors index[vectors] # 向量检索 q_vec model.encode([query], normalize_embeddingsTrue)[0] sims vectors q_vec vec_ranks np.argsort(-sims)[:top_k*2].tolist() # BM25检索 bm25_results index[bm25].search(query, top_ktop_k*2) bm25_ranks [idx for _, idx in bm25_results] # 融合 fused rrf_fusion([vec_ranks, bm25_ranks]) return fused[:top_k] def answer(query, index): ranked retrieve(query, index) context build_context(query, ranked) prompt f基于以下资料回答问题如果资料中没有相关信息请明确说明。 资料 {context} 问题{query} 回答 # 这里替换成你实际使用的模型调用 # response llm.generate(prompt) return prompt, ranked注意answer函数返回了prompt和ranked方便调试时检查。实际调用模型的部分我留空了因为不同环境用的模型不一样但前面的检索链路是通用的。4.5 评测环节没有评测就没有优化很多人做完上面几步就上线了结果效果好坏全凭感觉。我强烈建议建一个小的评测集哪怕只有二三十条问答对。评测指标至少看三个指标含义计算方式Hit RateK前K个结果里有没有正确答案命中数/总数MRR正确答案排名的倒数和平均1/rank答案准确率生成答案是否正确人工或模型打分def eval_retrieval(test_cases, index, k5): hits 0 mrr 0 for query, gold_chunk_id in test_cases: ranked retrieve(query, index, top_kk) ids [index[chunks][idx].id for idx, _ in ranked] if gold_chunk_id in ids: hits 1 rank ids.index(gold_chunk_id) 1 mrr 1 / rank n len(test_cases) return {hit_rate: hits/n, mrr: mrr/n}这个评测函数虽然简单但能让你在每次改动切分策略、换向量模型、调检索参数时立刻知道是变好还是变坏。没有它所有优化都是盲猜。5. 常见问题与排查技巧实录5.1 检索效果差的排查顺序遇到“答非所问”按这个顺序查先看切分把检索到的chunk打印出来看内容是否完整。如果chunk被切得七零八落先调切分参数。再看向量模型用几个典型query手动算相似度看排序是否符合直觉。如果明显不对可能是模型不适合中文或领域不匹配。然后看融合把向量检索和BM25的结果分别打印看融合后是否把好的结果挤掉了。最后看上下文组装确认塞进提示词的chunk顺序和数量是否合理。我踩过最坑的一次是切分时overlap设成了200导致每个chunk都包含大量重复内容检索出来看着都对但实际信息密度极低。后来把overlap降到50效果立刻好转。5.2 常见问题速查表现象可能原因解决方向检索结果完全不相关向量模型不匹配 / query前缀缺失换模型 / 加正确前缀相关结果排在后位单一检索召回不足加BM25混合检索答案包含无关信息上下文塞太多降低token预算 / 提高检索阈值答案遗漏关键信息chunk被切断增大overlap / 调整切分粒度相同query结果不稳定模型随机性 / 索引未固定固定随机种子 / 检查索引版本成本异常高上下文过长 / 重复调用加缓存 / 压缩上下文5.3 几个独家避坑技巧技巧一给chunk加“邻居扩展”。检索到某个chunk后把它前后相邻的chunk也带上能有效缓解切分导致的信息割裂。实现很简单在chunk元数据里存前后id检索后扩展。技巧二query改写。用户的问题往往口语化直接检索效果差。可以用一个小模型把query改写成更规范的检索式或者生成多个变体分别检索再融合。这一步对效果提升很明显但会增加一次模型调用。技巧三缓存检索结果。相同或相似的query没必要重复走完整链路。用query的向量做key相似度超过阈值就命中缓存。注意缓存要设过期时间文档更新后要失效。技巧四日志要记全。每次请求记录query、检索到的chunk id、最终prompt、模型输出、耗时。出问题时这些日志就是破案线索。我一般用jsonl按天存方便后续分析。6. 从手搓到上生产的扩展思路把最小版本跑通之后往生产环境走还有几块要补。第一是并发和性能向量检索用numpy是单线程的数据量大了要换faiss的IVF索引或者上专门的向量库。第二是增量更新文档会变索引要能局部更新而不是全量重建。第三是权限控制不同用户能检索的文档范围不同这个要在检索层做过滤。第四是监控告警检索命中率、响应延迟、token消耗这些指标要持续盯着。但我想说的是这些扩展都应该建立在“你已经手搓过一遍核心链路”的基础上。因为只有亲手实现过你才知道每个环节的瓶颈在哪、哪些优化是真正有效的、哪些是过度设计。框架能帮你省掉重复劳动但省不掉理解成本。我见过太多团队在没搞懂检索原理的情况下盲目换向量库、调参数最后钱花了不少效果原地踏步。最后分享一个我自己的习惯每做完一个AI工程项目我都会把核心链路的代码精简成一个单文件版本去掉所有业务逻辑只保留最本质的流程。这个文件既是我的参考实现也是下次新项目的起点。ai-engineering-from-scratch的价值不在于“不用框架”而在于你随时有能力看穿框架、替换框架、甚至在必要时自己写一个更合适的。这种底气是调包调不出来的。