1. 从零搭建AI工程体系为什么我劝你别一上来就啃框架这两年AI应用开发的门槛肉眼可见地降低了随便拉个前端都能用现成的API搭出一个能跑通的对话Demo。但真到了要把这套东西塞进生产环境、扛住真实流量、控制住成本、还得让模型输出稳定可控的时候你会发现光会调接口根本不够用。ai-engineering-from-scratch这个项目标题说的就是从零开始把AI工程这套东西一层一层搭起来而不是站在LangChain或者各种封装框架的肩膀上做个调包侠。我自己带过几个从零到一的AI项目踩过的坑基本都集中在同一个地方前期图快直接上高层框架结果出了问题根本不知道是哪一层崩的。检索召回不准、上下文拼接超长、模型输出格式飘忽、并发一上来延迟直接爆炸这些问题在Demo阶段全被掩盖了上线之后集中爆发。所以这个项目的核心价值是帮你建立一套完整的AI工程心智模型知道一个AI应用从输入到输出中间到底经过了哪些环节每个环节的边界在哪出问题该往哪个方向排查。这篇文章适合三类人看。第一类是后端或者全栈工程师想转AI应用方向但不想只做API搬运工第二类是做了一段时间AI Demo发现往生产推的时候处处碰壁的开发者第三类是想系统理解AI工程全貌的技术负责人需要知道一个AI系统到底由哪些模块组成、各模块的成熟度如何。我会按照从底层到上层的顺序把数据准备、检索增强、提示工程、模型调用、输出解析、评估监控这几个核心环节拆开讲每个环节都给出可落地的方案和我在实际项目里验证过的参数。需要提前说明的是AI工程这个领域变化极快今天的最佳实践可能半年后就被推翻。所以我更侧重讲清楚每个环节的设计逻辑和取舍依据而不是给你一份死记硬背的配置清单。理解了为什么这么设计你才能在工具迭代的时候快速迁移。2. 整体架构设计与技术选型思路2.1 为什么选择从零手搓而不是直接用框架市面上主流的AI应用框架比如LangChain、LlamaIndex确实能让你在半小时内跑通一个RAG Demo。但我在实际项目里的经验是这类框架的抽象层太厚一旦你需要做精细化的控制比如自定义检索策略、调整上下文压缩逻辑、实现特定的重排序算法就会发现自己一直在跟框架的设计理念做斗争。从零搭建的好处是每一层都透明。你知道用户的问题进来之后先经过什么处理再经过什么处理每一步的输入输出长什么样。出了问题你可以精确地定位到是检索层召回率不够还是提示词模板拼接有问题还是模型本身对某类输入不敏感。这种可观测性在生产环境里是刚需框架帮你省下的那点开发时间在排查线上问题的时候会加倍还回去。当然从零搭建不意味着所有东西都自己写。向量数据库、嵌入模型、推理服务这些基础设施该用现成的就用现成的没必要重复造轮子。我的原则是业务逻辑层自己控制基础设施层用成熟方案。这样既保证了灵活性又不会在底层浪费太多时间。2.2 核心模块拆解与数据流向一个完整的AI工程系统从用户输入到最终输出大致会经过这么几个阶段。用户的问题首先进入输入预处理层做意图识别、查询改写、敏感内容过滤。然后进入检索层如果是知识密集型任务会去向量库或者关键词索引里召回相关文档。召回的结果经过重排序和压缩筛选出最相关的片段。接着进入提示组装层把系统指令、检索到的上下文、用户问题、历史对话按特定模板拼成最终的提示词。再往下是模型调用层负责请求分发、重试、降级、流式输出。模型返回的原始文本进入输出解析层做结构化提取、格式校验、后处理。最后是评估与监控层记录每次请求的完整链路数据用于后续的效果分析和迭代优化。这个数据流向看起来线性但实际实现的时候会有很多分支。比如有些查询不需要检索直接走模型知识回答有些查询需要多轮检索先召回再根据中间结果二次召回有些场景需要并行调用多个模型然后做结果融合。这些分支逻辑就是AI工程真正复杂的地方也是从零搭建时你需要自己设计清楚的部分。2.3 技术栈选型的取舍逻辑在具体技术选型上我倾向于把选择分成三类来看。第一类是必须自己控制的包括提示词模板管理、检索策略、输出解析逻辑、评估指标定义。这些东西直接决定了系统的效果上限必须掌握在自己手里。第二类是可以用开源方案的包括向量数据库、嵌入模型、重排序模型、推理服务框架。这些组件相对标准化社区方案已经比较成熟。第三类是建议用云服务的包括底层算力、模型API、监控告警基础设施。这些自建成本太高除非有特殊合规要求否则没必要自己搞。具体到向量数据库我在中小规模场景下更倾向用轻量级方案比如基于FAISS或者Chroma做本地索引部署简单、调试方便。数据量上到千万级再考虑Milvus或者Qdrant这类分布式方案。嵌入模型的选择要看语言和领域通用场景下多语言模型基本够用垂直领域建议做微调或者至少做一轮领域适配的评估。重排序模型是提升检索精度的关键很多团队会忽略这一步但实测下来加上重排序之后Top-3的命中率能有明显提升。3. 核心模块的细节实现与实操要点3.1 数据准备与索引构建的关键细节AI工程的上限很大程度上由数据质量决定。我见过太多项目模型和框架都选得不错但知识库里的文档乱七八糟切分粒度不合理元数据缺失导致检索效果怎么调都上不去。数据准备这一步核心要解决三个问题文档怎么切、元数据怎么打、索引怎么建。文档切分不是简单地按固定字数切。我的经验是切分粒度要跟内容的语义结构对齐。技术文档按章节切FAQ按问答对切长文章按段落切代码按函数或类切。切分的时候要保留一定的重叠通常设置10%到20%的重叠比例避免关键信息刚好被切断。重叠太多会浪费存储和检索开销太少会丢上下文这个比例需要根据实际内容做几轮测试来定。元数据的设计往往被低估。除了文档来源、创建时间这些基础字段我建议至少加上内容类型、所属主题、置信度这几个维度。内容类型用于区分是事实性知识还是观点性内容检索的时候可以给不同类型设置不同权重。所属主题用于做检索时的范围过滤比如用户问的是产品功能就只在产品文档范围内检索。置信度用于标记内容的可靠性低置信度的内容在拼接上下文时可以做降权处理。索引构建阶段我通常会同时建向量索引和关键词索引两套。向量索引负责语义召回关键词索引负责精确匹配。用户查询里如果包含特定的产品名、错误码、专有名词关键词索引的召回效果往往比向量索引更准。两套索引的结果做融合用RRF或者加权求和的方式合并排序实测下来比单走向量索引的召回率高出一截。注意索引构建完成后一定要做一轮召回测试准备一批典型查询人工标注哪些文档是真正相关的然后计算召回率和精确率。没有这步验证后面所有调优都是盲调。3.2 检索增强生成的参数调优实战RAG这套东西看起来简单但参数特别多每个参数都会影响最终效果。我按重要性排个序讲一下实际调优时的思路。Top-K的取值是最先要定的。K太小召回不全模型没有足够信息回答K太大上下文里塞了一堆无关内容反而干扰模型判断还浪费token。我的经验是先用一个中等值比如10做基线然后观察不同K值下最终答案的质量。通常K在5到15之间会有一个比较明显的拐点超过这个点之后效果提升就很小了。如果用了重排序可以先把召回K设大一些比如50重排序之后再取Top-5或者Top-8送给模型。相似度阈值是第二个关键参数。低于某个相似度的文档宁可不要也不要硬塞进去。这个阈值跟嵌入模型的特性有关不同模型的分值分布不一样不能直接套用。我的做法是拿一批标注数据画出相关文档和不相关文档的相似度分布找两条分布曲线的交叉点附近作为阈值。实际用的时候可以稍微放宽一点避免漏掉边缘相关的文档。上下文压缩是提升效果的重要手段。召回的长文档里往往只有一两句话是真正相关的把整篇文档塞给模型既浪费token又引入噪声。压缩的做法有两种一种是抽取式用一个小模型或者规则从文档里挑出最相关的句子另一种是生成式让模型对文档做摘要只保留跟问题相关的信息。抽取式速度快、成本低适合对延迟敏感的场景生成式效果好但会增加一次模型调用适合对质量要求高的场景。提示词模板的设计直接决定模型能不能用好检索到的上下文。我见过很多模板就是把文档一股脑塞进去然后问模型问题这样模型很容易被无关内容带偏。好的模板会明确告诉模型下面给你一些参考资料请基于这些资料回答问题如果资料里没有相关信息就明确说不知道不要自己编。同时会给模型一些格式上的约束比如要求分点回答、要求引用来源、要求控制回答长度。这些约束看起来是小事但对输出稳定性的影响很大。3.3 模型调用层的稳定性设计模型调用看起来就是发个HTTP请求但在生产环境里要考虑的事情很多。超时和重试是最基础的。模型推理的延迟波动很大同样的请求可能这次200毫秒返回下次要3秒。超时时间设置太短会导致大量请求失败太长会让用户等太久。我的经验是设置一个分级超时比如首token超时设5秒整体超时设30秒超过就触发重试或者降级。降级策略是保证可用性的关键。当主模型不可用或者响应太慢的时候要能自动切换到备用模型。备用模型可以是更小更快的版本也可以是不同厂商的同类模型。降级的时候要注意提示词模板可能需要微调因为不同模型对指令的遵循程度不一样。我通常会给每个模型维护一套适配过的模板切换的时候连带模板一起切。流式输出对用户体验的影响很大。用户不需要等整个回答生成完才能看到内容首token返回之后就可以开始展示。实现流式输出要注意几个细节一是要处理好流式返回的拼接和解析特别是当输出需要做结构化提取的时候二是要设置合理的缓冲策略避免网络抖动导致输出断断续续三是要考虑流式中断的情况用户可能看到一半就不想看了这时候要及时取消后端的生成请求释放资源。并发控制是另一个容易被忽略的点。模型API通常有速率限制并发太高会被限流。我的做法是在调用层做一个令牌桶或者漏桶的限流器控制单位时间内的请求数。同时要做好请求排队超过并发上限的请求进入队列等待而不是直接拒绝。队列的长度和等待超时也要设置合理避免用户等太久。3.4 输出解析与结构化提取的实操方案模型返回的原始文本往往不能直接使用需要做解析和结构化。最简单的场景是要求模型返回JSON但实际用下来你会发现模型经常会在JSON外面包一层解释性文字或者JSON格式有细微错误。我的处理流程是先尝试直接解析失败的话用正则提取JSON部分再解析再失败的话用一个小模型做格式修复最后还不行就返回兜底结果并记录日志。对于需要提取特定字段的场景我倾向于用函数调用或者结构化输出的能力而不是靠提示词约束。现在主流模型基本都支持指定输出格式让模型按照给定的schema返回解析成功率比纯提示词约束高很多。如果模型不支持那就退回到提示词约束加后处理校验的方案。输出校验这步不能省。即使模型返回了格式正确的内容也要做业务逻辑上的校验。比如要求返回日期就要检查是不是合法日期要求返回枚举值就要检查是不是在允许范围内要求返回数值就要检查是不是在合理区间。校验不通过的输出要么触发重试要么走降级逻辑绝对不能直接透传给用户。提示建议在开发阶段把每次模型调用的原始输入输出都完整记录下来包括提示词、模型返回、解析结果、校验结果。这些数据在排查问题和做效果分析的时候非常有用等到线上出问题再想加日志就来不及了。4. 完整实操流程与关键环节实现4.1 环境搭建与依赖管理从零搭建的第一步是把环境弄干净。我强烈建议用虚拟环境或者容器来隔离依赖AI工程涉及的东西太多不同项目之间的依赖冲突很常见。Python环境下用venv或者conda都行关键是每个项目独立。依赖管理用requirements.txt或者poetry把版本号锁死避免因为某个库自动升级导致行为变化。核心依赖大概分这么几类模型调用客户端比如openai的官方库或者兼容接口的第三方库向量计算库比如numpy、faiss-cpu或者faiss-gpu文本处理库比如用于分词的jieba或者用于解析的beautifulsoupWeb框架比如fastapi或者flask用于对外提供服务监控和日志库比如prometheus客户端和结构化日志库。这些库的版本要仔细选特别是faiss和numpy之间有版本兼容性要求装错了会直接报错。配置管理我习惯用环境变量加配置文件的方式。敏感信息比如API密钥只放环境变量不写进代码和配置文件。业务配置比如模型名称、超时时间、检索参数放配置文件方便不同环境切换。配置文件用YAML或者TOML格式比JSON可读性好支持注释。4.2 索引构建的完整代码实现索引构建的流程可以拆成四步加载文档、切分文本、生成嵌入、写入索引。加载文档这步要根据数据源来定如果是本地文件就用文件读取如果是数据库就写查询如果是网页就做抓取和解析。我通常会把加载逻辑抽象成一个接口不同数据源实现不同的加载器这样扩展起来方便。文本切分我一般用递归切分的方式先按大结构切比如章节再按小结构切比如段落最后按长度切。切分的时候维护一个元数据字典记录每个片段的来源、位置、类型等信息。切分完的片段列表要做一个去重完全相同的片段只保留一份避免检索的时候重复召回。嵌入生成这步要注意批处理。一条一条调嵌入接口太慢通常一次批处理几十到几百条。批大小要根据接口的限制和内存情况来定太大容易超时或者OOM太小效率低。生成嵌入的时候要做好错误处理和重试网络抖动导致的失败要能自动恢复。嵌入向量建议做归一化这样后面算余弦相似度的时候可以直接用点积速度快一些。写入索引的时候如果用的是FAISS要注意索引类型的选择。小规模数据用Flat索引就行精度最高。数据量大了考虑IVF或者HNSW用一定的精度损失换速度。索引文件要定期持久化避免进程重启后需要重新构建。如果数据会更新还要考虑增量索引的方案全量重建的成本太高。import numpy as np import faiss from typing import List, Dict class VectorIndex: def __init__(self, dim: int): self.dim dim self.index faiss.IndexFlatIP(dim) # 内积索引配合归一化向量使用 self.metadata: List[Dict] [] def add(self, vectors: np.ndarray, metas: List[Dict]): # 向量归一化 faiss.normalize_L2(vectors) self.index.add(vectors) self.metadata.extend(metas) def search(self, query_vector: np.ndarray, top_k: int 10): faiss.normalize_L2(query_vector) scores, indices self.index.search(query_vector, top_k) results [] for score, idx in zip(scores[0], indices[0]): if idx -1: continue results.append({ score: float(score), metadata: self.metadata[idx] }) return results上面这段代码是一个最简版的向量索引封装实际项目里还要加上持久化、增量更新、多索引管理等逻辑。但核心思路就是这样归一化向量、建索引、查询时归一化查询向量、返回相似度和元数据。4.3 检索与重排序的串联实现检索环节我通常做成两阶段粗排召回和精排重排序。粗排用向量索引加关键词索引并行召回各取Top-50然后合并去重。合并的时候用RRF算法它对不同来源的分数尺度不敏感比较适合这种多路召回的场景。RRF的核心思想是根据每个文档在各路召回中的排名来计算综合得分排名越靠前得分越高公式是 score sum(1 / (k rank))k一般取60。精排阶段用重排序模型对粗排结果重新打分。重排序模型通常是交叉编码器把查询和文档拼在一起输入模型输出相关性分数。这种方式的精度比向量相似度高很多但计算量大所以只适合对少量候选做精排。重排序之后取Top-5或者Top-8作为最终上下文。上下文组装的时候要注意顺序和格式。我的经验是把最相关的放最前面和最后面中间放次相关的因为模型对开头和结尾的内容注意力更集中。每个片段前面加上来源标记方便模型引用也方便后续做溯源。片段之间用分隔符隔开避免模型把不同片段的内容混在一起。def rrf_fusion(result_lists: List[List[Dict]], k: int 60) - List[Dict]: RRF融合多路召回结果 score_map {} doc_map {} for results in result_lists: for rank, item in enumerate(results): doc_id item[metadata][doc_id] if doc_id not in score_map: score_map[doc_id] 0.0 doc_map[doc_id] item score_map[doc_id] 1.0 / (k rank 1) fused [] for doc_id, score in sorted(score_map.items(), keylambda x: -x[1]): item doc_map[doc_id].copy() item[rrf_score] score fused.append(item) return fused4.4 提示组装与模型调用的完整链路提示组装这步看起来简单但细节很多。系统指令要写清楚角色定位、任务目标、输出格式要求、约束条件。用户问题要原样保留不要做过度改写避免丢失信息。检索到的上下文要标注来源并且明确告诉模型这些是参考资料不是绝对正确的。历史对话要控制长度太长的历史做摘要或者只保留最近几轮。模型调用我封装成一个统一的客户端对外暴露一个generate方法内部处理重试、降级、流式、日志等逻辑。调用的时候传入提示词、模型名称、温度、最大token数等参数。温度这个参数要根据任务类型来定事实性问答用低温度比如0.1到0.3创意生成用高温度比如0.7到0.9。最大token数要设置合理太小会导致回答被截断太大浪费资源。流式输出的处理要特别注意。如果最终输出需要做结构化解析流式返回的片段要缓存起来等全部返回完再解析。如果只是展示给用户看可以边收边展示但要处理好markdown格式的渲染避免因为片段不完整导致格式错乱。class LLMClient: def __init__(self, primary_model: str, fallback_model: str, timeout: int 30): self.primary_model primary_model self.fallback_model fallback_model self.timeout timeout def generate(self, prompt: str, temperature: float 0.3, max_tokens: int 1024): try: return self._call(self.primary_model, prompt, temperature, max_tokens) except Exception as e: # 记录主模型失败日志 print(fPrimary model failed: {e}, falling back) return self._call(self.fallback_model, prompt, temperature, max_tokens) def _call(self, model: str, prompt: str, temperature: float, max_tokens: int): # 实际调用逻辑包含重试 for attempt in range(3): try: # 这里替换成实际的模型调用 response call_model_api(model, prompt, temperature, max_tokens, self.timeout) return response except TimeoutError: if attempt 2: raise continue4.5 评估体系的搭建与指标定义没有评估就没有优化。AI工程的评估分两个层面组件级评估和端到端评估。组件级评估针对检索、重排序、解析这些单独模块端到端评估看最终回答的质量。检索评估的核心指标是召回率和精确率。准备一批查询人工标注每个查询对应的相关文档然后看检索结果里有多少相关文档被召回了召回的里面有多少是真正相关的。重排序评估看的是排序质量用NDCG或者MRR这类指标。解析评估看的是结构化提取的准确率和格式合规率。端到端评估更复杂一些因为回答质量很难用单一指标衡量。我通常从几个维度来评事实准确性回答里的信息是否跟知识库一致完整性是否覆盖了问题的所有方面相关性是否回答了用户真正想问的格式合规性是否符合要求的输出格式。每个维度可以人工打分也可以用模型来打分但模型打分需要先做校准。评估数据集的构建是个持续的过程。初期可以手工构造一批典型查询覆盖主要场景。上线之后从真实用户请求里采样把有问题的case补充进评估集。评估集要定期更新避免过拟合到旧数据上。5. 常见问题与排查技巧实录5.1 检索效果差的排查路径检索效果差是最常见的问题排查的时候按这个顺序来。先看查询本身用户的问题是不是太短或者太模糊如果是的话考虑做查询改写或者扩展。再看嵌入模型是不是模型跟你的领域不匹配通用模型在垂直领域的效果可能很差考虑换模型或者做微调。然后看切分粒度片段是不是太大或者太小太大导致噪声多太小导致信息不完整。接着看索引类型Flat索引精度最高如果用了IVF或者HNSW调一下nprobe或者efSearch参数。最后看相似度阈值是不是设得太高导致相关文档被过滤掉了。我遇到过一个典型案例用户问的是产品的一个具体功能怎么用但检索出来的全是产品介绍和营销文案。排查下来发现是切分粒度的问题产品文档被按固定长度切了功能介绍和操作步骤被切到了不同片段里检索的时候只召回了介绍部分。改成按章节切分之后问题就解决了。5.2 模型输出不稳定的应对策略模型输出不稳定表现为同样的问题有时候回答得好有时候答非所问有时候格式不对。这个问题要从几个方面入手。温度参数先调低事实性任务温度设0.1到0.2基本能消除大部分随机性。提示词要写得更明确把要求一条一条列清楚避免模糊表述。输出格式用结构化输出能力来约束比纯提示词约束可靠得多。上下文要控制好无关内容太多会干扰模型该压缩就压缩该过滤就过滤。还有一个容易被忽略的点是模型版本。同一个模型的不同版本行为可能差异很大升级模型版本之后一定要重新跑评估集确认效果没有退化。我习惯在配置里把模型版本号写死避免自动升级带来的意外。5.3 性能瓶颈的定位与优化性能问题通常出在几个地方。嵌入计算如果是实时做的会显著增加延迟能预计算的就预计算。向量检索在数据量大或者索引类型不合适的时候会变慢考虑换索引类型或者加缓存。模型调用是最大的延迟来源能流式就流式能并行就并行能用小模型就用小模型。网络传输在跨区域调用的时候影响很大尽量让服务离模型端点近一些。优化的时候要有数据支撑不能凭感觉。在每个环节打点记录耗时找出真正的瓶颈再动手。我见过团队花大力气优化检索结果发现90%的时间花在模型调用上优化方向完全错了。问题现象可能原因排查方法解决方向检索召回不全切分粒度过大、相似度阈值过高检查召回Top-K的相似度分布调整切分策略、降低阈值回答答非所问提示词不明确、上下文噪声多检查提示词模板和上下文内容优化模板、增加压缩过滤输出格式错误缺少格式约束、模型能力不足统计格式错误率用结构化输出、换模型延迟过高模型调用慢、检索慢分环节打点流式、缓存、换小模型成本超预算token用量大、调用次数多统计token消耗压缩上下文、缓存结果5.4 成本控制的实操经验AI应用的成本主要是token消耗。控制成本的核心是减少不必要的token。上下文压缩是最直接的手段把长文档压缩成关键句能省很多token。缓存是另一个手段相同或者相似的问题直接返回缓存结果不用重复调用模型。模型分级也很重要简单任务用小模型复杂任务才用大模型。我通常会做一个token预算管理给每个请求设置token上限超过就触发压缩或者截断。同时监控每天的token消耗设置告警阈值避免因为异常流量导致成本失控。缓存要注意失效策略知识库更新之后相关缓存要及时清除避免返回过时信息。注意缓存相似度匹配的时候要小心太宽松会导致返回不相关的缓存结果太严格又起不到缓存效果。建议用向量相似度做缓存匹配阈值设高一些比如0.95以上只缓存几乎完全相同的问题。6. 迭代方向与个人实操体会这套从零搭建的AI工程体系我在几个项目里跑下来最大的体会是前期慢就是快。花时间把每一层的边界划清楚把评估体系建起来后面迭代的时候效率会高很多。相反如果一开始图快直接上框架后面每改一个地方都要跟框架的抽象做斗争反而更慢。后续的迭代方向我觉得有几个值得深入。一是检索策略的自动化调优现在参数还是靠人工试可以做成基于评估集的自动搜索。二是提示词的版本管理把提示词当成代码来管理每次修改都跑评估避免改坏。三是多模型路由根据查询类型自动选择最合适的模型在效果和成本之间找平衡。四是在线学习把用户的反馈信号收集起来持续优化检索和生成策略。最后分享一个小技巧。我在每个项目里都会维护一个问题案例库把线上遇到的bad case记录下来标注问题类型和修复方案。这个库既是排查问题的参考也是评估集的来源还是团队知识沉淀的载体。坚持记下来几个月之后你会发现这是项目里最有价值的资产之一。