
1. 从零搭建AI工程能力为什么“手搓一遍”比调包更值钱这两年AI应用层的工具链成熟得吓人一个RAG问答机器人从向量库到编排框架再到前端UI几乎每个环节都有现成的轮子。很多人上手第一周就能拼出一个能跑通Demo的系统但一旦线上流量上来、数据分布漂移、延迟抖动、成本失控问题就全冒出来了。我见过太多团队卡在这个阶段Demo很惊艳上线就拉胯。根本原因不是工具不好而是对底层链路的理解是断层的——知道怎么调API但不知道token是怎么被切分的、embedding是怎么被检索的、推理延迟到底花在哪一段。ai-engineering-from-scratch这个方向说白了就是把AI工程里那些被框架封装掉的环节自己动手实现一遍。它不是让你去造一个大模型而是让你亲手写一个最小可用的推理服务、一个朴素的向量检索、一个不依赖任何编排框架的Agent循环。目标读者很明确已经会用LangChain或LlamaIndex跑通Demo但想搞清楚“框架到底替我做了什么”的工程师以及准备从传统后端/算法岗转向AI工程岗需要一套能写进简历、能扛住面试追问的实战项目的人。我自己走过这条路。最早做RAG的时候检索效果不好就换向量模型换了还是不好就调top_k调了还是不好就加rerank全程像在盲人摸象。直到我把整个链路用最原始的方式重写了一遍——手动分块、手动算余弦相似度、手动拼prompt——才发现问题出在分块策略上一个参数错了后面全白搭。这种“从零实现”带来的掌控感是调包永远给不了的。接下来的内容我会把这条路上最关键的几个模块拆开讲包括设计取舍、实操细节、踩过的坑以及一套可以直接照着复现的方案。2. 整体架构设计先想清楚“从零”的边界在哪2.1 为什么选择“最小依赖”而不是“完全裸写”“from scratch”这个词很容易被误解成“什么库都不用”。真这么干光是矩阵运算就能劝退99%的人。我的理解是核心逻辑必须自己写基础设施可以用成熟库。比如向量检索你可以用numpy做矩阵乘法但没必要自己实现BLAS你可以用FastAPI起服务但没必要自己写HTTP解析。判断标准很简单——这个环节的决策会直接影响系统效果和成本那就自己写纯粹是工程脚手架那就用现成的。具体到AI工程我建议自己实现的部分包括文本分块逻辑、embedding调用与缓存、相似度计算与检索排序、prompt组装与上下文管理、Agent的工具调用循环、以及最基础的评估指标计算。这些环节每一个都藏着影响最终效果的“魔鬼细节”框架帮你做了默认选择但默认选择往往不适合你的场景。可以用现成库的部分Web框架、向量运算库、日志监控、部署工具。这样既保证了学习深度又不至于把时间浪费在重复造轮子上。2.2 分层架构与数据流向我习惯把整个系统分成四层从下往上依次是数据层、检索层、推理层、编排层。数据层负责原始文档的加载、清洗、分块和向量化产出的是一批带元数据的向量记录。检索层接收查询向量做相似度计算返回top_k结果这里可以加BM25做混合检索。推理层封装大模型调用包括prompt模板、参数控制、重试和降级。编排层就是Agent的大脑决定什么时候检索、什么时候调工具、什么时候直接回答。数据流向是这样的用户输入 - 编排层判断意图 - 如果需要知识检索调用检索层 - 检索层从数据层拿向量做匹配 - 返回相关片段 - 编排层组装prompt - 推理层调用模型 - 返回结果。每一层之间用明确的接口隔离这样你可以单独替换某一层的实现比如把numpy检索换成FAISS上层代码完全不用动。这种设计还有一个好处调试的时候可以逐层验证先确认检索结果对不对再确认prompt组装有没有问题最后看模型输出而不是面对一个黑盒抓瞎。2.3 技术选型背后的取舍逻辑选型这件事我的原则是优先选你能完全掌控的。向量运算用numpy而不是直接上FAISS因为数据量在万级以下时numpy的矩阵乘法完全够用而且你能清楚看到每一步计算。等数据量上到十万级再换FAISS也不迟接口早就抽象好了。Web框架用FastAPI因为它自带异步支持和自动文档省事。模型调用层自己写一个薄封装不要直接用SDK因为你需要加缓存、加重试、加token计数这些逻辑放在自己的封装里最灵活。有一个坑我踩过一开始为了“看起来专业”直接上了Milvus做向量库结果本地开发环境跑不起来光配Docker就花了两天。后来退回来用numpy半小时就跑通了全流程。工具的选择应该服务于你的当前阶段而不是反过来。Demo阶段用最简单的方案快速验证验证通过后再逐步替换成生产级组件这个节奏比一上来就堆全套要高效得多。3. 核心模块拆解每个环节的魔鬼都在细节里3.1 文本分块最容易被低估的环节分块策略直接决定了检索质量的上限。我见过太多人直接用RecursiveCharacterTextSplitter的默认参数chunk_size1000overlap200然后抱怨检索不准。问题在于1000个字符对于技术文档来说可能截断一个完整的代码示例对于法律合同来说可能切断一个条款。分块的本质是在“语义完整性”和“检索粒度”之间找平衡。我的做法是按文档类型走不同策略。技术文档按标题层级切每个二级标题下的内容作为一个块如果超过800字符再按段落细分。对话记录按轮次切一问一答作为一个块。长篇文章用滑动窗口窗口大小500字符步长250保证重叠。这里有个细节重叠区域不是简单复制而是要在元数据里标记这是重叠部分检索命中时去重否则同一个信息会出现两次浪费上下文窗口。还有一个容易被忽略的点分块时要保留上下文信息。比如一个块讲的是“该函数返回一个列表”如果不知道前面在讲哪个函数这个块就是废的。我的解决方案是在每个块的文本前面拼接上它的父级标题路径比如“第三章 3.2 数据加载 函数说明该函数返回一个列表”。这样即使块被单独检索出来模型也能理解它在原文中的位置。实测下来这个简单的改动能让检索准确率提升15%以上。3.2 Embedding与缓存省钱和提速的关键Embedding调用是RAG系统里最频繁的外部调用每次检索都要把query转向量每次入库都要把文档转向量。如果不做缓存成本会线性增长延迟也下不来。我的缓存策略分两层内存缓存用LRU磁盘缓存用SQLite。内存缓存存最近1000条query的向量磁盘缓存存所有文档的向量key是文本的MD5值。这里有个坑不要用文本本身做key要用文本的哈希。因为文本可能很长直接做key会占用大量内存。另外缓存要设置过期策略因为embedding模型可能会升级旧向量和新向量不在同一个空间里混用会导致检索结果完全错乱。我的做法是在缓存key里带上模型版本号模型一换缓存自动失效。还有一个优化点批量embedding。如果你要入库1000个文档块不要循环调用1000次API而是攒成10批每批100个。大多数embedding服务都支持批量输入这样能把调用次数降到十分之一延迟也大幅降低。但要注意批量大小不要超过服务的限制一般128是个安全值。我实测过批量调用比单条调用快8倍左右成本也略有下降。3.3 相似度计算与检索排序相似度计算本身很简单就是余弦相似度numpy一行代码的事。但检索排序的策略才是决定效果的关键。最朴素的做法是只按向量相似度排但这样会漏掉关键词精确匹配的情况。比如用户搜“错误码 500”向量检索可能返回一堆讲“服务器错误”的文档但真正包含“500”这个具体数字的文档反而排后面。我的方案是混合检索向量相似度占70%权重BM25关键词匹配占30%权重加权求和后重新排序。BM25的实现可以用rank_bm25这个库很轻量。加权系数不是拍脑袋定的我是在自己的测试集上网格搜索出来的70/30在我的场景下最优。你的场景可能不同建议用A/B测试确定。还有一个技巧MMR最大边际相关性去重。检索出来的top_k结果里经常有内容高度相似的块如果全塞进prompt既浪费上下文又可能让模型困惑。MMR的做法是在选下一个结果时既考虑它和query的相关性也考虑它和已选结果的差异性。实现起来不复杂就是在排序分数上减去一个和已选结果的最大相似度乘以lambda系数。lambda取0.5到0.7之间比较合适太小会导致结果太分散太大又起不到去重效果。3.4 Prompt组装与上下文管理Prompt组装看起来就是拼字符串但顺序和格式对模型输出的影响巨大。我的经验是指令放最前面上下文放中间问题放最后。因为很多模型对开头和结尾的内容注意力更集中把关键指令放开头能提高遵循度。上下文之间用明确的分隔符隔开比如---并且给每个片段编号方便模型引用。上下文长度管理是另一个重点。模型的上下文窗口是有限的比如8k token你不可能把所有检索结果都塞进去。我的做法是按相关性排序后从高到低累加token数直到接近窗口的80%就停止。留20%的余量给模型输出和系统prompt。这里要精确计算token数不能用字符数估算因为中英文的token比例差异很大。可以用tiktoken库它和OpenAI的tokenizer一致算得很准。还有一个细节上下文中要包含元数据。比如每个片段前面标注来源文件名和页码这样模型在回答时可以引用来源用户也能验证。如果检索结果来自不同文档还要告诉模型这些文档的关系避免它把不相关的信息混在一起。我通常会在prompt里加一句“以下片段按相关性排序请优先使用靠前的片段”。4. 实操全流程从零跑通一个最小可用系统4.1 环境准备与依赖安装先建一个干净的虚拟环境Python 3.10以上。依赖装这些就够了numpy做向量运算fastapi和uvicorn起服务openai调模型或者你用的任何模型SDKtiktoken算tokenrank_bm25做关键词检索sqlite3是Python自带的不用装。总共不到10个包非常轻量。python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate pip install numpy fastapi uvicorn openai tiktoken rank_bm25这里有个建议把模型调用的base_url和api_key放在环境变量里不要硬编码在代码中。用os.getenv读取本地开发可以建一个.env文件用python-dotenv加载。这样代码可以安全地分享不会泄露密钥。4.2 数据入库从原始文档到向量库假设你有一批Markdown格式的技术文档放在docs/目录下。第一步是遍历目录读取每个文件的内容。然后按前面说的分块策略切分每个块生成一个字典包含text、source、chunk_id三个字段。接着批量调用embedding接口把每个块的文本转向量。最后把所有块和向量存到SQLite里表结构很简单id、text、source、vector存成二进制、model_version。import sqlite3 import numpy as np import hashlib def init_db(pathvectors.db): conn sqlite3.connect(path) conn.execute( CREATE TABLE IF NOT EXISTS chunks ( id TEXT PRIMARY KEY, text TEXT, source TEXT, vector BLOB, model_version TEXT ) ) conn.commit() return conn def save_chunk(conn, text, source, vector, model_version): chunk_id hashlib.md5((text source).encode()).hexdigest() conn.execute( INSERT OR REPLACE INTO chunks VALUES (?, ?, ?, ?, ?), (chunk_id, text, source, vector.astype(np.float32).tobytes(), model_version) ) conn.commit()注意向量要转成float32再存因为float64会占双倍空间而检索精度损失可以忽略。读取的时候用np.frombuffer还原。这个方案在万级数据量下全量加载到内存也就几十MB检索时直接做矩阵乘法毫秒级返回。4.3 检索服务查询向量与相似度排序检索的入口是一个函数接收query字符串返回top_k个相关块。流程是先把query转向量走缓存然后从SQLite加载所有向量组成矩阵计算余弦相似度同时用BM25算关键词分数加权融合后排序最后做MMR去重。def retrieve(query, conn, top_k5, alpha0.7): query_vec get_embedding(query) # 带缓存的embedding调用 rows conn.execute(SELECT id, text, source, vector FROM chunks).fetchall() vectors np.array([np.frombuffer(r[3], dtypenp.float32) for r in rows]) # 余弦相似度 query_norm query_vec / np.linalg.norm(query_vec) vec_norms vectors / np.linalg.norm(vectors, axis1, keepdimsTrue) cosine_scores vec_norms query_norm # BM25分数 corpus [r[1] for r in rows] bm25 BM25Okapi([doc.split() for doc in corpus]) bm25_scores np.array(bm25.get_scores(query.split())) bm25_scores bm25_scores / (bm25_scores.max() 1e-8) # 归一化 # 融合 final_scores alpha * cosine_scores (1 - alpha) * bm25_scores top_indices np.argsort(final_scores)[::-1][:top_k * 2] # 多取一些给MMR # MMR去重 selected mmr_select(vectors[top_indices], final_scores[top_indices], top_k) return [(rows[i][1], rows[i][2]) for i in selected]这里有个性能优化点每次检索都重新算所有向量的norm是浪费的。可以在入库时就把norm算好存起来检索时直接用。另外如果数据量超过5万numpy的全量矩阵乘法会开始变慢这时候可以考虑用FAISS的IVF索引但那是后话先跑通再说。4.4 Agent循环让模型自己决定下一步Agent的核心是一个循环模型输出要么是最终答案要么是一个工具调用请求。如果是工具调用就执行工具把结果追加到对话历史再次调用模型直到模型输出最终答案或达到最大轮次。def agent_loop(user_input, max_turns5): messages [{role: system, content: SYSTEM_PROMPT}] messages.append({role: user, content: user_input}) for turn in range(max_turns): response call_llm(messages) if response.tool_calls: for tool_call in response.tool_calls: result execute_tool(tool_call) messages.append({role: assistant, content: None, tool_calls: [tool_call]}) messages.append({role: tool, content: result, tool_call_id: tool_call.id}) else: return response.content return 达到最大轮次未能完成工具定义用JSON Schema描述包括工具名、功能说明、参数列表。模型会根据说明决定是否调用。这里的关键是工具描述要写得足够清晰否则模型会乱调。比如检索工具的描述要写明“当需要查询技术文档中的具体信息时使用”而不是简单的“检索文档”。5. 常见问题与排查技巧实录5.1 检索结果不相关怎么办这是最高频的问题。排查顺序是先看分块是否合理把检索到的块打印出来人工判断它是否真的和query相关。如果块本身就不相关那是分块或embedding的问题。如果块相关但排序靠后那是排序策略的问题。如果块相关且排序靠前但模型没用那是prompt的问题。我遇到过一个典型案例用户问“如何配置超时时间”检索返回的块全是讲“超时错误处理”的。原因是分块时把“配置”和“超时”切到了不同的块里。解决方案是调整分块策略让标题和内容在同一个块里。分块问题占了检索问题的六成以上优先排查这个。5.2 模型回答出现幻觉怎么破幻觉的根源通常是上下文里没有答案但模型硬编了一个。解决方案有三个层次第一在prompt里明确要求“如果上下文中没有相关信息请回答不知道”第二在检索后加一个相关性阈值低于阈值的块直接丢弃宁可少给上下文也不给不相关的第三让模型在回答时引用来源比如“根据文档A的第3段”这样用户能验证模型也会更谨慎。我实测下来加阈值过滤是最有效的。具体做法是计算检索分数如果top1的分数低于某个值比如0.3就直接返回“未找到相关信息”不调用模型。这个阈值需要在你的测试集上校准不同embedding模型的分数分布不一样。5.3 延迟太高怎么优化延迟主要花在三个地方embedding调用、向量检索、模型推理。embedding调用用缓存和批量可以大幅降低。向量检索在万级数据下是毫秒级不是瓶颈。模型推理是大头优化手段包括用更小的模型、减少上下文长度、开启流式输出让用户感知更快。还有一个容易被忽略的点并发。FastAPI默认是同步的如果模型调用是阻塞的并发请求会排队。把模型调用改成异步或者用线程池能显著提升吞吐。我试过用asyncio.to_thread包装同步的SDK调用QPS提升了3倍。5.4 常见问题速查表问题现象可能原因排查方法解决方案检索结果不相关分块不合理打印检索块人工检查调整分块策略保留标题上下文模型回答幻觉上下文无答案检查检索结果是否包含答案加相关性阈值prompt要求引用来源延迟高模型推理慢分段计时换小模型减上下文异步调用成本高embedding调用频繁统计调用次数加缓存批量调用重复回答上下文有重复块检查检索结果MMR去重重叠区域标记工具调用错误工具描述不清看模型输出的调用参数完善工具描述加参数示例6. 从Demo到生产还需要补哪些课跑通最小系统只是第一步。要上生产还需要补几块评估体系、监控告警、降级策略。评估体系包括检索准确率、回答忠实度、用户满意度可以用LLM做自动评估但要有黄金测试集做校准。监控要记录每次请求的检索分数、token消耗、延迟异常时告警。降级策略是当模型服务不可用时返回检索结果原文至少保证用户能拿到信息。我个人在实际操作中的体会是从零实现的价值不在于代码本身而在于建立对每个环节的直觉。当你亲手调过分块参数、算过相似度、拼过prompt再回头看那些框架你就能一眼看出它的默认选择适不适合你的场景。这种判断力是调包调不出来的。后续可以扩展的方向包括多路召回融合、查询改写、自适应检索让模型决定要不要检索每一个都值得单独写一篇。