1. 从零搭建AI工程能力为什么“手搓一遍”比调包更值钱这两年AI应用层的工具链成熟得吓人。LangChain、LlamaIndex、各种Agent框架、向量数据库托管服务几乎每个月都有新东西冒出来。你打开任何一个技术社区满屏都是“三行代码搭建RAG”“十分钟上线一个智能客服”。工具确实好用但我在带团队和面试候选人的过程中发现一个很普遍的现象很多人能把框架跑通却说不清楚一次检索背后到底发生了什么向量是怎么存的相似度是怎么算的Token是怎么被切分的模型返回的延迟到底卡在哪一环。ai-engineering-from-scratch这个方向说白了就是反其道而行——不急着上框架先把AI工程里那些最核心的环节用最朴素的方式自己实现一遍。它不是一个具体的开源项目名而是一种学习路径和工程实践思路从零开始手写文本切分、手写向量化调用、手写相似度检索、手写Prompt组装、手写对话状态管理最后再把它们串成一个能跑通的最小可用系统。做完这一轮你再回头看那些框架会发现它们帮你封装的东西一目了然出了问题你也知道该去哪个环节排查。这篇文章适合几类人一是刚转行做AI应用开发框架会用但底层模糊的二是有后端或算法基础想系统补齐AI工程链路的三是团队里需要做技术选型和架构评审得知道每个环节的取舍和坑在哪的。我会按照真实的搭建顺序把每个模块的设计思路、关键参数、实操代码和踩坑经验都摊开讲。全文基于我自己的实践和常见工程做法来写代码以Python为主但思路跟语言无关。2. 整体架构设计与技术选型思路2.1 为什么选择“从零实现”而不是直接上框架先说清楚一个前提我不是反对用框架。生产环境里该用LangChain就用LangChain该上托管向量库就上托管向量库重复造轮子没有意义。但“从零实现”和“用框架”解决的是两个不同的问题。框架解决的是交付效率从零实现解决的是认知深度。我打个比方。你天天开车自动挡踩油门就走这没问题。但如果你连发动机怎么点火、变速箱怎么换挡都不知道车在半路抛锚你只能叫拖车。AI工程现在的情况就是很多人只会踩油门一旦检索结果不准、延迟飙高、成本失控就完全不知道从哪下手。从零实现一遍相当于你把发动机拆开看了一遍再装回去以后听到异响你就知道大概是哪个零件的问题。具体来说从零实现能帮你建立三个层面的认知。第一是数据流认知一段原始文本从进入系统到变成模型能用的上下文中间经过了哪些变换每一步的数据形态是什么。第二是成本认知一次问答到底消耗了多少Token embedding花了多少钱检索花了多少时间这些在框架里往往被隐藏了。第三是故障认知当结果不对时你能快速定位是切分问题、检索问题还是Prompt问题而不是对着框架的黑盒干瞪眼。2.2 最小可用系统的模块拆解一个能跑通的AI问答系统拆开来看其实就五个核心模块我按数据流动的顺序列一下。文档加载与切分模块把各种格式的原始文档读进来切成大小合适的文本块chunk。向量化模块把每个文本块通过Embedding模型转成向量存起来。检索模块把用户问题也转成向量在向量集合里找最相似的几个块。Prompt组装模块把检索到的上下文和用户问题拼成一个完整的Prompt。生成与状态管理模块调用大模型生成回答并管理多轮对话的上下文。这五个模块每一个都有它的设计取舍。比如切分是按固定字数切还是按语义切向量化用哪个模型维度多少检索用余弦相似度还是点积要不要加关键词混合检索Prompt组装上下文放前面还是后面放多少条生成怎么控制幻觉怎么管理历史对话这些问题在框架里都有默认答案但默认答案不一定适合你的场景。从零实现的过程就是逼着你把每个问题都想清楚。2.3 技术栈的取舍与依赖控制从零实现不等于什么都要自己写。Embedding模型和大模型本身你不可能自己训练该调API就调API该用本地模型就用本地模型。我的原则是核心逻辑自己写重型计算用现成的。具体来说文本切分、相似度计算、检索排序、Prompt模板、对话状态管理这些逻辑层面的东西全部自己实现代码量不大但认知收益极高。Embedding和生成用现成的模型服务。向量存储初期直接用NumPy数组或者Python列表就够了数据量上万条之后再考虑上专业的向量库。依赖控制上我建议初期只装最少的包一个HTTP请求库比如requests或httpx一个数值计算库numpy其他能不用就不用。这样你能清楚地看到每一步在干什么而不是被一堆抽象层挡住视线。等最小系统跑通了再逐步引入优化组件每引入一个你都知道它替代了你手写的哪部分为什么值得替代。提示从零实现的目标是建立认知不是造一个生产级系统。所以初期不要追求性能不要过早优化先把链路跑通把每个环节的数据形态和参数搞清楚。3. 核心模块的细节解析与实操要点3.1 文本切分最容易被低估的环节文本切分看起来简单实际上是我见过出问题最多的环节。很多人直接按固定字符数切比如每500字一刀结果把一句话从中间切断把表格切散把代码块切碎。检索出来的上下文语义不完整模型自然答不好。切分的核心目标是每个块在语义上尽量自包含同时大小控制在Embedding模型和生成模型的舒适区间内。块太小语义不完整块太大检索精度下降而且塞进Prompt会挤占空间。我的实操做法是分层切分。第一层按文档的自然结构切比如Markdown按标题层级切PDF按段落切代码按函数切。第二层在自然结构内部如果还是太长再按句子边界切优先在句号、问号、换行处断开。第三层加一个重叠窗口让相邻块之间有10%到20%的重叠避免边界处的信息丢失。def split_text(text, chunk_size500, overlap80): 按句子边界切分带重叠窗口 # 先按句子切 import re sentences re.split(r(?[。.!?])\s*, text) chunks [] current for sent in sentences: if len(current) len(sent) chunk_size: current sent else: if current: chunks.append(current) # 重叠保留上一块尾部 current current[-overlap:] sent if overlap else sent if current: chunks.append(current) return chunks这段代码的关键在于overlap参数。我实测下来中文文本重叠80到120字比较合适英文按词算大概20到30个词。重叠太少起不到衔接作用太多则冗余严重、浪费存储和检索时间。还有一个细节是元数据的保留。每个块除了文本内容还要记录它来自哪个文档、在文档中的位置、所属章节标题。这些元数据在检索后组装Prompt时非常有用可以让模型知道这段内容的出处回答时更准确。注意切分参数没有万能值必须根据你的文档类型和Embedding模型来调。短文本问答场景块可以小一些200到300字长文档摘要场景块可以大一些800到1000字。调参时拿一批真实问题测检索命中率别凭感觉。3.2 向量化模型选择与批量处理向量化就是把文本变成一串数字让语义相近的文本在向量空间里距离也相近。这一步的核心决策是选哪个Embedding模型。选型主要看三个维度语言支持、维度、成本。中文场景我一般推荐用支持中文的模型维度在768到1536之间比较常见。维度不是越高越好高维度检索更精细但存储和计算成本也更高。对于大多数问答场景1024维左右是性价比不错的区间。实操中有一个容易被忽略的点批量处理。如果你一条一条调Embedding接口几千个块要调几千次又慢又费钱。大多数Embedding服务都支持一次传多条文本返回多个向量。批量大小一般控制在16到64条之间太大可能触发接口限制太小则效率低。import numpy as np def embed_texts(texts, batch_size32): 批量向量化返回numpy数组 all_vectors [] for i in range(0, len(texts), batch_size): batch texts[i:ibatch_size] # 这里替换成你实际使用的Embedding调用 vectors call_embedding_api(batch) all_vectors.extend(vectors) return np.array(all_vectors)存下来之后向量的归一化很重要。如果你用余弦相似度最好在存储前就把向量归一化成单位向量这样检索时直接算点积就等于余弦相似度省一次计算。这个优化在数据量大时效果明显。3.3 相似度检索从暴力搜索到近似检索检索的本质是给定一个查询向量在向量集合里找最相似的Top-K个。最朴素的做法是暴力搜索把查询向量和每一个存储向量都算一遍相似度然后排序取前K。数据量在几万条以内暴力搜索完全够用NumPy的矩阵运算一秒钟能算几十万次。def search(query_vector, stored_vectors, top_k5): 暴力检索返回Top-K索引和分数 # stored_vectors已归一化query_vector也归一化 scores stored_vectors query_vector # 点积即余弦相似度 top_indices np.argsort(scores)[::-1][:top_k] return top_indices, scores[top_indices]数据量上去之后比如几十万上百万条暴力搜索就慢了这时候需要近似最近邻ANN算法。常见的有基于图的HNSW、基于量化的IVF等。但我要提醒一句不要过早引入ANN。ANN是近似算法会牺牲一点召回率换速度。如果你的数据量还没到暴力搜索扛不住的程度用ANN反而可能因为召回率下降导致检索结果变差。我一般建议数据量超过十万条再考虑。检索这里还有一个重要技巧是混合检索。纯向量检索擅长语义匹配但对精确的关键词、专有名词、数字不敏感。比如用户问“2023年的营收是多少”向量检索可能召回一堆讲营收的段落但不一定是2023年的。这时候加一路关键词检索比如BM25把两路结果融合效果会好很多。融合方法可以用简单的加权分数也可以用RRF倒数排名融合。3.4 Prompt组装上下文怎么放有讲究检索到Top-K个块之后要把它们和用户问题拼成一个Prompt。这里有几个实操要点。第一是上下文的顺序。有研究表明模型对Prompt开头和结尾的信息更敏感中间的信息容易被忽略。所以最重要的块放在最前面或最后面次要的放中间。我通常按相似度排序后把最相似的放最前面。第二是上下文的数量。不是越多越好。塞太多块一是挤占生成空间二是引入噪声三是增加成本。一般3到5个块比较合适具体看块的大小和模型上下文窗口。如果块是500字5个块就是2500字加上问题和系统提示控制在模型窗口的一半以内比较安全。第三是格式和指令。要明确告诉模型哪些是参考资料哪些是问题以及当资料里没有答案时该怎么回答。我常用的模板是这样的PROMPT_TEMPLATE 你是一个严谨的问答助手。请根据下面提供的参考资料回答用户问题。 如果参考资料中没有相关信息请直接说根据现有资料无法回答不要编造。 参考资料 {context} 用户问题{question} 请给出准确、简洁的回答这个模板里“不要编造”和“无法回答时怎么说”这两句很关键能显著降低幻觉。另外参考资料里最好带上来源标记比如[文档1]这样模型回答时可以引用用户也能追溯。3.5 对话状态管理多轮对话的上下文怎么管单轮问答跑通之后多轮对话是下一个坎。多轮的核心问题是历史对话怎么存、怎么用、什么时候丢弃。最朴素的做法是把所有历史对话都拼进Prompt。但这样Token会线性增长几轮之后就把窗口占满了而且成本飙升。我的做法是维护一个滑动窗口只保留最近N轮对话N一般取3到5。更精细一点的做法是当历史超过窗口时用模型对早期对话做一个摘要把摘要作为长期记忆保留。还有一个细节是指代消解。多轮对话里用户经常说“它”“这个”“那刚才那个”如果直接把当前问题拿去检索向量里没有指代对象的信息检索会失败。解决办法是在检索前先用模型把当前问题改写成包含完整信息的独立问题。比如用户上一句问“LangChain是什么”这一句问“它和LlamaIndex有什么区别”改写后变成“LangChain和LlamaIndex有什么区别”再去检索就准了。def rewrite_query(history, current_question): 把带指代的问题改写成独立问题 if not history: return current_question prompt f根据对话历史把用户的最新问题改写成不依赖上下文也能理解的独立问题。 只输出改写后的问题不要解释。 对话历史 {history} 最新问题{current_question} 独立问题 return call_llm(prompt)这个改写步骤会增加一次模型调用但对手感提升很大尤其是多轮场景。我实测下来加了改写之后多轮检索的命中率能提升不少。4. 完整实操流程从零跑通一个最小问答系统4.1 环境准备与依赖安装先把环境搭起来。我建议用虚拟环境避免污染全局。Python版本3.9以上都行我用3.10测试的。python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate pip install numpy requests就这两个包。Embedding和生成我都用HTTP接口调所以只需要requests。如果你用本地模型再装对应的推理库。向量存储初期用NumPy数组不装向量库。4.2 文档加载与切分的完整实现假设我们有一批Markdown文档放在docs/目录下。先写加载和切分。import os import re def load_documents(doc_dir): 加载目录下所有md文件 docs [] for filename in os.listdir(doc_dir): if filename.endswith(.md): path os.path.join(doc_dir, filename) with open(path, r, encodingutf-8) as f: content f.read() docs.append({source: filename, content: content}) return docs def split_by_heading(text): 按Markdown标题切分保留标题作为元数据 sections [] current_title 正文 current_content [] for line in text.split(\n): if line.startswith(#): if current_content: sections.append({ title: current_title, content: \n.join(current_content).strip() }) current_title line.lstrip(#).strip() current_content [] else: current_content.append(line) if current_content: sections.append({ title: current_title, content: \n.join(current_content).strip() }) return sections def chunk_section(section, chunk_size500, overlap80): 对单个章节做句子级切分 text section[content] sentences re.split(r(?[。.!?])\s*, text) chunks [] current for sent in sentences: if len(current) len(sent) chunk_size: current sent else: if current: chunks.append(current) current (current[-overlap:] if overlap else ) sent if current: chunks.append(current) return [{title: section[title], text: c} for c in chunks]这套流程下来每个块都带着它所属的章节标题检索后组装Prompt时可以把标题也带上帮助模型理解上下文。4.3 向量化与存储的落地代码向量化我用一个统一的接口封装方便替换不同的模型服务。import numpy as np import requests EMBED_API 你的Embedding服务地址 EMBED_KEY 你的密钥 def call_embedding_api(texts): 调用Embedding接口返回向量列表 resp requests.post( EMBED_API, headers{Authorization: fBearer {EMBED_KEY}}, json{input: texts, model: 你的模型名} ) data resp.json() return [item[embedding] for item in data[data]] def build_index(chunks, batch_size32): 构建向量索引 texts [c[text] for c in chunks] vectors [] for i in range(0, len(texts), batch_size): batch texts[i:ibatch_size] vectors.extend(call_embedding_api(batch)) vectors np.array(vectors, dtypenp.float32) # 归一化 norms np.linalg.norm(vectors, axis1, keepdimsTrue) vectors vectors / norms return vectors归一化这一步别省。归一化之后检索时点积就是余弦相似度省一次除法而且数值更稳定。4.4 检索与生成的串联把前面的模块串起来就是一个完整的问答流程。def answer_question(question, chunks, vectors, top_k4): # 1. 问题向量化 q_vec np.array(call_embedding_api([question])[0], dtypenp.float32) q_vec q_vec / np.linalg.norm(q_vec) # 2. 检索 scores vectors q_vec top_indices np.argsort(scores)[::-1][:top_k] # 3. 组装上下文 context_parts [] for idx in top_indices: chunk chunks[idx] context_parts.append(f[{chunk[title]}] {chunk[text]}) context \n\n.join(context_parts) # 4. 组装Prompt并生成 prompt PROMPT_TEMPLATE.format(contextcontext, questionquestion) answer call_llm(prompt) return answer, top_indices, scores[top_indices]跑通之后你可以拿几个真实问题测一下看看检索出来的块是不是相关生成的回答是不是准确。如果不对就回到对应环节调参。4.5 参数调优的实操记录我拿一批技术文档做了几轮调优记录一下关键参数的变化和效果。参数初始值调整后效果变化chunk_size500400检索命中率略升上下文更聚焦overlap080边界信息丢失问题明显改善top_k35召回提升但噪声也增加需配合重排上下文顺序按相似度最相似放首尾模型对关键信息的利用率提升调参的核心方法是准备一批问题和标准答案每次只调一个参数看检索命中率和最终回答质量的变化。别一次调好几个否则你分不清是哪个参数起的作用。5. 常见问题与排查技巧实录5.1 检索结果不相关的排查思路检索不准是最常见的问题。排查顺序我一般是这样先看切分再看向量化最后看检索本身。切分问题表现为检索出来的块语义不完整或者一个完整的意思被切到两个块里。解决办法是调整chunk_size和overlap或者改用按语义切分。向量化问题表现为语义明显相近的文本向量相似度却不高。这通常是Embedding模型不适合你的语言或领域。解决办法是换模型或者在领域数据上做微调。检索问题表现为Top-K里明明有相关的但排序靠后。这可能是相似度算法的问题也可能是查询本身有指代或歧义。解决办法是加查询改写或者引入混合检索和重排。5.2 生成幻觉的抑制手段模型编造答案根子上是两个原因一是检索没给到正确上下文二是Prompt没约束好。前者靠优化检索解决后者靠Prompt工程。我的Prompt里一定会包含三句话一是“只根据参考资料回答”二是“资料没有就说不知道”三是“不要编造”。这三句能挡掉大部分幻觉。另外降低生成温度temperature也有帮助问答场景我一般设0.1到0.3。还有一个技巧是让模型引用来源。在上下文里给每个块编号要求模型回答时标注用了哪个块。这样一方面约束模型基于资料回答另一方面方便你事后核查。5.3 延迟与成本的优化经验延迟主要花在三个地方Embedding调用、检索计算、生成调用。Embedding和生成是网络调用延迟取决于服务商。检索计算在数据量不大时可以忽略。优化延迟的手段一是缓存把常见问题的Embedding和回答缓存起来二是批量把多个请求合并三是流式输出生成时用流式接口让用户先看到部分结果。成本主要是Token消耗。Embedding按输入Token算生成按输入加输出算。控制成本的关键是控制上下文长度。top_k别设太大块别切太大历史对话别无限累积。我一般会把每次请求的Token数打日志跑一段时间后看哪些请求消耗大针对性优化。5.4 常见问题速查表问题现象可能原因排查方向解决办法检索结果不相关切分不当/模型不匹配检查块语义完整性调切分参数或换模型回答编造上下文缺失/Prompt约束弱看检索命中情况优化检索强化Prompt多轮指代失败查询含指代检查查询改写加查询改写步骤延迟高网络调用多/上下文长打点各环节耗时缓存批量流式成本高上下文过长统计Token消耗减top_k控历史提示排查问题时一定要打日志把每次请求的查询、检索到的块、相似度分数、最终Prompt、Token消耗都记下来。没有日志排查就是盲猜。6. 从最小系统到生产系统的演进路径最小系统跑通之后你会对整条链路有清晰的认知。接下来如果要往生产系统演进有几个方向可以逐步推进。第一是向量存储的升级。数据量上来后把NumPy数组换成专业的向量库支持持久化、增量更新和ANN检索。换的时候你会很清楚它替代了你手写的哪部分不会迷失。第二是检索质量的提升。引入混合检索、重排模型、查询改写、多路召回融合。每一步都是在你的最小系统上加一个可插拔的模块而不是推倒重来。第三是工程化的完善。加缓存、加限流、加监控、加评测。评测尤其重要要有一套自动化的评测集每次改动都跑一遍看检索命中率和回答质量有没有退步。第四是Prompt的版本管理。Prompt是系统的核心资产要像管理代码一样管理它每次改动有记录、有对比、可回滚。我自己走完这一轮之后最大的体会是框架不再是黑盒了。看到LangChain的某个组件我能立刻反应过来它对应我手写的哪部分它做了什么优化它的默认参数意味着什么。这种掌控感是单纯调包永远给不了的。而且当线上出问题时我能快速定位到是切分、检索还是生成环节而不是对着框架的报错一脸茫然。如果你也在做AI应用我强烈建议你抽时间把这条链路自己走一遍哪怕只是跑通一个最小版本收获也会远超预期。