做AI工程这么多年我一直有个判断标准能真正落地的东西关键不在于用了多贵的模型、多新的框架而在于你能不能把一个想法从零开始一步步搭成能跑、能修、能量化的系统。市面上讲“AI应用开发”的教程很多但大多是把别人封装好的组件拼一拼出了问题根本不知道从哪下手。“ai-engineering-from-scratch”这个主题说白了就是反着来——不依赖黑盒模板把数据、模型、推理服务这些核心环节一个一个亲手搭出来。这篇文章我会用自己实际做过的项目走一遍完整流程把每个环节“为什么这么做”“遇到坑怎么填”都讲透。适合已经开始接触大模型、但不满足于只会调接口的人也适合团队里负责把原型推上生产环境的同学参考。1. 项目整体设计与思路拆解先聊聊为什么“从零开始”这件事本身值钱。很多人问我现在调用大模型API这么方便Few-shot一写、LangChain一接一个聊天机器人半天就出来了干嘛还要自己折腾工程链路我的回答是Demo和产品之间隔着一整条工程化的河。1.1 AI工程的本质不是搭积木而是建管道AI工程和传统软件工程最大的区别在于它的核心产物不是“写出来的代码”而是一条完整的数据管道加模型服务链路。这条链路里数据的质量、embedding的切分粒度、检索回召的准确性、大模型推理的稳定性任何一个环节飘了整个产品体验就崩了。从零开始搭建的意义就是把这条管道里的每一个阀门都亲手拧一遍知道水从哪来、往哪去、在哪儿最容易漏水。我见过一个团队直接用开源的RAG模板搭了个客服问答系统上线第一周还行第二周用户反馈开始答非所问。查了半天最后发现是文档更新的增量同步逻辑有问题新知识根本没有进向量库。这就是典型的只用了“轮子”却不了解“轮子”内部结构的后果。如果当初是自己负责写索引更新这块绝不会犯这种低级错误。1.2 三条必须前置的工程原则在我真正动手之前给自己定了三个硬约束这三个约束贯穿了整个项目周期。第一条叫“最小可复现”意思是任何模块都必须能在本地用最小代价跑起来哪怕是调云端大模型API我也要在本地留一套mock接口保证开发环境不依赖外部服务。这样做的好处很直接——CI流程稳定测试不飘新同学上手也不用申请一堆权限。第二条叫“显式胜于隐式”大模型推理本质是概率性的但工程却不允许“大概、可能”。所有的Prompt模板、模型参数、上下文策略全部显式地写进配置文件不允许在代码里到处硬编码。这样每次效果波动的时候才能快速定位是哪一个变量变了。第三条叫“先测后调”这是我和很多算法出身的朋友最大的分歧。他们习惯先做个模型看看效果再回头补测试但工程化的做法正相反——先把评估集和指标定好再开始调Prompt和参数。没有评估指标的优化都是自我感觉良好。1.3 技术选型的取舍思路选型这件事我觉得最重要的是分清“能跑的方案”和“好维护的方案”。在这个项目里我刻意选择了一些相对底层、透明的工具而不是大而全的框架。比如向量检索环节我没有直接用专门的向量数据库而是先用了FAISS加内存索引配一个简单的SQLite元数据管理。很多人觉得这样不够“工业化”但对我来说小规模场景下它足够快、足够直观索引结构一清二楚。等规模真正上来需要分布式了那时候再迁移到专门的服务心智负担也小。大模型推理方面我连了一家常规的云端API但封装层是自己写的包括超时重试、token用量统计、流式响应的解析。这个封装层工作量不大却让整个系统的可观测性上了一个台阶。后面排查问题和计算成本时全靠这套自己埋点的日志。2. 核心系统拆解数据、知识库和检索策略一个典型的AI工程应用无论表面是聊天还是问答内部都逃不过三件事数据进得来、知识管得住、答案出得准。下面把我在这三个层面的具体做法和踩过的坑展开讲讲。2.1 数据准备解析、切分与清洗的细节我先拿了一批混合格式的文档有PDF、Word、Markdown还有几个网页。第一直觉是直接用现成的解析库统统转成纯文本但实际跑下来发现一个很严重的问题PDF里的表格被解析成了乱七八糟的换行符Markdown里的代码块被当成了普通段落。这些问题在“看着能用”的阶段完全暴露不出来一旦进入问答阶段检索到的片段语义残缺模型给出的答案自然漏洞百出。所以我在数据准备阶段就做了三层过滤第一层格式清洗统一转成带结构化标记的文本表格保留CSV形式代码块单独标记第二层质量过滤把过短的碎片、重复度高的段落、以及明显是导航内容的噪声剔除第三层人工抽检随机抽5%的切分结果看一眼确认段落边界是否合理。切分这一步是最值得反复调的。固定按字符数切分肯定不行。我现在习惯用“递归结构切分”优先按Markdown的二级标题切标题太长的再按段落切段落太长的才按固定窗口切。切分之后还要做重叠处理我一般让相邻切片之间有100到200个字符的重叠避免关键上下文恰好被切在边界上。2.2 Embedding模型的选择和调优思路Embedding模型的选择直接影响检索的上限。我在项目里对比过几种类型的模型结论是通用型开源Embedding模型做中文场景效果其实已经足够好但要注意它的上下文长度限制。很多模型的默认最大长度是512个token如果你的切片本身就偏长embedding时超出部分会被直接截断那这部分语义就丢了。我自己在实践里会把切片的token数控制在300到500之间并且尽量让每个切片表达一个完整的意思。切片表达“一个意思”这条听起来简单做起来很难依赖的还是前面说的结构切分。同一份资料按天真的固定长度切出来的检索命中率和按结构切出来的命中率在同一个测试集上能差10个点以上。另外还要做一件很多人忽略的事情——归一化。不管是余弦距离还是内积计算统一把向量做过归一化处理之后检索分数才具有跨查询可比性。否则不同查询之间的分数波动会很大你很难定一个可靠的相似度阈值来过滤不合格的召回结果。2.3 检索策略从纯向量到混合检索的挣扎项目做到中段我遇到一个典型问题用户问的是“部署的时候需要修改哪个配置文件”这个问题的关键词是“部署”和“配置文件”但在文档里对应的内容分布的章节标题可能叫“服务启动指南”。这就暴露出纯向量检索的一个通病——它擅长语义匹配但不擅长关键词精确匹配尤其是专有名词、配置项、版本号之类的信息向量检索很容易给漏掉。解决办法就是上混合检索。我在项目里加了BM25关键词检索的通道和向量检索并行跑最后用RRFReciprocal Rank Fusion算法把两路结果融合排序。这条路我已经走了很多遍实话说效果非常明显。RRF的原理不复杂就是分别取两路结果的排名然后按1/(krank)的方式累积分数k一般取60最后按总分数重新排序。用上混合检索之后还必须在召回层就做过滤比如元数据过滤文档类型过滤时间范围过滤。我踩过的一个坑一次对话里检索到了未来时间的数据导致回答里出现了用户还看不到的功能说明。后来我养成了一个习惯把所有切片的元数据来源、更新时间、权限标签都准备好检索阶段直接用结构化条件过滤宁可少召回不可错召回。3. 实操一条龙实现一个最小可用的RAG问答服务理论聊再多不如把代码贴上。这一章我直接用自己项目的简化版带大家走一遍从初始化到部署的最小闭环。为了演示方便我用FastAPI做服务框架FAISS做向量存储后端接大模型API。3.1 环境初始化与数据索引构建我习惯用一个Makefile把整个流程串起来避免每个人在本地配置环境时各自为政。第一步是安装依赖。pip install fastapi uvicorn sentence-transformers faiss-cpu bm25s pydantic httpx接下来是初始化数据索引。核心思路是把文档读进来、切分、向量化、写入FAISS。下面是一段简化但完整的索引构建代码from sentence_transformers import SentenceTransformer import faiss import numpy as np import json model SentenceTransformer(BAAI/bge-m3) # 根据实践中文场景效果好 def split_text(text, max_len350, overlap50): # 简单按字符切分生产环境建议结合文档结构 chunks [] start 0 while start len(text): end start max_len chunks.append(text[start:end]) start end - overlap return chunks all_chunks [] for doc_id, doc_text in load_documents().items(): chunks split_text(doc_text) for i, chunk in enumerate(chunks): all_chunks.append({doc_id: doc_id, chunk_id: i, text: chunk}) vectors model.encode([c[text] for c in all_chunks], normalize_embeddingsTrue) dimension vectors.shape[1] index faiss.IndexFlatIP(dimension) index.add(vectors.astype(float32)) faiss.write_index(index, data/faiss.index) with open(data/chunks.json, w, encodingutf-8) as f: json.dump(all_chunks, f, ensure_asciiFalse)这段代码看起来简单但要提醒几个关键点。一个是我用IndexFlatIP即内积距离配合归一化后的向量等于用余弦相似度排序。另一个是normalize_embeddingsTrue这个参数前面专门提过一定要开否则检索分数没有可比性。还有一个非常容易被忽视的坑是把文本直接传给模型时没有检查token长度上限实际使用时应该在前面加一个截断逻辑。提示建索引的时候养成良好的习惯给每个切片分配doc_id和chunk_id。后面排查“这个答案是从哪段文本里来的”时这两个ID就是救命稻草。3.2 检索与答案生成的完整链路索引建好之后就到了核心的服务端逻辑。我单独写了一个Retriever类先把向量检索和关键词检索都封装好然后做融合。class Retriever: def __init__(self, faiss_index_path, chunks_path): self.index faiss.read_index(faiss_index_path) with open(chunks_path, r, encodingutf-8) as f: self.chunks json.load(f) self.model SentenceTransformer(BAAI/bge-m3) # BM25索引在服务启动时构建 self.bm25 None self._build_bm25_index() def search(self, query: str, top_k: int 10): # 向量检索 q_vec self.model.encode([query], normalize_embeddingsTrue) scores, indices self.index.search(q_vec.astype(float32), top_k) # BM25检索 bm25_scores, bm25_indices self.bm25.search(query, top_k) # 用RRF融合两种结果 fused_scores {} for rank, idx in enumerate(indices[0]): fused_scores[idx] fused_scores.get(idx, 0) 1 / (60 rank) for rank, idx in enumerate(bm25_indices[0]): fused_scores[idx] fused_scores.get(idx, 0) 1 / (60 rank) ranked sorted(fused_scores.items(), keylambda x: x[1], reverseTrue) return [(self.chunks[idx], score) for idx, score in ranked[:top_k]]下面是检索结果组装大模型回答的部分。这里我想强调一下Prompt模板设计很多人不重视这个觉得反正模型很强把内容拼进去就行。实际差远了。一段检索文本拼接到Prompt里如果没有明确的逻辑引导模型容易自己脑补内容甚至直接忽略检索到的证据。def build_prompt(query, retrieved): context \n\n.join( f[片段{doc_id}-{chunk_id}] {text} for doc_id, chunk_id, text in retrieved ) prompt f请根据以下检索到的资料片段回答用户的问题。 要求 1. 只能依据给定片段不要使用片段之外的知识如果片段内容不足直接回答“根据现有资料无法回答”。 2. 回答时先简要列出参考依据。 3. 不要复述“片段说”“文档中提到”这类表述直接给出专业回答。 片段如下 {context} 用户问题{query} return promptPrompt里加了“参考依据”和“无法回答”的限制之后幻觉率明显下降。这其实是用工程手段来约束模型的概率输出让它不那么自由发挥。在实际项目中我还会把温度参数调低到0.2到0.3。3.3 API服务封装与流式输出最后一步是把上面的逻辑包在API里。我用的是FastAPI它原生支持流式响应配合大模型API的流式输出能显著提升用户的等待体验。下面是我常用的一个接口模式from fastapi import FastAPI from fastapi.responses import StreamingResponse import httpx app FastAPI() retriever Retriever(data/faiss.index, data/chunks.json) LLM_API_URL https://your-llm-api.example/v1/chat/completions async def stream_qa(query: str): # 1. 检索 hits retriever.search(query, top_k5) prompt build_prompt(query, [(h[doc_id], h[chunk_id], h[text]) for h, _ in hits]) # 2. 回调大模型并流式转发 payload { model: your-model-name, messages: [{role: system, content: 你是专业助手}], temperature: 0.2, stream: True, } async with httpx.AsyncClient(timeout60) as client: async with client.stream(POST, LLM_API_URL, jsonpayload) as resp: async for line in resp.aiter_lines(): if line.strip().startswith(data:): yield line.strip()[5:] \n app.post(/v1/qa) async def qa(request: dict): query request[query] return StreamingResponse(stream_qa(query), media_typetext/event-stream)这个封装有个细节值得展开讲讲我把检索结果的doc_id和chunk_id也放到了返回流里客户端可以在界面上展示“答案参考了哪份文档的哪一段”。不要小看这个功能它直接影响用户对AI回答的信任度。我做过简单的用户反馈统计展示依据后用户把回答标记为“有用”的比例提升了将近两成。还有一件事必须做——记录每次问答的全链路日志。我这里的做法是生成一个request_id把检索到了哪几段、每段的相似度分数、大模型返回的token数、响应耗时全部写进结构化日志。这组数据是你后面优化效果、排查线上问题的唯一线索。3.4 部署与运维要点别把服务跑挂了才知道监控重要代码写完只是开始部署和运维才是工程化的试金石。我见过太多项目开发环境一切正常一上生产就各种状况百出。这里有两个我在实践中付出过代价的点。一是冷启动问题。bge-m3这类Embedding模型在CPU上加载一次要几十秒如果服务采用多副本部署每次发布新版本所有实例都要重新加载模型可能触发服务不可用。我的解决办法是预加载加预热的组合在Dockerfile里启动命令前加一个python -c import retriever; retriever.preload()的步骤同时这段时间挂个healthz接口等模型加载完成再宣布就绪。二是超时和重试策略。大模型API的响应普遍偏慢但也不能无限等。我在请求层设置了连接超时10秒、读取超时60秒并采用指数退避重试机制。重试的时机必须想清楚如果用户已经看到了前几个字突然断掉再重连那就乱了。所以我只在模型返回首个token之前重试一旦开始留数据了就不再重试。4. 常见问题与排查实录做AI工程这一年多踩过的坑比写过的代码还多。下面挑几个特别典型的附上排查思路希望对大家有帮助。4.1 检索效果差是切分问题是模型问题还是排序问题“检索效果差”是一句特别模糊的抱怨因为差在哪个环节是完全不同的事。我现在的排查顺序是这样的。第一步看召回结果本身。把检索返回的前10个片段打印出来人肉看一眼判断到底是没有相关片段被召回还是相关片段排名太靠后。如果连相关片段都压根没出现多半是切分策略或嵌入模型的问题如果相关片段有但排名靠后问题更可能在融合排序或重排环节。第二步检查切分边界。很多时候相关的内容被切得七零八落Embedding模型拿到的是一段没有主语的碎片召回自然不理想。解决办法是回到2.1的结构切分或者调整切分长度。第三步才是考虑换模型或者上重排序模型。重排序模型虽然会增加一次推理开销但它对于“检索结果有但排序不对”的情况确实有效建议在知识库规模较大的时候安排上。4.2 模型回答出现幻觉Prompt约束失效怎么办前面提到我在Prompt里加了“不能使用片段之外的知识”的约束但必须实话实说——这只是一个软约束模型该输出幻觉还是会输出。真正能兜底的是工程手段。第一个兜底手段是全链路追踪。每次问答都能回溯到片段依据一旦线上反馈有幻觉能立刻看看到底是基于哪个片段出的错。第二个手段是置信度过滤。我给检索环节设了相似度阈值低于阈值就直接回答“我在资料中没有找到相关信息”坚决不给大模型发挥的机会。第三个手段更硬核是用一个独立的校验层把大模型生成的回答重新Embedding一遍跟检索到片段做相似度比对如果答案的关键实体在证据片段中根本不存在那这个回答就不允许直接返回而是触发一次重答或者转人工。这个方案我现在验证下来确实有效缺点是多一次推理成本适合对准确率要求高的场景。4.3 成本超支怎么追模型、token和缓存的三笔账做AI工程的人一定会被问到一个问题这玩意儿一个月要烧多少钱成本优化这个话题很大但最核心的账其实就三笔推理token费、Embedding费用、以及向量库存储费用。用好缓存可以帮你砍掉第一笔的大头。我的办法很简单也推荐给预算受限的团队给问答加一层语义缓存。用户问过的问题把问题和答案存进数据库等新问题来的时候先做一次向量检索搜历史问题记录如果相似度超过0.92直接复用旧答案不再调用大模型API。我自己的项目里这套缓存大约节省了35%的API消耗而且响应速度直接从秒级降到了几十毫秒。Embedding费用也要注意。很多人在数据预处理阶段反复重复计算相同文档的向量这是完全没有必要的。我的做法是给每个文档算一个哈希值文档没变就直接从库里取向量不重新推理成本能省不少。4.4 本地熟悉问答速查表下面这张表是我平时排查问题时的第一参考整理出来供各位直接抄作业。现象优先排查点次优先排查点常见处理方式召回内容不相关切分方式和重叠度Embedding模型与维度改结构切分做归一化相关片段排名太后融合排序权重RRF的k值是否缺重排模型调k值或加cross-encoder回答内容偏离题目Prompt模板引导不足温度或top_p参数过高收紧“仅依据片段”约束降低温度答案缺乏时效性元数据过滤没生效增量更新逻辑有bug加时间范围过滤检查索引更新任务API响应过慢是否走了流式输出是否有缓存加语义缓存开流式响应成本突然飙升是否有循环调用是否有无效检索加日志计量按request_id聚合5. 给同样在从零开始做AI工程的人几句心里话最后分享几句我做这个项目积累下的真实体会。很多人一上来就喜欢堆框架觉得组件用得越多系统就越先进。但AI工程的复杂度主要来自不确定性而不是组件数量你把模型输出的不确定性管住了比接十个花哨框架都有用。从零开始做AI工程最大的收获不是那个能跑的Demo而是你不怕它出错了。因为链路里的每个环节都是自己搭的问题来了你知道往哪看。把日志和评估集这两件事做好就已经超越了大多数团队。如果这项目要续着往下做我建议你可以从这几个方向扩充一是开源模型的本地化部署把网关、推理加速这些自己搭一遍二是提示词工程的评估自动化搞一个小型的回归测试集三是长上下文记忆管理目前能做好这个的产品还不算多。路很长但每一步都看得见。