1. 从零手搓AI工程为什么我不建议你直接调包1.1 一个让我彻底改变想法的真实案例去年帮一个朋友排查线上问题场景很典型一个基于开源框架搭的RAG问答服务平时跑得好好的某天突然开始返回大量无关内容。团队里几个工程师查了两天换了向量模型、调了检索参数、甚至重写了提示词问题依旧。最后我让他们把整个链路的数据流打印出来才发现是文本分块环节的一个边界条件——当文档段落恰好落在分块边界时元数据被截断了导致检索时无法正确过滤来源。这个问题如果对整条流水线的每个环节都有“从零实现过”的认知半小时就能定位。这件事让我更加坚定一个判断AI工程不是调包工程。你可以用LangChain、LlamaIndex、各种向量数据库但如果你没有亲手实现过一遍核心组件你就不具备排查问题的能力更不具备做技术选型的能力。“ai-engineering-from-scratch”这个项目标题核心就是一件事不依赖高层框架用最基础的库把AI工程的核心组件一个个搭出来。它解决的不是“怎么快速上线一个Demo”而是“怎么真正理解并掌控你的AI系统”。适合谁看适合那些已经会用现成框架、但遇到问题就抓瞎的工程师适合想从传统后端转AI工程、但不想只做个API调用侠的开发者也适合技术负责人你需要知道每个技术选型背后的代价是什么。1.2 从零实现到底“零”到什么程度先明确边界。这里的“从零”不是让你用C语言写矩阵乘法也不是让你手推反向传播。那是造轮子不是做工程。我定义的“从零”是不依赖任何AI应用层框架只使用基础数值计算库和标准库把AI工程中每个核心环节的最小可用版本实现出来。具体来说以下这些东西我会带你一个个手搓文本分块器不用框架的TextSplitter自己实现基于语义和固定窗口的分块逻辑向量化流水线直接调用模型API或本地模型自己管理批处理、重试、缓存向量索引与检索不依赖向量数据库用NumPy实现暴力检索和IVF索引提示词模板引擎自己实现变量替换、条件渲染、少样本示例管理对话状态管理自己维护多轮对话的上下文窗口和摘要压缩评估与监控自己实现检索命中率、生成质量评分的计算逻辑这些东西加起来代码量并不大但每一个都逼你面对真实工程中的边界条件。比如分块框架帮你处理了大部分情况但你的文档格式千奇百怪时你就需要自己写规则。比如检索向量数据库帮你封装了ANN算法但你不理解IVF和HNSW的区别就没法根据数据规模做选型。我个人的经验是手搓一遍之后再用框架你会从“照着文档抄”变成“知道它在干什么以及它哪里可能出问题”。2. 核心组件拆解每个环节到底在解决什么问题2.1 文本分块不是切字符串那么简单文本分块是RAG系统的第一道工序也是最容易被低估的环节。很多人觉得分块就是按固定长度切实际上分块策略直接决定了检索质量的上限。我手搓分块器时核心考虑三个维度块大小、重叠窗口、语义边界。块大小决定了检索的粒度——太小则信息不完整太大则噪声多。重叠窗口是为了避免关键信息恰好被切断。语义边界则是尽量在段落、句子结束处切分而不是在词中间切。具体实现上我采用分层策略先按文档结构标题、段落做粗切再对超长段落做固定窗口细切最后对每个块做元数据标注来源、位置、层级。这里有个关键细节元数据必须和文本一起存储和检索否则后续无法做来源过滤和引用追溯。def semantic_chunk(text, max_chunk_size512, overlap64): # 先按段落切分 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()) # 超长段落做窗口切分 if len(para) max_chunk_size: for i in range(0, len(para), max_chunk_size - overlap): chunks.append(para[i:i max_chunk_size]) current_chunk else: current_chunk para \n\n if current_chunk: chunks.append(current_chunk.strip()) return chunks这段代码看起来简单但实际使用时你会发现很多坑。比如中文文档没有空格按字符切分效果很差比如代码块被切断后语义完全丢失比如表格数据切分后行列关系断裂。这些都需要根据你的数据类型做针对性处理。实操心得分块大小没有万能值。我的经验是问答类场景用256-512字符摘要类场景用1024-2048字符。但一定要用你的真实数据做A/B测试看检索命中率的变化。2.2 向量化流水线批处理、重试与缓存一个都不能少向量化就是把文本变成向量听起来就是调个API的事。但工程上你需要处理的问题包括API限流、网络超时、批量大小优化、失败重试、结果缓存、成本控制。我手搓的向量化流水线包含以下核心逻辑批处理把待向量化的文本按批次发送批次大小根据API限制和延迟做权衡。太大容易触发限流太小则吞吐低。指数退避重试遇到限流或临时错误时按指数退避策略重试避免雪崩。本地缓存对已经向量化过的文本做哈希缓存避免重复计算。这在开发调试阶段能省大量成本。并发控制用线程池或异步IO控制并发数既要压满带宽又不能把API打挂。import hashlib import time from concurrent.futures import ThreadPoolExecutor class Vectorizer: def __init__(self, embed_fn, cacheNone, max_workers4): self.embed_fn embed_fn self.cache cache or {} self.max_workers max_workers def _hash(self, text): return hashlib.md5(text.encode()).hexdigest() def embed_batch(self, texts, batch_size32, max_retries3): results [None] * len(texts) to_embed [] indices [] for i, text in enumerate(texts): key self._hash(text) if key in self.cache: results[i] self.cache[key] else: to_embed.append(text) indices.append(i) for start in range(0, len(to_embed), batch_size): batch to_embed[start:start batch_size] batch_indices indices[start:start batch_size] for attempt in range(max_retries): try: vectors self.embed_fn(batch) for idx, vec, text in zip(batch_indices, vectors, batch): results[idx] vec self.cache[self._hash(text)] vec break except Exception as e: if attempt max_retries - 1: raise time.sleep(2 ** attempt) return results这里的关键设计是缓存层。很多人忽略这一点结果每次调试都重新向量化整个知识库既慢又贵。缓存用文本哈希做键简单有效。如果文本有微小改动哈希会变这是合理的——因为向量确实需要重新计算。注意事项缓存要考虑持久化。内存缓存重启就没了生产环境建议用Redis或本地文件做持久化。另外如果换了向量模型缓存必须全部失效否则向量空间不一致会导致检索完全错乱。2.3 向量检索暴力检索、IVF与HNSW的取舍向量检索是RAG的核心。框架通常帮你封装了向量数据库但你不理解背后的索引结构就没法做性能调优。我手搓了三种检索方式分别对应不同数据规模暴力检索把所有向量加载到内存查询时计算查询向量与所有向量的相似度排序返回Top-K。实现最简单召回率100%但数据量大时延迟高。适合数据量小于10万条的场景。IVF索引先用K-Means把向量聚成N个簇查询时只搜索与查询向量最近的几个簇。牺牲少量召回率换取大幅速度提升。适合百万级数据。HNSW索引构建多层图结构查询时在图中做贪心搜索。速度最快召回率也高但内存占用大构建时间长。适合千万级数据且对延迟敏感的场景。import numpy as np class BruteForceIndex: def __init__(self, dim): self.vectors [] self.metadata [] def add(self, vector, meta): self.vectors.append(vector) self.metadata.append(meta) def search(self, query_vector, top_k5): if not self.vectors: return [] matrix np.array(self.vectors) query np.array(query_vector) # 余弦相似度 norms np.linalg.norm(matrix, axis1) * np.linalg.norm(query) similarities matrix query / (norms 1e-10) top_indices np.argsort(similarities)[::-1][:top_k] return [(self.metadata[i], similarities[i]) for i in top_indices]暴力检索的代码就这么简单但它是理解所有ANN算法的基础。IVF和HNSW本质上都是在“用少量精度换大量速度”。我建议你先用暴力检索跑通全流程确认检索质量达标后再根据数据规模决定是否引入ANN索引。踩过的坑余弦相似度和内积在向量归一化后是等价的。很多向量模型输出的向量已经归一化了这时候用内积计算更快。但如果你不确定就老老实实做归一化别为了省一点计算量引入bug。2.4 提示词模板引擎变量、条件与少样本管理提示词工程不是写字符串而是管理一套模板系统。我手搓的模板引擎支持以下功能变量替换用{{variable}}语法做占位符替换条件渲染根据上下文决定是否包含某个片段少样本示例管理动态选择最相关的示例注入提示词Token预算控制自动截断超长上下文import re class PromptTemplate: def __init__(self, template): self.template template self.variables set(re.findall(r\{\{(\w)\}\}, template)) def render(self, **kwargs): missing self.variables - set(kwargs.keys()) if missing: raise ValueError(fMissing variables: {missing}) result self.template for key, value in kwargs.items(): result result.replace(f{{{{{key}}}}}, str(value)) return result def render_with_condition(self, condition_var, condition_value, **kwargs): # 简单条件渲染如果condition_var等于condition_value保留条件块 pattern r\{\{#if condition_var r\}\}(.*?)\{\{/if\}\} def replacer(match): if kwargs.get(condition_var) condition_value: return match.group(1) return template re.sub(pattern, replacer, self.template, flagsre.DOTALL) return self.render(**kwargs)模板引擎的核心价值在于可维护性。当你有几十个提示词模板时硬编码字符串会让你疯掉。用模板系统你可以统一管理变量、做版本控制、做A/B测试。实操心得少样本示例的选择很关键。我的做法是先用检索找到与当前问题最相似的几个历史问答对把它们作为示例注入提示词。这比固定示例效果好得多但要注意Token预算别把上下文撑爆了。3. 完整实操从零搭建一个可用的RAG问答系统3.1 环境准备与依赖选择我选择的技术栈非常克制Python 3.10基础运行环境NumPy向量计算requests调用模型API标准库json、hashlib、concurrent.futures等不引入LangChain、不引入向量数据库、不引入Web框架。整个系统就是一个Python包可以通过命令行或简单HTTP服务调用。为什么这么克制因为依赖越少可控性越强。每引入一个依赖你就多了一个可能出问题的环节。而且当你手搓过一遍之后再用框架你会知道框架帮你做了什么以及它可能在哪些地方埋坑。pip install numpy requests就这两行。如果你要用本地模型做向量化再加一个sentence-transformers。但为了演示清晰我用API方式。3.2 数据准备与分块实操假设我们有一个Markdown格式的知识库包含多个文档。第一步是加载和分块。import os import json def load_documents(doc_dir): docs [] for filename in os.listdir(doc_dir): if filename.endswith(.md): with open(os.path.join(doc_dir, filename), r, encodingutf-8) as f: content f.read() docs.append({ source: filename, content: content }) return docs def chunk_documents(docs, max_chunk_size512, overlap64): all_chunks [] for doc in docs: chunks semantic_chunk(doc[content], max_chunk_size, overlap) for i, chunk in enumerate(chunks): all_chunks.append({ source: doc[source], chunk_index: i, text: chunk }) return all_chunks分块完成后我建议先人工抽查几个块看看切分是否合理。特别是检查有没有把关键信息切断有没有产生大量无意义的碎片。注意事项分块后一定要保留元数据。我见过太多人只存文本结果检索到内容后无法追溯来源也无法做引用。元数据至少包含来源文件名、块序号、原始位置。3.3 向量化与索引构建接下来把分块后的文本向量化并构建索引。def build_index(chunks, embed_fn): vectorizer Vectorizer(embed_fn) texts [c[text] for c in chunks] vectors vectorizer.embed_batch(texts, batch_size32) index BruteForceIndex(dimlen(vectors[0])) for chunk, vector in zip(chunks, vectors): index.add(vector, chunk) return index, vectorizer这里有个细节向量维度。不同模型输出的维度不同常见的有384、768、1536。维度越高表达能力越强但存储和计算成本也越高。我一般用768维平衡效果和成本。索引构建完成后建议做一次自检随机选几个块用它们的向量去检索看能不能检索到自己。如果检索不到说明向量化或索引有问题。3.4 检索与生成全流程串联最后把检索和生成串起来形成一个完整的问答流程。def answer_question(question, index, vectorizer, llm_fn, top_k5): # 1. 向量化问题 query_vector vectorizer.embed_batch([question])[0] # 2. 检索相关块 results index.search(query_vector, top_ktop_k) # 3. 构建上下文 context \n\n.join([r[0][text] for r in results]) # 4. 构建提示词 prompt f基于以下上下文回答问题。如果上下文不包含答案请明确说明。 上下文 {context} 问题{question} 回答 # 5. 调用LLM生成 answer llm_fn(prompt) return { answer: answer, sources: [r[0][source] for r in results], scores: [r[1] for r in results] }这个流程看起来简单但每个环节都有优化空间。比如检索时可以做重排序用更精细的模型对Top-K结果重新打分比如生成时可以注入对话历史支持多轮问答比如可以加一个“不知道”的兜底逻辑当检索分数低于阈值时直接返回“无法回答”。实操心得检索分数阈值很重要。我一般设0.7左右低于这个值就认为没有相关内容。但阈值需要根据你的数据和模型做调整不能照搬。4. 常见问题与排查技巧实录4.1 检索质量差从分块到模型逐层排查检索质量差是最常见的问题。我的排查顺序是检查分块随机抽几个块看内容是否完整、语义是否清晰。如果块太碎或太大调整分块参数。检查向量模型用几个明显相关的文本对看它们的向量相似度是否高。如果模型本身不行换模型。检查检索算法如果是暴力检索召回率应该是100%问题不在检索算法。如果是ANN索引检查召回率是否达标。检查查询改写用户的问题可能和文档表述不一致。可以加一个查询改写步骤用LLM把用户问题改写成更适合检索的形式。下面这个表格是我总结的常见症状和对应解法症状可能原因排查方法解决方案检索结果完全不相关向量模型不匹配检查模型是否支持中文换用多语言模型检索结果部分相关分块粒度不当抽查分块质量调整块大小和重叠相关结果排在后位相似度计算问题检查向量是否归一化统一做归一化检索不到已知答案索引未包含该内容检查索引构建日志重新构建索引延迟过高数据量过大统计向量数量引入ANN索引4.2 生成质量差上下文与提示词的双重优化检索到相关内容但生成质量差通常是两个原因上下文组织不当或者提示词设计有问题。上下文组织方面我建议按相关度排序最相关的放最前面。因为LLM对上下文开头的注意力更强。另外如果上下文太长要做截断或摘要别把无关内容也塞进去。提示词方面我总结了一个模板结构[角色定义] 你是一个基于给定上下文回答问题的助手。 [约束条件] 只使用上下文中的信息不要编造。 [上下文] {context} [问题] {question} [输出格式] 直接给出答案不要重复问题。这个结构的关键是约束条件。不加约束LLM很容易自由发挥产生幻觉。踩过的坑不要用“请根据以下内容回答”这种模糊指令。要用“只使用以下内容回答如果内容不包含答案请说不知道”。明确的约束能大幅降低幻觉率。4.3 性能与成本批处理、缓存与模型选择性能问题通常出现在两个环节向量化阶段和生成阶段。向量化阶段核心优化手段是批处理和缓存。批处理把多个文本合并成一次API调用减少网络开销。缓存避免重复计算特别是开发调试阶段。生成阶段核心优化手段是控制上下文长度和选择合适的模型。上下文越长生成越慢、越贵。我一般把上下文控制在2000 Token以内。模型选择上简单问题用便宜的小模型复杂问题才用大模型。下面是我实测的一些数据供参考优化手段效果代价批处理32条/批吞吐提升5-8倍单次延迟略增本地缓存重复请求成本降为0需要管理缓存失效上下文截断生成延迟降低40%可能丢失信息小模型替代成本降低80%复杂问题质量下降4.4 我踩过的三个典型坑第一个坑向量维度不一致。有一次我换了向量模型但忘了清空缓存结果新旧向量混在一起检索完全乱套。教训是换模型必须清缓存向量空间变了旧向量就是噪声。第二个坑分块边界截断元数据。前面提到的那个案例分块时把元数据也切了导致检索到内容但无法追溯来源。教训是元数据要和文本绑定存储不能分开。第三个坑提示词注入。用户输入的问题里如果包含“忽略以上指令”之类的内容可能覆盖系统提示词。教训是对用户输入做清洗或者在提示词里加防护指令。最后分享一个小技巧在检索和生成之间加一个“重排序”步骤。用一个小模型对Top-20结果重新打分取Top-5送给LLM。这个步骤成本很低但能显著提升最终答案质量。我实测下来重排序能让答案准确率提升15%左右。这个从零搭建的RAG系统代码量不到500行但涵盖了AI工程的核心环节。你把它跑通之后再用任何框架都会觉得心里有底。因为你知道每个环节在干什么也知道哪里可能出问题。这才是“ai-engineering-from-scratch”的真正价值——不是拒绝框架而是理解框架最终超越框架。